jetconsumer

Documentation

Index

• Variables
• type AccessLogOption
  • func WithAccessLogLevel(level slog.Level) AccessLogOption
  • func WithAccessLogMessage(message string) AccessLogOption
• type Consumer
  • func (c *Consumer) Start() (err error)
  • func (c *Consumer) Stop()
• type Handler
  • func ChainMiddleware(h Handler, mws ...Middleware) Handler
  • func HandleType[T any](h TypedHandler[T]) Handler
• type HandlerFunc
  • func (f HandlerFunc) Serve(ctx context.Context, msg jetstream.Msg) error
• type JetstreamConsumer
• type Middleware
  • func Router(routes ...Route) Middleware
  • func WithAccessLog(logger *slog.Logger, opts ...AccessLogOption) Middleware
  • func WithAck() Middleware
  • func WithAckOnArrival() Middleware
  • func WithErrorLog(logger *slog.Logger, propagate bool) Middleware
  • func WithNak() Middleware
• type Route
  • func Handle(subj string, handler Handler) Route
• type TypedHandler
• type TypedHandlerFunc
  • func (f TypedHandlerFunc[T]) Serve(arg peanats.TypedRequest[T]) error
• type TypedMessage

Variables

var NotFoundHandler = HandlerFunc(func(_ context.Context, _ jetstream.Msg) error {
	return peanats.Error{
		Code:    http.StatusNotFound,
		Message: "handler not found",
		Cause:   nil,
	}
})

Types

type AccessLogOption

type AccessLogOption func(*accessLogParams)

func WithAccessLogLevel

func WithAccessLogLevel(level slog.Level) AccessLogOption

func WithAccessLogMessage

func WithAccessLogMessage(message string) AccessLogOption

type Consumer

type Consumer struct {
	BaseContext context.Context
	Consumer    JetstreamConsumer
	Handler     Handler
	Executor    func(func())
	// contains filtered or unexported fields
}

func (*Consumer) Start

func (c *Consumer) Start() (err error)

func (*Consumer) Stop

func (c *Consumer) Stop()

type Handler

type Handler interface {
	Serve(ctx context.Context, msg jetstream.Msg) error
}

func ChainMiddleware

func ChainMiddleware(h Handler, mws ...Middleware) Handler

func HandleType

func HandleType[T any](h TypedHandler[T]) Handler

type HandlerFunc

type HandlerFunc func(ctx context.Context, msg jetstream.Msg) error

func (HandlerFunc) Serve

func (f HandlerFunc) Serve(ctx context.Context, msg jetstream.Msg) error

type JetstreamConsumer

type JetstreamConsumer interface {
	Consume(jetstream.MessageHandler, ...jetstream.PullConsumeOpt) (jetstream.ConsumeContext, error)
}

type Middleware

type Middleware func(handler Handler) Handler

func Router

func Router(routes ...Route) Middleware

func WithAccessLog

func WithAccessLog(logger *slog.Logger, opts ...AccessLogOption) Middleware

WithAccessLog returns a middleware that logs access using the provided slog.Logger.

func WithAck

func WithAck() Middleware

WithAck returns a middleware that ACKs the message after handling it successfully.

func WithAckOnArrival

func WithAckOnArrival() Middleware

WithAckOnArrival returns a middleware that ACKs the message immediately on arrival.

func WithErrorLog

func WithErrorLog(logger *slog.Logger, propagate bool) Middleware

WithErrorLog returns a middleware that logs errors using the provided logger. If propagate is true, the error is propagated to the outer handler.

func WithNak

func WithNak() Middleware

WithNak returns a middleware that NAKs the message if the handler returns an error.

type Route

type Route interface {
	peanats.Matcher
	Handler
}

func Handle

func Handle(subj string, handler Handler) Route

type TypedHandler

type TypedHandler[T any] interface {
	Serve(context.Context, TypedMessage[T]) error
}

type TypedHandlerFunc

type TypedHandlerFunc[T any] func(peanats.TypedRequest[T]) error

func (TypedHandlerFunc[T]) Serve

func (f TypedHandlerFunc[T]) Serve(arg peanats.TypedRequest[T]) error

type TypedMessage

type TypedMessage[T any] interface {
	jetstream.Msg
	Payload() *T
}