libbus

type Config ¶

type Config struct {
	NATSURL      string
	NATSPassword string
	NATSUser     string
}

type Handler ¶

type Handler func(ctx context.Context, data []byte) ([]byte, error)

Handler is a function that processes a request and returns a response. It is used by the Serve method to handle incoming requests.

type InMem ¶

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

InMem is an in-memory Messenger for single-process use: no NATS, no network. It intentionally reproduces the NATS backend's observable contract (at-most-once delivery, Request failing immediately with no handler registered) — see the Messenger interface docs for the full matrix.

func NewInMem() *InMem

func (p *InMem) Close() error

func (p *InMem) Publish(ctx context.Context, subject string, data []byte) error

func (p *InMem) Request(ctx context.Context, subject string, data []byte) ([]byte, error)

func (p *InMem) Serve(ctx context.Context, subject string, handler Handler) (Subscription, error)

func (p *InMem) Stream(ctx context.Context, subject string, ch chan<- []byte) (Subscription, error)

type Messenger ¶

type Messenger interface {
	// Publish sends a fire-and-forget message to a given subject.
	Publish(ctx context.Context, subject string, data []byte) error

	// Stream creates a subscription to a subject and delivers messages asynchronously
	// to the provided channel. The subscription is automatically managed and will
	// be closed when the provided context is canceled.
	Stream(ctx context.Context, subject string, ch chan<- []byte) (Subscription, error)

	// Request sends a request message and waits for a reply. The context can be
	// used to set a timeout or to cancel the request.
	Request(ctx context.Context, subject string, data []byte) ([]byte, error)

	// Serve registers a handler for a given subject to respond to requests.
	// It starts a worker that listens for requests and executes the handler.
	// The returned Subscription can be used to stop serving.
	Serve(ctx context.Context, subject string, handler Handler) (Subscription, error)

	// Close disconnects from the messaging server and cleans up any underlying resources.
	Close() error
}

Messenger is a high-level pub-sub/request-reply interface for real-time notifications and distributing lightweight messages between services.

Guaranteed by every backend (enforced by conformance_test.go): Publish to a subject with no subscribers is a no-op and never blocks; Stream delivers in publish order until Unsubscribe or context cancel; a handler error still yields a reply — Request returns a non-nil error only on transport failure, never on handler failure; and after Close, every method returns ErrConnectionClosed.

Backends differ in ways callers must tolerate: NATS/InMem are at-most-once under backpressure (drop once a subscriber's buffer fills) while SQLiteBus is durable; NATS/InMem require Serve to return before Request is called, SQLiteBus does not; handler concurrency and delivery latency (SQLiteBus is poll-driven) vary per backend. Always give Request a deadline.

func NewPubSub(ctx context.Context, cfg *Config) (Messenger, error)

func NewTestPubSub() (Messenger, func(), error)

type SQLiteBus ¶ added in v0.4.0

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

SQLiteBus implements Messenger over a SQLite database.

Schema tables (bus_events, bus_requests, bus_replies) must exist before use. They are part of runtimetypes.SchemaSQLite and are created automatically when the CLI database is opened.

Usage:

bus := libbus.NewSQLite(dbManager.WithoutTransaction())
defer bus.Close()

func NewSQLite(exec sqlExec) *SQLiteBus

func NewSQLiteWithOptions(exec sqlExec, opt SQLiteBusOptions) *SQLiteBus

func (b *SQLiteBus) Close() error

func (b *SQLiteBus) Publish(ctx context.Context, subject string, data []byte) error

func (b *SQLiteBus) Request(ctx context.Context, subject string, data []byte) ([]byte, error)

func (b *SQLiteBus) Serve(ctx context.Context, subject string, handler Handler) (Subscription, error)

func (b *SQLiteBus) Stream(ctx context.Context, subject string, ch chan<- []byte) (Subscription, error)

type SQLiteBusOptions ¶ added in v0.6.8

type SQLiteBusOptions struct {
	EventPoll   time.Duration
	RequestPoll time.Duration
}

SQLiteBusOptions overrides poll intervals (e.g. tests use 1ms so request/reply is deterministic).

type Subscription ¶

type Subscription interface {
	// Unsubscribe removes the subscription, stopping the delivery of messages.
	Unsubscribe() error
}

Subscription represents an active subscription to a subject.