pubsubnats

Documentation

Index

• func EncodeJSONRequest(_ context.Context, msg *nats.Msg, request interface{}) error
• func NopRequestDecoder(_ context.Context, _ *nats.Msg) (interface{}, error)
• func WithDefaultConnectOptions(natsClientName string, logger log.Logger) []nats.Option
• type DecodeRequestFunc
• type Handler
  • func NewRecoveryMiddleware(l log.Logger, h Handler) Handler
  • func NewSubscriber(e endpoint.Endpoint, dec natstransport.DecodeRequestFunc, ...) Handler
• type Publisher
  • func NewPublisher(publisher *nats.Conn, options ...PublisherOption) *Publisher
  • func (p Publisher) Publish(ctx context.Context, subject string, e interface{}) error
• type PublisherOption
  • func PublisherBefore(before ...natstransport.RequestFunc) PublisherOption
  • func PublisherLogger(l log.Logger) PublisherOption
  • func PublisherTimeout(timeout time.Duration) PublisherOption
  • func PublisherVerbose() PublisherOption
• type RecoveryMiddleware
  • func (mw RecoveryMiddleware) ServeMsg(nc *nats.Conn) func(msg *nats.Msg)
• type Subscriber
  • func (s Subscriber) ServeMsg(nc *nats.Conn) func(msg *nats.Msg)
• type SubscriberFunc
• type SubscriberOption
  • func SubscriberBefore(before ...natstransport.RequestFunc) SubscriberOption
  • func SubscriberErrorLogger(logger log.Logger) SubscriberOption
• type SubscriptionSet
  • func RegisterSubscribers(funcs ...SubscriberFunc) SubscriptionSet
  • func (registry SubscriptionSet) Close() error
  • func (registry SubscriptionSet) Register(funcs ...SubscriberFunc) SubscriptionSet

Functions

func EncodeJSONRequest

func EncodeJSONRequest(_ context.Context, msg *nats.Msg, request interface{}) error

EncodeJSONRequest is an EncodeRequestFunc that serializes the request as a JSON object to the Data of the Msg. Many JSON-over-NATS services can use it as a sensible default.

func NopRequestDecoder

func NopRequestDecoder(_ context.Context, _ *nats.Msg) (interface{}, error)

NopRequestDecoder is a DecodeRequestFunc that can be used for requests that do not need to be decoded, and simply returns nil, nil.

func WithDefaultConnectOptions

func WithDefaultConnectOptions(natsClientName string, logger log.Logger) []nats.Option

Types

type DecodeRequestFunc

type DecodeRequestFunc natstransport.DecodeRequestFunc

DecodeRequestFunc extracts a user-domain request object from a publisher request object. It's designed to be used in NATS subscribers, for subscriber-side endpoints. One straightforward DecodeRequestFunc could be something that JSON decodes from the request body to the concrete response type.

type Handler

type Handler interface {
	ServeMsg(nc *nats.Conn) func(msg *nats.Msg)
}

Handler serves messages for NATS

func NewRecoveryMiddleware

func NewRecoveryMiddleware(l log.Logger, h Handler) Handler

NewRecoveryMiddleware returns as RecoveryMiddleware handler

func NewSubscriber

func NewSubscriber(
	e endpoint.Endpoint,
	dec natstransport.DecodeRequestFunc,
	options ...SubscriberOption,
) Handler

NewSubscriber constructs a new subscriber, which provides nats.MsgHandler and wraps the provided endpoint.

type Publisher

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

Publisher wraps a URL and provides a method that implements endpoint.Endpoint.

func NewPublisher

func NewPublisher(
	publisher *nats.Conn,
	options ...PublisherOption,
) *Publisher

NewPublisher constructs a usable Publisher for a single remote method.

func (Publisher) Publish

func (p Publisher) Publish(ctx context.Context, subject string, e interface{}) error

Publish returns a usable endpoint that invokes the remote endpoint.

type PublisherOption

type PublisherOption func(*Publisher)

PublisherOption sets an optional parameter for clients.

func PublisherBefore

func PublisherBefore(before ...natstransport.RequestFunc) PublisherOption

PublisherBefore sets the RequestFuncs that are applied to the outgoing NATS request before it's invoked.

func PublisherLogger

func PublisherLogger(l log.Logger) PublisherOption

func PublisherTimeout

func PublisherTimeout(timeout time.Duration) PublisherOption

PublisherTimeout sets the available timeout for NATS request.

func PublisherVerbose

func PublisherVerbose() PublisherOption

type RecoveryMiddleware

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

RecoveryMiddleware intercepts the nats messages and performs recovery on panic

func (RecoveryMiddleware) ServeMsg

func (mw RecoveryMiddleware) ServeMsg(nc *nats.Conn) func(msg *nats.Msg)

ServeMsg wraps the serveMsg handler with the stacktrace

type Subscriber

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

Subscriber wraps an endpoint and provides nats.MsgHandler.

func (Subscriber) ServeMsg

func (s Subscriber) ServeMsg(nc *nats.Conn) func(msg *nats.Msg)

Serve provides nats.MsgHandler.

type SubscriberFunc

type SubscriberFunc func() (*nats.Subscription, error)

type SubscriberOption

type SubscriberOption func(*Subscriber)

SubscriberOption sets an optional parameter for subscribers.

func SubscriberBefore

func SubscriberBefore(before ...natstransport.RequestFunc) SubscriberOption

SubscriberBefore functions are executed on the subscriber request object before the request is decoded.

func SubscriberErrorLogger

func SubscriberErrorLogger(logger log.Logger) SubscriberOption

SubscriberErrorLogger is used to log non-terminal errors. By default, no errors are logged. This is intended as a diagnostic measure. Finer-grained control of error handling, including logging in more detail, should be performed in a custom SubscriberErrorEncoder which has access to the context.

type SubscriptionSet

type SubscriptionSet struct {
	io.Closer
	Err error
	// contains filtered or unexported fields
}

func RegisterSubscribers

func RegisterSubscribers(funcs ...SubscriberFunc) SubscriptionSet

func (SubscriptionSet) Close

func (registry SubscriptionSet) Close() error

func (SubscriptionSet) Register

func (registry SubscriptionSet) Register(funcs ...SubscriberFunc) SubscriptionSet