bus

type BusOption func(*BusOpts)

type BusOpts struct {
	Compression CompressionOpts
}

type Channel struct {
	// Flat '|'-delimited key, for brokers with no subject hierarchy.
	Legacy string
	// Hierarchical subject, e.g. SRV.<service>.<topic>.
	Server string
	// Demultiplexes several logical channels sharing one Server subject. Carried
	// in the payload, not alongside it.
	Local string
}

type CompressionOpts struct {
	// Zero disables compression; otherwise the gzip level.
	Quality int
	// Payload bytes below which compression is skipped.
	Threshold int
	// Caps an inbound payload after decompression, bounding amplification from a
	// hostile publisher. Zero is unlimited.
	MaxDecompressedSize int
}

type EmptySubscription[MessageType any] struct{}

type MessageBus interface {
	Publish(ctx context.Context, channel Channel, msg proto.Message) error
	Subscribe(ctx context.Context, channel Channel, channelSize int) (Reader, error)
	SubscribeQueue(ctx context.Context, channel Channel, channelSize int) (Reader, error)
	// Caps an inbound payload after decompression. Zero is unlimited.
	MaxDecompressedSize() int
}

type PublishHandler func(ctx context.Context, channel Channel, msg proto.Message) error

type PublishInterceptor func(next PublishHandler) PublishHandler

type ReadHandler func() ([]byte, bool)

type Reader interface {
	// Reports false once the reader is closed.
	Read() ([]byte, bool)
	Close() error
}

type SubscribeInterceptor func(ctx context.Context, channel Channel, next ReadHandler) ReadHandler

type Subscription[MessageType proto.Message] interface {
	Channel() <-chan MessageType
	Close() error
}

type TestBusOption func(*TestBusOpts)

type TestBusOpts struct {
	PublishInterceptors   []PublishInterceptor
	SubscribeInterceptors []SubscribeInterceptor
}

type Transport interface {
	Publish(ctx context.Context, channel Channel, payload []byte) error
	Subscribe(ctx context.Context, channel Channel, channelSize int) (Reader, error)
	SubscribeQueue(ctx context.Context, channel Channel, channelSize int) (Reader, error)
}