mq

Documentation

Index

• func DLQStreamName() string
• func StreamName() string
• type EmbeddedNATS
  • func NewEmbedded(storeDir string, maxBytes int64, logger ...*slog.Logger) (*EmbeddedNATS, error)
  • func (e *EmbeddedNATS) Close() error
  • func (e *EmbeddedNATS) GetServer() *natsserver.Server
  • func (e *EmbeddedNATS) JetStream() jetstream.JetStream
  • func (e *EmbeddedNATS) NatsConn() *nats.Conn
  • func (e *EmbeddedNATS) Publish(ctx context.Context, subject string, data []byte, opts ...PublishOpt) error
  • func (e *EmbeddedNATS) Subscribe(ctx context.Context, subject, consumerName string, ...) error
• type Message
  • func NewMessage(ctx context.Context, subject string, data []byte, ts time.Time, ...) *Message
  • func (m *Message) Ack() error
  • func (m *Message) DoubleAck(ctx context.Context) error
  • func (m *Message) Nak() error
• type PublishOpt
  • func WithHeader(key, value string) PublishOpt
• type Publisher
• type Subscriber

Functions

func DLQStreamName

func DLQStreamName() string

DLQStreamName returns the dead-letter-queue JetStream stream name. Used for failed ClickHouse batch inserts (see `internal/api/dlq.go`). Same rationale as StreamName for being hardcoded.

func StreamName

func StreamName() string

StreamName returns the primary JetStream stream name. Hardcoded — the embedded NATS server is private to the WaveHouse process, so there's no multi-tenant case to namespace against. Kept as a function (rather than a const) so all callers go through one symbol; future migrations could swap the constant out without breaking the API surface.

Types

type EmbeddedNATS

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

EmbeddedNATS runs an in-process NATS server with JetStream.

func NewEmbedded

func NewEmbedded(storeDir string, maxBytes int64, logger ...*slog.Logger) (*EmbeddedNATS, error)

NewEmbedded starts an embedded NATS server with JetStream enabled. An optional *slog.Logger can be passed to control server log output; if omitted, slog.Default() is used. The stream name is fixed (see StreamName / DLQStreamName) — the embedded server is private to this process, so there's no namespacing to do.

func (*EmbeddedNATS) Close

func (e *EmbeddedNATS) Close() error

func (*EmbeddedNATS) GetServer

func (e *EmbeddedNATS) GetServer() *natsserver.Server

func (*EmbeddedNATS) JetStream

func (e *EmbeddedNATS) JetStream() jetstream.JetStream

JetStream returns the underlying JetStream handle for direct access (e.g. gap-fill).

func (*EmbeddedNATS) NatsConn

func (e *EmbeddedNATS) NatsConn() *nats.Conn

func (*EmbeddedNATS) Publish

func (e *EmbeddedNATS) Publish(ctx context.Context, subject string, data []byte, opts ...PublishOpt) error

func (*EmbeddedNATS) Subscribe

func (e *EmbeddedNATS) Subscribe(ctx context.Context, subject, consumerName string, handler func(msg *Message) error) error

type Message

type Message struct {
	Ctx       context.Context
	Subject   string
	Data      []byte
	Timestamp time.Time
	// contains filtered or unexported fields
}

Message represents a message received from the queue.

func NewMessage

func NewMessage(ctx context.Context, subject string, data []byte, ts time.Time, doubleAck func(context.Context) error, ack func() error, nak func() error) *Message

NewMessage constructs a Message with ack/nak callbacks.

func (*Message) Ack

func (m *Message) Ack() error

Ack acknowledges the message asynchronously (fire-and-forget). Use for low-criticality consumers or high-throughput scenarios where latency is more important than a "received" confirmation from the server.

func (*Message) DoubleAck

func (m *Message) DoubleAck(ctx context.Context) error

DoubleAck acknowledges the message synchronously, blocking until the NATS server confirms receipt. Use this for critical ingest paths (ClickHouse writes).

func (*Message) Nak

func (m *Message) Nak() error

Nak negatively acknowledges the message asynchronously for redelivery. This is fire-and-forget and does not require a context.

type PublishOpt

type PublishOpt func(*nats.Msg)

PublishOpt defines a functional option for modifying a NATS message before publishing.

func WithHeader

func WithHeader(key, value string) PublishOpt

WithHeader adds a key-value pair to the NATS message headers.

type Publisher

type Publisher interface {
	Publish(ctx context.Context, subject string, data []byte, opts ...PublishOpt) error
	Close() error
}

Publisher publishes messages to a subject.

type Subscriber

type Subscriber interface {
	// Subscribe registers a handler for incoming messages.
	//
	// CONTRACT: If the handler intends to return an error to trigger automatic
	// redelivery, it MUST NOT manually call msg.Ack() or msg.Nak() beforehand.
	//
	// CONTRACT: Symmetrically, if you explicitly call msg.Nak(), do NOT also
	// return an error — the consume loop will call Nak() again on any non-nil
	// error return.
	//
	// CONTRACT: Calling msg.DoubleAck(ctx) and then returning a non-nil error is
	// undefined behaviour — the consume loop will Nak() after a successful
	// server-confirmed Ack. Call DoubleAck, then return nil on success.
	Subscribe(ctx context.Context, subject, consumerName string, handler func(msg *Message) error) error
	Close() error
}

Subscriber subscribes to messages on a subject.