consumer

Documentation

Index

• Constants
• func WithNatsOptionExitOnStreamNotFound(exitOnStreamNotFound bool) func(opt *NatsConsumerOptions)
• func WithNatsOptionInterceptors(interceptors ...UnaryInterceptorFunc) func(opt *NatsConsumerOptions)
• type ConfirmationType
• type Consumer
  • func NewNatsConsumer[T any](natsJetStream nats.JetStreamContext, cfg NatsConsumerConfiguration, fn Fn[T], ...) Consumer[T]
• type Fn
• type MessageRequest
• type NatsConsumer
  • func (n *NatsConsumer[T]) Close() error
  • func (n *NatsConsumer[T]) ConsumeAsync() error
• type NatsConsumerConfiguration
• type NatsConsumerOptions
• type Spec
• type UnaryFunc
• type UnaryInterceptorFunc

Constants

const (
	ConfirmationTypeNack = ConfirmationType(0)
	ConfirmationTypeAck  = ConfirmationType(1)
)

Functions

func WithNatsOptionExitOnStreamNotFound

func WithNatsOptionExitOnStreamNotFound(exitOnStreamNotFound bool) func(opt *NatsConsumerOptions)

func WithNatsOptionInterceptors

func WithNatsOptionInterceptors(interceptors ...UnaryInterceptorFunc) func(opt *NatsConsumerOptions)

Types

type ConfirmationType

type ConfirmationType byte

type Consumer

type Consumer[T any] interface {
	ConsumeAsync() error
	Close() error
}

func NewNatsConsumer

func NewNatsConsumer[T any](
	natsJetStream nats.JetStreamContext,
	cfg NatsConsumerConfiguration,
	fn Fn[T],
	options ...func(opts *NatsConsumerOptions),
) Consumer[T]

type Fn

type Fn[T any] func(ctx context.Context, event *common.Event[T]) (ConfirmationType, error)

type MessageRequest

type MessageRequest interface {
	Header() map[string][]string
	Spec() Spec
	Any() any
}

type NatsConsumer

type NatsConsumer[T any] struct {
	// contains filtered or unexported fields
}

func (*NatsConsumer[T]) Close

func (n *NatsConsumer[T]) Close() error

func (*NatsConsumer[T]) ConsumeAsync

func (n *NatsConsumer[T]) ConsumeAsync() error

type NatsConsumerConfiguration

type NatsConsumerConfiguration struct {
	Concurrency  int    `json:"Concurrency"`
	ConsumerName string `json:"ConsumerName"`
	Stream       string `json:"Stream"`
}

type NatsConsumerOptions

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

type Spec

type Spec struct {
	ConsumerName  string
	ConsumerQueue string
	Version       string
}

type UnaryFunc

type UnaryFunc = func(ctx context.Context, request MessageRequest) (ConfirmationType, error)

type UnaryInterceptorFunc

type UnaryInterceptorFunc = func(next UnaryFunc) UnaryFunc