axrmq

Documentation

Overview

Package axrmq provides a RabbitMQ-based message transport.

Index

• Variables
• type Acknowledger
  • func (a *Acknowledger) Ack(_ context.Context) error
  • func (a *Acknowledger) Reject(ctx context.Context, _ error) error
  • func (a *Acknowledger) Retry(ctx context.Context, _ error, d time.Duration) error
• type Transport
  • func (t *Transport) Initialize(ctx context.Context, ep string) error
  • func (t *Transport) Receive(ctx context.Context) (env endpoint.InboundEnvelope, ack endpoint.Acknowledger, err error)
  • func (t *Transport) Send(ctx context.Context, env endpoint.OutboundEnvelope) error
  • func (t *Transport) Subscribe(ctx context.Context, op endpoint.Operation, mt ax.MessageTypeSet) error

Variables

var DefaultReceiveConcurrency = runtime.NumCPU() * 2

DefaultReceiveConcurrency is the default number of messages to process concurrently.

var DefaultSendConcurrency = runtime.NumCPU() * 10

DefaultSendConcurrency is the default number of messages to send concurrently.

Types

type Acknowledger

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

Acknowledger is an implementation of bus.Acknowledger that acknowledges AMQP messages.

func (*Acknowledger) Ack

func (a *Acknowledger) Ack(_ context.Context) error

Ack acknowledges the message, indicating that is was handled successfully and does not need to be retried.

func (*Acknowledger) Reject

func (a *Acknowledger) Reject(ctx context.Context, _ error) error

Reject indicates that the message could not be handled and should not be retried. Depending on the transport, this may move the message to some form of error queue or otherwise drop the message completely.

func (*Acknowledger) Retry

func (a *Acknowledger) Retry(ctx context.Context, _ error, d time.Duration) error

Retry requeues the message so that it is retried at some point in the future.

d is a hint as to how long the transport should wait before retrying this message.

type Transport

type Transport struct {
	Conn               *amqp.Connection
	Exclusive          bool
	SendConcurrency    int
	ReceiveConcurrency int
	Tracer             opentracing.Tracer
	// contains filtered or unexported fields
}

Transport is an implementation of endpoint.InboundTransport and endpoint.OutboundTransport that uses RabbitMQ to communicate messages between endpoints.

func (*Transport) Initialize

func (t *Transport) Initialize(ctx context.Context, ep string) error

Initialize sets up the transport to communicate as an endpoint named ep.

func (*Transport) Receive

func (t *Transport) Receive(ctx context.Context) (env endpoint.InboundEnvelope, ack endpoint.Acknowledger, err error)

Receive returns the next message sent to this endpoint. It blocks until a message is available, or ctx is canceled.

func (*Transport) Send

func (t *Transport) Send(ctx context.Context, env endpoint.OutboundEnvelope) error

Send sends env via the transport.

func (*Transport) Subscribe

func (t *Transport) Subscribe(ctx context.Context, op endpoint.Operation, mt ax.MessageTypeSet) error

Subscribe instructs the transport to listen to multicast messages of the given type.