events

package
v0.3.0 Latest Latest
Warning

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

 Go to latest Published: Jul 4, 2026 License: MIT Imports: 9 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) 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) 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)

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
	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) 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.