nats

Documentation

Index

• type SubOption
  • func WithMsgCodec[K eventsourcing.ID](codec sink.Codec[K]) SubOption[K]
• type Subscriber
  • func NewSubscriber[K eventsourcing.ID, PK eventsourcing.IDPt[K]](logger *slog.Logger, js nats.JetStreamContext, topic projection.ConsumerTopic, ...) *Subscriber[K]
  • func NewSubscriberWithConn[K eventsourcing.ID, PK eventsourcing.IDPt[K]](logger *slog.Logger, nc *nats.Conn, topic projection.ConsumerTopic) (*Subscriber[K], error)
  • func NewSubscriberWithURL[K eventsourcing.ID, PK eventsourcing.IDPt[K]](ctx context.Context, logger *slog.Logger, url string, ...) (*Subscriber[K], error)
  • func (s *Subscriber[K]) Positions(ctx context.Context) (map[uint32]projection.SubscriberPosition, error)
  • func (s *Subscriber[K]) StartConsumer(ctx context.Context, subPos map[uint32]projection.SubscriberPosition, ...) (er error)
  • func (s *Subscriber[K]) TopicPartitions() (string, []uint32)

Types

type SubOption

type SubOption[K eventsourcing.ID] func(*Subscriber[K])

func WithMsgCodec

func WithMsgCodec[K eventsourcing.ID](codec sink.Codec[K]) SubOption[K]

type Subscriber

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

func NewSubscriber

func NewSubscriber[K eventsourcing.ID, PK eventsourcing.IDPt[K]](
	logger *slog.Logger,
	js nats.JetStreamContext,
	topic projection.ConsumerTopic,
	options ...SubOption[K],
) *Subscriber[K]

func NewSubscriberWithConn

func NewSubscriberWithConn[K eventsourcing.ID, PK eventsourcing.IDPt[K]](
	logger *slog.Logger,
	nc *nats.Conn,
	topic projection.ConsumerTopic,
) (*Subscriber[K], error)

func NewSubscriberWithURL

func NewSubscriberWithURL[K eventsourcing.ID, PK eventsourcing.IDPt[K]](
	ctx context.Context,
	logger *slog.Logger,
	url string,
	topic projection.ConsumerTopic,
) (*Subscriber[K], error)

func (*Subscriber[K]) Positions

func (s *Subscriber[K]) Positions(ctx context.Context) (map[uint32]projection.SubscriberPosition, error)

func (*Subscriber[K]) StartConsumer

func (s *Subscriber[K]) StartConsumer(ctx context.Context, subPos map[uint32]projection.SubscriberPosition, projName string, handler projection.ConsumerHandler[K], options ...projection.ConsumerOption[K]) (er error)

func (*Subscriber[K]) TopicPartitions

func (s *Subscriber[K]) TopicPartitions() (string, []uint32)