nats

Documentation

Index

• func ComposeTopic(topic string, partitionID uint32) (_ string, e error)
• type Sink
  • func NewSink[K eventsourcing.ID, PK eventsourcing.IDPt[K]](kvStore store.KVStore, logger *slog.Logger, topic string, ...) (_ *Sink[K], err error)
  • func (s *Sink[K]) Close()
  • func (s *Sink[K]) ResumeTokens(ctx context.Context, forEach func(resumeToken encoding.Base64) error) error
  • func (s *Sink[K]) SetCodec(codec sink.Codec[K])
  • func (s *Sink[K]) Sink(ctx context.Context, e *eventsourcing.Event[K], m sink.Meta) error

Functions

func ComposeTopic

func ComposeTopic(topic string, partitionID uint32) (_ string, e error)

Types

type Sink

type Sink[K eventsourcing.ID] struct {
	// contains filtered or unexported fields
}

func NewSink

func NewSink[K eventsourcing.ID, PK eventsourcing.IDPt[K]](kvStore store.KVStore, logger *slog.Logger, topic string, totalPartitions uint32, url string, options ...nats.Option) (_ *Sink[K], err error)

NewSink instantiate NATS sink

func (*Sink[K]) Close

func (s *Sink[K]) Close()

func (*Sink[K]) ResumeTokens

func (s *Sink[K]) ResumeTokens(ctx context.Context, forEach func(resumeToken encoding.Base64) error) error

ResumeTokens iterates over all the last saved resumed token per partition It will return 0 if there is no last message

func (*Sink[K]) SetCodec

func (s *Sink[K]) SetCodec(codec sink.Codec[K])

func (*Sink[K]) Sink

func (s *Sink[K]) Sink(ctx context.Context, e *eventsourcing.Event[K], m sink.Meta) error

Sink sends the event to the message queue