events

package
v0.7.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 12, 2026 License: MIT Imports: 10 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

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 CompressionType added in v0.1.0 

type CompressionType int
const (
	S2Compression CompressionType = iota
	NoCompression
)

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) Close added in v0.1.0 

func (c *Conn) Close() 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

func (*Conn) KVDelete added in v0.6.0 

func (c *Conn) KVDelete(bucket, key string) error

func (*Conn) KVGet added in v0.6.0 

func (c *Conn) KVGet(bucket, key string) ([]byte, error)

func (*Conn) KVKeys added in v0.7.0 

func (c *Conn) KVKeys(bucket string) ([]string, error)

func (*Conn) KVPut added in v0.6.0 

func (c *Conn) KVPut(bucket, key string, value []byte) (uint64, error)

func (*Conn) KVReset added in v0.7.0 

func (c *Conn) KVReset(bucket string) error

func (*Conn) Name added in v0.1.0 

func (c *Conn) Name() string

func (*Conn) Publish added in v0.1.0 

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

func (*Conn) PublishJSON added in v0.6.0 

func (c *Conn) PublishJSON(ctx context.Context, subject string, msg any) error

func (*Conn) PullSubscribe added in v0.1.0 

func (c *Conn) PullSubscribe(ctx context.Context, subject string, durable string) (PullConsumer, error)

func (*Conn) Request added in v0.1.0 

func (c *Conn) Request(ctx context.Context, subject string, data []byte, timeout time.Duration) ([]byte, error)

func (*Conn) Subscribe added in v0.1.0 

func (c *Conn) Subscribe(ctx context.Context, subject string, durable string, handler MessageHandler) (Subscription, error)

func (*Conn) SubscribeRaw added in v0.6.0 

func (c *Conn) SubscribeRaw(subject string, handler func(msg []byte)) error

func (*Conn) SubscribeRawReply added in v0.6.0 

func (c *Conn) SubscribeRawReply(subject string, handler func(msg []byte) []byte) error

type ConnOptions 

type ConnOptions struct {
	Name          string
	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 EventBroker added in v0.1.0 

type EventBroker interface {
	Name() string
	Publish(ctx context.Context, subject string, data []byte) error
	PublishJSON(ctx context.Context, subject string, msg any) error
	Subscribe(ctx context.Context, subject string, durable string, handler MessageHandler) (Subscription, error)
	PullSubscribe(ctx context.Context, subject string, durable string) (PullConsumer, error)
	Request(ctx context.Context, subject string, data []byte, timeout time.Duration) ([]byte, error)
	EnsureStream(cfg StreamConfig) error
	EnsureStreams(configs ...StreamConfig) error
	Close() error
}

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     StorageType
}

func DefaultKVConfig 

func DefaultKVConfig(bucket string) KVConfig

type KafkaBroker added in v0.1.0 

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

func NewKafkaBroker added in v0.1.0 

func NewKafkaBroker(name string, brokers []string, consumerGroup string) *KafkaBroker

func (*KafkaBroker) Close added in v0.1.0 

func (b *KafkaBroker) Close() error

func (*KafkaBroker) EnsureStream added in v0.1.0 

func (b *KafkaBroker) EnsureStream(cfg StreamConfig) error

func (*KafkaBroker) EnsureStreams added in v0.1.0 

func (b *KafkaBroker) EnsureStreams(configs ...StreamConfig) error

func (*KafkaBroker) Name added in v0.1.0 

func (b *KafkaBroker) Name() string

func (*KafkaBroker) Publish added in v0.1.0 

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

func (*KafkaBroker) PublishJSON added in v0.6.0 

func (b *KafkaBroker) PublishJSON(ctx context.Context, subject string, msg any) error

func (*KafkaBroker) PullSubscribe added in v0.1.0 

func (b *KafkaBroker) PullSubscribe(_ context.Context, _ string, _ string) (PullConsumer, error)

func (*KafkaBroker) Request added in v0.1.0 

func (b *KafkaBroker) Request(_ context.Context, _ string, _ []byte, _ time.Duration) ([]byte, error)

func (*KafkaBroker) Subscribe added in v0.1.0 

func (b *KafkaBroker) Subscribe(ctx context.Context, subject string, durable string, handler MessageHandler) (Subscription, error)

type Message added in v0.1.0 

type Message interface {
	Data() []byte
	Subject() string
	Ack() error
	Nak(delay ...time.Duration) error
	Term() error
	Respond(data []byte) error
}

type MessageHandler added in v0.1.0 

type MessageHandler func(ctx context.Context, msg Message) error

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 PullConsumer added in v0.1.0 

type PullConsumer interface {
	Fetch(batch int, maxWait time.Duration) ([]Message, error)
	Unsubscribe() error
}

type StorageType added in v0.1.0 

type StorageType int
const (
	FileStorage StorageType = iota
	MemoryStorage
)

type StreamConfig 

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

func DefaultStreamConfig 

func DefaultStreamConfig(name string) StreamConfig

type Subscription added in v0.1.0 

type Subscription interface {
	Unsubscribe() error
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.