topic

README

topic

Efficient in-process publish-subscribe broker for Go in which types are the topics

Publishers send ordinary values, and subscribers choose the types they want to receive. No names, keys, identifiers, schemas, or registries: a Go type is the topic.

import "github.com/ardnew/topic"

type Tick struct{ N int }

var b topic.Broker            // zero value is ready to use

ch, cancel := b.Subscribe[Tick]()
defer cancel()

b.Publish(Tick{N: 1})         // topic is inferred from the value
fmt.Println((<-ch).N)         // 1

Internally, subscriber routing is highly-optimized and features no reflection, serialization, code generation, or dependencies.

Publishing a value is allocation-free in nearly all cases.

[!IMPORTANT] This module uses generic methods, which sets the minimum supported toolchain version at Go 1.27.

All source code is pure Go and standard library-only.

API

type Broker struct{ ... }

func (b *Broker) Publish[T any](v T)
func (b *Broker) Subscribe[T any](opts ...Option[T]) (<-chan T, func())

type Option[T any] struct{ ... }

func Buffer[T any](n int) Option[T]
func From[Pub, Sub any](f func(Sub) (Pub, bool)) Option[Pub]

Wildcard semantics are achieved using Go interfaces. Subscribing to an interface type receives every published value that implements it; publishers do not need to name the interface. Subscribing to any receives every value published with that broker.

errs, cancel := b.Subscribe(topic.Buffer[error](16))

Stricter than wildcards, From subscribes to an additional source type and converts it, statically. The same function can also be used to filter which values are delivered.

ch, cancel := b.Subscribe(
    topic.From(func(c Celsius) (Fahrenheit, bool) { return Fahrenheit(c*9/5 + 32), true }),
    topic.From(func(f Fahrenheit) (Fahrenheit, bool) { return f, f > 0 }),
)

Delivery semantics

Delivery is best effort and lossy, by design. Each subscription has its own channel and capacity; if it cannot accept a value immediately, that value is dropped for that subscription and publication continues for everyone else. A slow or abandoned subscriber never blocks a publisher. Nothing is queued, retried, or persisted.

Cancelling closes the subscription's channel, and after it returns nothing is ever sent on that channel again.

Performance

Publication is lock-free with respect to the broker.

Publication allocates nothing on a broker whose subscriptions all name concrete types.

Measured with go test -run XXX -bench . -benchtime 500000x on an AMD Ryzen Threadripper 1950X, Go 1.27rc2:

Path	Time	Allocations
no subscriber	4.8 ns	0
unmatched topic	6.4 ns	0
routing only (saturated subscriber)	32 ns	0
direct delivery, identical type	57 ns	0
pointer to interface subscription	71 ns	0
interface value to interface subscription	116 ns	0
directly matched transformation	79 ns	0
directly matched filter (rejecting)	21 ns	0
small non-pointer value to an any subscription	62 ns	0
non-pointer value to an any subscription	140 ns	1
saturated publication from 32 goroutines	47 ns	0

The single allocation is the one documented exception: A subscription to an interface type will receive the value in an interface. Boxing a value in an interface costs one allocation per publication, unless the value is pointer-shaped, already held in an interface, zero-sized, or a pointer-free value of 1/2/4/8 bytes with uint8-representable bit pattern (e.g., float32(0), true, 'x' are free; -1, float32(1), [3]byte{} are not). This is an optimization of the Go runtime and not this module.

In either case, the interface conversion is always done once per publication regardless of how many subscribers match.

Validation

gofmt -l .
go vet ./...
go test ./...
go test -race -count=2 ./...
go test -run XXX -bench . ./...

Makefile

Common targets, all writing to dist/:

make test          # run the suite
make race          # run it under the race detector
make cover report  # coverage profile, then its HTML
make bench         # run the benchmarks
make flame         # CPU flame graph in the browser
make stat          # compare benchmarks against a saved baseline
make debug-test    # open the suite in dlv
make clean

Any target can be narrowed with RUN, BENCH, PKG, COUNT, or TIME:

make race RUN=TestDirectDelivery COUNT=5
make flame BENCH=BenchmarkPublishParallel

Credits

[!NOTE] OpenAI Codex and Claude Opus contributed to this project as adversarial AI agents.

Documentation

Overview

Package topic provides an in-process publish-subscribe broker in which Go types are the topics.

A publisher offers an ordinary Go value; a subscriber selects a Go type and receives values of that type on a channel. Nothing else identifies a topic: there are no names, keys, identifiers, schemas, or registries to declare or maintain, and the package uses no reflection, serialization, or code generation.

Basic use

The zero Broker is ready to use.

type Tick struct{ N int }

var b topic.Broker

ch, cancel := b.Subscribe[Tick]()
defer cancel()

b.Publish(Tick{N: 1})
fmt.Println((<-ch).N) // 1

The topic of a publication is the compile-time type of the published value, and the topic of a subscription is its result type. Publishing through an interface variable therefore publishes on that interface type, and a subscription naming a concrete type does not see it.

Matching

A subscription accepts a publication when one of its source types matches it. The source types are, in order, those declared with From followed by the subscription's own type. A source type S matches a publication of type T when S and T are identical, when S is an interface type that the published value implements, or when S is any.

An interface source is matched against the published value itself, so it sees the value's dynamic type. Publishing through an interface variable can therefore reach a subscription to a different interface that the underlying value happens to implement.

The first matching source decides the outcome; no later source is consulted. A source declared with From delivers the value it returns when it reports true and drops the value when it reports false, which is how filtering is expressed. Because the subscription's own type is considered last, direct matching always remains available.

A nil value of an interface type carries no dynamic type, so it reaches only subscriptions whose type is identical to the published interface type, and subscriptions to any.

Delivery and loss

Delivery is a non-blocking send performed while Publish runs. Each subscription has its own channel and its own capacity, set with Buffer. If a subscription's buffer is full, or it is unbuffered and no receiver is waiting, the value is dropped for that subscription and publication continues for the others. A slow or abandoned subscriber therefore never blocks a publisher or another subscriber.

Delivery is best effort and confined to one process. Nothing is queued, retried, persisted, or sent over a network, and no delivery is guaranteed. Values one subscription accepts from sequential, non-overlapping publications are received in publication order; ordering between concurrent publications is unspecified.

Lifecycle

The function returned by Broker.Subscribe unregisters the subscription and closes its channel, which ends a range loop over it. Values already accepted remain receivable. It is idempotent, and after it returns nothing is ever sent on that channel again.

Concurrency

A Broker is safe for concurrent publication, subscription, cancellation, and delivery. A Broker must not be copied after first use.

Allocation

Publication allocates nothing, whatever the value's type and however many subscriptions match, on a broker whose subscriptions all name concrete types. A subscription to an interface type or to any needs the value in an interface, which costs one allocation per publication unless the value is pointer-shaped, already held in an interface, zero-sized, or a pointer-free value of one, two, four, or eight bytes whose bits are below 256, which the runtime boxes from read-only storage. That conversion is done once and shared by every such subscription, so the cost never depends on how many subscribers match, but it does apply to every publication on that broker.

Example

A subscriber selects a Go type; publishers send ordinary values of it.

package main

import (
	"fmt"

	"github.com/ardnew/topic"
)

func main() {
	type Tick struct{ N int }

	var b topic.Broker

	ch, cancel := b.Subscribe[Tick]()
	defer cancel()

	b.Publish(Tick{N: 1})

	fmt.Println((<-ch).N)
}

Output:
1

Index

• type Broker
  • func (b *Broker) Publish[T any](v T)
  • func (b *Broker) Subscribe[T any](opts ...Option[T]) (<-chan T, func())
• type Option
  • func Buffer[T any](n int) Option[T]
  • func From[Pub, Sub any](f func(Sub) (Pub, bool)) Option[Pub]

Examples

• Package
• Broker.Publish
• Broker.Subscribe (Any)
• Broker.Subscribe (Cancel)
• Broker.Subscribe (Interface)
• Buffer
• From
• From (Filter)
• From (Multiple)

Types

type Broker

type Broker struct {
	// contains filtered or unexported fields
}

A Broker routes published values to subscriptions by Go type.

The zero Broker is ready to use. Brokers are independent: a value published to one is never observed through another. A Broker must not be copied after first use.

func (*Broker) Publish

func (b *Broker) Publish[T any](v T)

Publish offers v to every subscription that matches it.

The topic is T, the compile-time type of v. Publishing a value no subscription matches is valid and has no effect.

Delivery to each matching subscription is a non-blocking send: if the subscription cannot accept the value immediately, the value is dropped for that subscription only. Publish never blocks on a subscriber and never returns before delivery has been attempted for every subscription registered when it began.

Example

Distinct types are distinct topics, with no identifiers to declare.

package main

import (
	"fmt"

	"github.com/ardnew/topic"
)

func main() {
	type Login struct{ User string }
	type Logout struct{ User string }

	var b topic.Broker

	logins, cancelLogins := b.Subscribe[Login]()
	defer cancelLogins()
	logouts, cancelLogouts := b.Subscribe[Logout]()
	defer cancelLogouts()

	b.Publish(Logout{User: "ada"})
	b.Publish(Login{User: "grace"})

	fmt.Println((<-logins).User, "in")
	fmt.Println((<-logouts).User, "out")
}

Output:
grace in
ada out

func (*Broker) Subscribe

func (b *Broker) Subscribe[T any](opts ...Option[T]) (<-chan T, func())

Subscribe registers a subscription for values of type T and returns its channel and a cancellation function.

By default the subscription accepts publications whose type is T, or whose value implements T when T is an interface type. From adds source types it also accepts, and Buffer sets its capacity; without it the capacity is one.

Cancelling unregisters the subscription and closes the channel, which ends a range loop over it. Cancelling is idempotent, and after it returns nothing is ever sent on the channel again. A subscription that is never cancelled is retained by the broker, so callers should cancel when they stop receiving.

Example (Any)

A subscription to any receives every published value.

package main

import (
	"errors"
	"fmt"

	"github.com/ardnew/topic"
)

func main() {
	var b topic.Broker

	all, cancel := b.Subscribe(topic.Buffer[any](4))
	defer cancel()

	b.Publish("hello")
	b.Publish(7)
	b.Publish(errors.New("oops"))

	for len(all) > 0 {
		v := <-all
		fmt.Printf("%T %v\n", v, v)
	}
}

Output:
string hello
int 7
*errors.errorString oops

Example (Cancel)

Cancelling closes the channel, so a range loop over it ends.

package main

import (
	"fmt"

	"github.com/ardnew/topic"
)

func main() {
	type Tick struct{ N int }

	var b topic.Broker

	ch, cancel := b.Subscribe(topic.Buffer[Tick](4))

	for i := range 3 {
		b.Publish(Tick{N: i})
	}
	cancel()
	b.Publish(Tick{N: 99}) // never delivered: the subscription is gone

	for tick := range ch { // values accepted before cancel remain receivable
		fmt.Println(tick.N)
	}
	fmt.Println("done")
}

Output:
0
1
2
done

Example (Interface)

A subscription to an interface type receives every published value that implements it.

package main

import (
	"fmt"

	"github.com/ardnew/topic"
)

func main() {
	var b topic.Broker

	errs, cancel := b.Subscribe(topic.Buffer[error](4))
	defer cancel()

	b.Publish(fmt.Errorf("disk full")) // implements error
	b.Publish(42)                      // does not

	fmt.Println(<-errs)
	fmt.Println("pending:", len(errs))
}

Output:
disk full
pending: 0

type Option

type Option[T any] struct {
	// contains filtered or unexported fields
}

An Option configures a subscription for values of type T.

The zero Option has no effect. See Buffer and From.

func Buffer

func Buffer[T any](n int) Option[T]

Buffer sets the capacity of a subscription's channel.

A capacity of zero makes the subscription unbuffered, so a value reaches it only if a receiver is already waiting; anything else is dropped. A negative capacity is treated as zero. Without this option the capacity is one, the smallest buffer that lets a publication survive a consumer that is not currently blocked in a receive. Given more than once, the last one wins.

The type argument cannot be inferred from n, so it is written explicitly:

ch, cancel := b.Subscribe(topic.Buffer[Tick](64))

Example

Buffer sets how much a subscription can hold. A publication that a subscription cannot accept immediately is dropped for that subscription alone.

package main

import (
	"fmt"

	"github.com/ardnew/topic"
)

func main() {
	type Tick struct{ N int }

	var b topic.Broker

	small, cancelSmall := b.Subscribe(topic.Buffer[Tick](2))
	defer cancelSmall()
	large, cancelLarge := b.Subscribe(topic.Buffer[Tick](8))
	defer cancelLarge()

	for i := range 5 {
		b.Publish(Tick{N: i})
	}

	fmt.Println("small held:", len(small))
	fmt.Println("large held:", len(large))
}

Output:
small held: 2
large held: 5

func From

func From[Pub, Sub any](f func(Sub) (Pub, bool)) Option[Pub]

From declares that a subscription to type Sub also accepts publications of type Pub, converting each one with f.

f returns the value to deliver and whether to deliver it: returning false drops that value for this subscription, so the same option expresses filtering, transformation, or both. Both type parameters are inferred from f.

// transform
b.Subscribe(topic.From(func(c Celsius) (Fahrenheit, bool) {
	return Fahrenheit(c*9/5 + 32), true
}))

// filter
b.Subscribe(topic.From(func(t Tick) (Tick, bool) {
	return t, t.N%2 == 0
}))

The option may be given more than once to declare several source types. A subscription considers its sources in the order given, followed by its own type, and the first source that matches a publication decides its outcome; no later source sees it. Declaring the same source type twice is allowed and the later declaration is unreachable.

f runs on the goroutine that published the value, before the value is offered to the subscription's channel, so it should be cheap and must not block.

Example

From converts an explicitly chosen source type into the subscription's type.

package main

import (
	"fmt"

	"github.com/ardnew/topic"
)

func main() {
	type Celsius float64
	type Fahrenheit float64

	var b topic.Broker

	ch, cancel := b.Subscribe(
		topic.Buffer[Fahrenheit](4),
		topic.From(func(c Celsius) (Fahrenheit, bool) { return Fahrenheit(c*9/5 + 32), true }),
	)
	defer cancel()

	b.Publish(Celsius(100))    // converted
	b.Publish(Fahrenheit(-40)) // direct match still works
	b.Publish("not a reading") // unrelated type

	fmt.Println(<-ch)
	fmt.Println(<-ch)
}

Output:
212
-40

Example (Filter)

Reporting false from the same function drops the value, which is filtering.

package main

import (
	"fmt"

	"github.com/ardnew/topic"
)

func main() {
	type Reading struct{ Volts float64 }

	var b topic.Broker

	ch, cancel := b.Subscribe(
		topic.Buffer[Reading](8),
		topic.From(func(r Reading) (Reading, bool) { return r, r.Volts > 3.0 }),
	)
	defer cancel()

	for _, v := range []float64{1.5, 3.3, 2.0, 5.0} {
		b.Publish(Reading{Volts: v})
	}

	for len(ch) > 0 {
		fmt.Println((<-ch).Volts)
	}
}

Output:
3.3
5

Example (Multiple)

Several source types can feed one subscription. The first source that matches a publication decides its fate, so ordering is deterministic.

package main

import (
	"fmt"

	"github.com/ardnew/topic"
)

func main() {
	type Bytes int64
	type Packet struct{ Size int }
	type Frame struct{ Size int }

	var b topic.Broker

	ch, cancel := b.Subscribe(
		topic.Buffer[Bytes](8),
		topic.From(func(p Packet) (Bytes, bool) { return Bytes(p.Size), true }),
		topic.From(func(f Frame) (Bytes, bool) { return Bytes(f.Size), f.Size > 0 }),
	)
	defer cancel()

	b.Publish(Packet{Size: 1500})
	b.Publish(Frame{Size: 0}) // rejected by its source
	b.Publish(Frame{Size: 64})
	b.Publish(Bytes(9))

	for len(ch) > 0 {
		fmt.Println(<-ch)
	}
}

Output:
1500
64
9