subscriber

Documentation

Index

• Constants
• Variables
• func ConcurrentSize(cfg mq.Config) int
• func WithResponseOnError(logger mq.Logger, handler mq.MsgHandler) mq.MsgHandler
• type Subscriber
  • func NewSubscriber(c mq.Client) *Subscriber
  • func (s *Subscriber) Close() error
  • func (s *Subscriber) ForceClose()
  • func (s *Subscriber) Get(subject, queue string) *Subscription
  • func (s *Subscriber) Subs() *memory.SafeMap[string, *Subscription]
  • func (s *Subscriber) Subscribe(subject, queue string, handler mq.MsgHandler)
  • func (s *Subscriber) SubscribeWithParameters(buffer int, timeout time.Duration, subject, queue string, ...)
  • func (s *Subscriber) Wait() error
• type Subscription
  • func NewSubscription(c mq.Client, subject, queue string) (*Subscription, error)
  • func (s *Subscription) Process(ctx context.Context, buffer int, timeout time.Duration, handler mq.MsgHandler) error
  • func (s *Subscription) Stop() error
  • func (s *Subscription) Sub() mq.Subscription
• type SubscriptionDetails

Constants

const (
	SubscriptionsPendingCount = "queue.subscriptions.pending.msgs"
	SubscriptionsPendingBytes = "queue.subscriptions.pending.bytes"
	SubscriptionsDroppedMsgs  = "queue.subscriptions.dropped.count"
	SubscriptionCountMsgs     = "queue.subscriptions.send.count"
)

const (
	HeaderConsumerError = "Error"
)

Variables

var ErrConnectionClosed = errors.New("connection closed")

Functions

func ConcurrentSize

func ConcurrentSize(cfg mq.Config) int

func WithResponseOnError

func WithResponseOnError(logger mq.Logger, handler mq.MsgHandler) mq.MsgHandler

WithResponseOnError wraps handler where if handler returns a non-nil error, the msg is then responded to with said error's string in the HeaderConsumerError header. Such message then causes nats publisher to return the error when reading response.

Note: this wrapper should NOT be used for observer-like handlers that do not send success responses. If subscribed to the same subject with an actual responder, the latter's response can potentially get snubbed.

Types

type Subscriber

type Subscriber struct {
	mq.Client
	*worker.Pool
	// contains filtered or unexported fields
}

func NewSubscriber

func NewSubscriber(c mq.Client) *Subscriber

func (*Subscriber) Close

func (s *Subscriber) Close() error

func (*Subscriber) ForceClose

func (s *Subscriber) ForceClose()

func (*Subscriber) Get

func (s *Subscriber) Get(subject, queue string) *Subscription

func (*Subscriber) Subs

func (s *Subscriber) Subs() *memory.SafeMap[string, *Subscription]

func (*Subscriber) Subscribe

func (s *Subscriber) Subscribe(subject, queue string, handler mq.MsgHandler)

func (*Subscriber) SubscribeWithParameters

func (s *Subscriber) SubscribeWithParameters(
	buffer int, timeout time.Duration, subject, queue string, handler mq.MsgHandler,
)

func (*Subscriber) Wait

func (s *Subscriber) Wait() error

type Subscription

type Subscription struct {
	mq.Client
	// contains filtered or unexported fields
}

func NewSubscription

func NewSubscription(c mq.Client, subject, queue string) (*Subscription, error)

func (*Subscription) Process

func (s *Subscription) Process(
	ctx context.Context,
	buffer int,
	timeout time.Duration,
	handler mq.MsgHandler,
) error

func (*Subscription) Stop

func (s *Subscription) Stop() error

func (*Subscription) Sub

func (s *Subscription) Sub() mq.Subscription

type SubscriptionDetails

type SubscriptionDetails struct {
	Pending      int64
	PendingBytes int64
	Dropped      int64
	Delivered    int64
}