kafka

Documentation

Index

• type Sink
  • func NewSink[K eventsourcing.ID, PK eventsourcing.IDPt[K]](logger *slog.Logger, kvStore store.KVStore, topic string, brokers []string, ...) (*Sink[K], error)
  • func (s *Sink[K]) Close()
  • func (s *Sink[K]) ResumeTokens(ctx context.Context, forEach func(resumeToken encoding.Base64) error) (e error)
  • func (s *Sink[K]) SetCodec(codec sink.Codec[K])
  • func (s *Sink[K]) Sink(_ context.Context, e *eventsourcing.Event[K], m sink.Meta) (er 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]](logger *slog.Logger, kvStore store.KVStore, topic string, brokers []string, config *sarama.Config) (*Sink[K], error)

NewSink instantiate a Kafka 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) (e 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(_ context.Context, e *eventsourcing.Event[K], m sink.Meta) (er error)

Sink sends the event to the message queue