client

Documentation

Overview

Package client is the Go client library for gombus brokers. It talks the gosocketio event protocol defined in internal/protocol over the data ("/") and control ("/control") namespaces.

Index

• func TrimNamespace(nsp string) string
• type Conn
  • func Dial(ctx context.Context, rawURL string, opts Options) (*Conn, error)
  • func (c *Conn) AddConsumer(req protocol.AddConsumerReq) error
  • func (c *Conn) Close() error
  • func (c *Conn) Connected() bool
  • func (c *Conn) CreateStream(req protocol.CreateStreamReq) (*protocol.StreamMeta, error)
  • func (c *Conn) DeleteConsumer(stream, name string) error
  • func (c *Conn) DeleteStream(name string) error
  • func (c *Conn) Emit(nsp, event string, req any, out any) error
  • func (c *Conn) Publish(subject string, data []byte, header map[string]string) error
  • func (c *Conn) PurgeStream(name string) error
  • func (c *Conn) Request(subject string, data []byte, timeout time.Duration) (protocol.ReplyMsg, error)
  • func (c *Conn) Stats() (protocol.StatsSnapshot, error)
  • func (c *Conn) StreamAck(consumer string, seq uint64) error
  • func (c *Conn) StreamInfo(name string) (protocol.StreamInfoAck, error)
  • func (c *Conn) StreamNak(consumer string, seq uint64, delay time.Duration) error
  • func (c *Conn) StreamPublish(stream, subject string, data []byte, msgID string, header map[string]string) (protocol.StreamPublishAck, error)
  • func (c *Conn) StreamPull(consumer string, batch int, wait time.Duration) ([]protocol.Message, error)
  • func (c *Conn) Subscribe(subject, queue string, handler func(protocol.Message)) (*Subscription, error)
• type Options
• type Subscription
  • func (s *Subscription) Close() error
  • func (s *Subscription) Queue() string
  • func (s *Subscription) Subject() string

Functions

func TrimNamespace

func TrimNamespace(nsp string) string

TrimNamespace trims leading '/' from a namespace for logging.

Types

type Conn

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

Conn is a single client session multiplexing data-plane and control-plane operations over one socket.io connection.

func Dial

func Dial(ctx context.Context, rawURL string, opts Options) (*Conn, error)

Dial connects to a gombus broker. rawURL must include the socket.io path, e.g. "http://localhost:8080/socket.io".

func (*Conn) AddConsumer

func (c *Conn) AddConsumer(req protocol.AddConsumerReq) error

AddConsumer attaches a consumer to a stream.

func (*Conn) Close

func (c *Conn) Close() error

Close terminates the session.

func (*Conn) Connected

func (c *Conn) Connected() bool

Connected reports whether the data namespace is connected.

func (*Conn) CreateStream

func (c *Conn) CreateStream(req protocol.CreateStreamReq) (*protocol.StreamMeta, error)

CreateStream creates a stream and returns its metadata.

func (*Conn) DeleteConsumer

func (c *Conn) DeleteConsumer(stream, name string) error

DeleteConsumer removes a consumer.

func (*Conn) DeleteStream

func (c *Conn) DeleteStream(name string) error

DeleteStream removes a stream and its data.

func (*Conn) Emit

func (c *Conn) Emit(nsp, event string, req any, out any) error

Emit sends an event on a namespace and waits for its acknowledgement, decoding the result into out (which may be nil). Peer connections and advanced clients use it for broker-to-broker and control-plane events.

func (*Conn) Publish

func (c *Conn) Publish(subject string, data []byte, header map[string]string) error

Publish sends a message on the data plane (at-most-once).

func (*Conn) PurgeStream

func (c *Conn) PurgeStream(name string) error

PurgeStream drops all stored messages of a stream.

func (*Conn) Request

func (c *Conn) Request(subject string, data []byte, timeout time.Duration) (protocol.ReplyMsg, error)

Request sends a request and waits for a reply. The server-side timeout bounds the wait; ErrNoResponders is returned when nobody replies.

func (*Conn) Stats

func (c *Conn) Stats() (protocol.StatsSnapshot, error)

Stats fetches the broker stats snapshot.

func (*Conn) StreamAck

func (c *Conn) StreamAck(consumer string, seq uint64) error

StreamAck acknowledges a delivered stream message.

func (*Conn) StreamInfo

func (c *Conn) StreamInfo(name string) (protocol.StreamInfoAck, error)

StreamInfo fetches stream metadata, consumers and cursors.

func (*Conn) StreamNak

func (c *Conn) StreamNak(consumer string, seq uint64, delay time.Duration) error

StreamNak redelivers a stream message immediately (or after delay).

func (*Conn) StreamPublish

func (c *Conn) StreamPublish(stream, subject string, data []byte, msgID string, header map[string]string) (protocol.StreamPublishAck, error)

StreamPublish appends a message to a stream. msgID (or a Header "Msg-Id") enables dedup within the stream's dedup window.

func (*Conn) StreamPull

func (c *Conn) StreamPull(consumer string, batch int, wait time.Duration) ([]protocol.Message, error)

StreamPull fetches up to batch messages for a pull consumer, waiting up to wait for the first one.

func (*Conn) Subscribe

func (c *Conn) Subscribe(subject, queue string, handler func(protocol.Message)) (*Subscription, error)

Subscribe registers a handler for a subject pattern, optionally inside a queue group. The handler is invoked for every matching message.

type Options

type Options struct {
	// Reconnection enables automatic reconnection after an unexpected
	// transport close.
	Reconnection bool
	// ControlToken authenticates the client on the /control namespace
	// ("" = no token required).
	ControlToken string
	Logger       *slog.Logger
}

Options configures a Conn.

type Subscription

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

Subscription is a registered subscription. It must be closed when done.

func (*Subscription) Close

func (s *Subscription) Close() error

Close unsubscribes from the broker and drops the local handler.

func (*Subscription) Queue

func (s *Subscription) Queue() string

Queue returns the queue group name ("" for plain pub/sub).

func (*Subscription) Subject

func (s *Subscription) Subject() string

Subject returns the subscribed pattern.