Documentation
¶
Index ¶
- Constants
- Variables
- func ConcurrentSize(cfg mq.Config) int
- func WithResponseOnError(logger mq.Logger, handler mq.MsgHandler) mq.MsgHandler
- type 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
- type SubscriptionDetails
Constants ¶
View Source
const ( SubscriptionsPendingCount = "queue.subscriptions.pending.msgs" SubscriptionsPendingBytes = "queue.subscriptions.pending.bytes" SubscriptionsDroppedMsgs = "queue.subscriptions.dropped.count" SubscriptionCountMsgs = "queue.subscriptions.send.count" )
View Source
const (
HeaderConsumerError = "Error"
)
Variables ¶
View Source
var ErrConnectionClosed = errors.New("connection closed")
Functions ¶
func ConcurrentSize ¶
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 ¶
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 ¶
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
Source Files
Directories
Click to show internal directories.
Click to hide internal directories.