events

Documentation

Index

• func ConsumePull[T any](ctx context.Context, js nats.JetStreamContext, cfg ConsumerConfig, ...) error
• func ConsumePush[T any](ctx context.Context, js nats.JetStreamContext, cfg ConsumerConfig, ...) error
• func WaitForNATS(ctx context.Context, url string, retries int, delay time.Duration) error
• type AckAction
• type Cache
  • func NewCache[T any](kv nats.KeyValue, ttl time.Duration) *Cache[T]
  • func (c *Cache[T]) Delete(ctx context.Context, key string) error
  • func (c *Cache[T]) Get(ctx context.Context, key string) (*T, error)
  • func (c *Cache[T]) GetOrSet(ctx context.Context, key string, fn func() (*T, error)) (*T, error)
  • func (c *Cache[T]) Set(ctx context.Context, key string, val T) error
• type Conn
  • func Connect(ctx context.Context, opts ConnOptions) (*Conn, error)
  • func (c *Conn) Context() context.Context
  • func (c *Conn) Drain()
  • func (c *Conn) EnsureKeyValue(cfg KVConfig) (nats.KeyValue, error)
  • func (c *Conn) EnsureStream(cfg StreamConfig) error
  • func (c *Conn) EnsureStreams(configs ...StreamConfig) error
• type ConnOptions
• type ConsumerConfig
  • func DefaultConsumerConfig(stream, durable string) ConsumerConfig
• type Handler
• type KVConfig
  • func DefaultKVConfig(bucket string) KVConfig
• type Msg
• type Producer
  • func NewProducer[T any](nc *nats.Conn, js nats.JetStreamContext, subject string) *Producer[T]
  • func (p *Producer[T]) Publish(msg T, opts ...nats.PubOpt) error
  • func (p *Producer[T]) PublishAndWait(ctx context.Context, msg T, timeout time.Duration) (*Msg[T], error)
  • func (p *Producer[T]) PublishAndWaitRaw(ctx context.Context, data []byte, timeout time.Duration) (*nats.Msg, error)
  • func (p *Producer[T]) PublishWithID(msg T, id string, opts ...nats.PubOpt) error
• type StreamConfig
  • func DefaultStreamConfig(name string) StreamConfig

Functions

func ConsumePull

func ConsumePull[T any](ctx context.Context, js nats.JetStreamContext, cfg ConsumerConfig, handler Handler[T]) error

func ConsumePush

func ConsumePush[T any](ctx context.Context, js nats.JetStreamContext, cfg ConsumerConfig, handler Handler[T]) error

func WaitForNATS

func WaitForNATS(ctx context.Context, url string, retries int, delay time.Duration) error

Types

type AckAction

type AckAction int

const (
	Ack AckAction = iota
	Nak
	NakDelay
	Term
)

type Cache

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

func NewCache

func NewCache[T any](kv nats.KeyValue, ttl time.Duration) *Cache[T]

func (*Cache[T]) Delete

func (c *Cache[T]) Delete(ctx context.Context, key string) error

func (*Cache[T]) Get

func (c *Cache[T]) Get(ctx context.Context, key string) (*T, error)

func (*Cache[T]) GetOrSet

func (c *Cache[T]) GetOrSet(ctx context.Context, key string, fn func() (*T, error)) (*T, error)

func (*Cache[T]) Set

func (c *Cache[T]) Set(ctx context.Context, key string, val T) error

type Conn

type Conn struct {
	NC *nats.Conn
	JS nats.JetStreamContext
	// contains filtered or unexported fields
}

func Connect

func Connect(ctx context.Context, opts ConnOptions) (*Conn, error)

func (*Conn) Context

func (c *Conn) Context() context.Context

func (*Conn) Drain

func (c *Conn) Drain()

func (*Conn) EnsureKeyValue

func (c *Conn) EnsureKeyValue(cfg KVConfig) (nats.KeyValue, error)

func (*Conn) EnsureStream

func (c *Conn) EnsureStream(cfg StreamConfig) error

func (*Conn) EnsureStreams

func (c *Conn) EnsureStreams(configs ...StreamConfig) error

type ConnOptions

type ConnOptions struct {
	URL           string
	MaxReconnects int
	ReconnectWait time.Duration
	Timeout       time.Duration
	RetryOnFail   bool
}

type ConsumerConfig

type ConsumerConfig struct {
	Stream      string
	Subject     string
	Durable     string
	QueueGroup  string
	DeliverAll  bool
	MaxDeliver  int
	AckWait     time.Duration
	BackOff     []time.Duration
	PullBatch   int
	PullMaxWait time.Duration
	NakDelay    time.Duration
	Reply       bool // if true, handler returns data for msg.Respond()
}

func DefaultConsumerConfig

func DefaultConsumerConfig(stream, durable string) ConsumerConfig

type Handler

type Handler[T any] func(ctx context.Context, msg Msg[T]) (AckAction, error)

type KVConfig

type KVConfig struct {
	Bucket      string
	Description string
	TTL         time.Duration
	MaxBytes    int64
	Storage     nats.StorageType
}

func DefaultKVConfig

func DefaultKVConfig(bucket string) KVConfig

type Msg

type Msg[T any] struct {
	Data T
	Raw  *nats.Msg
	Ack  func(opts ...nats.AckOpt) error
	Nak  func(opts ...nats.AckOpt) error
	NakD func(delay time.Duration, opts ...nats.AckOpt) error
	Term func(opts ...nats.AckOpt) error
}

type Producer

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

func NewProducer

func NewProducer[T any](nc *nats.Conn, js nats.JetStreamContext, subject string) *Producer[T]

func (*Producer[T]) Publish

func (p *Producer[T]) Publish(msg T, opts ...nats.PubOpt) error

func (*Producer[T]) PublishAndWait

func (p *Producer[T]) PublishAndWait(ctx context.Context, msg T, timeout time.Duration) (*Msg[T], error)

PublishAndWait publishes a message and waits for a reply (NATS request-reply). Returns the typed response or an error. Uses core NATS Request, not JetStream.

func (*Producer[T]) PublishAndWaitRaw

func (p *Producer[T]) PublishAndWaitRaw(ctx context.Context, data []byte, timeout time.Duration) (*nats.Msg, error)

PublishAndWaitRaw publishes and waits for a raw byte reply.

func (*Producer[T]) PublishWithID

func (p *Producer[T]) PublishWithID(msg T, id string, opts ...nats.PubOpt) error

type StreamConfig

type StreamConfig struct {
	Name        string
	Subjects    []string
	MaxAge      time.Duration
	MaxBytes    int64
	Storage     nats.StorageType
	Compression nats.StoreCompression
}

func DefaultStreamConfig

func DefaultStreamConfig(name string) StreamConfig