amqp

Documentation

Overview

Module amqp implements an AMQP binding using pack.ag/amqp module

Index

• func NewReceiver(amqp *amqp.Receiver) protocol.Receiver
• func NewSender(amqpSender *amqp.Sender, options ...SenderOptionFunc) protocol.Sender
• func WriteMessage(ctx context.Context, m binding.Message, amqpMessage *amqp.Message, ...) error
• type Message
  • func NewMessage(message *amqp.Message) *Message
  • func (m *Message) Finish(err error) error
  • func (m *Message) ReadBinary(ctx context.Context, encoder binding.BinaryWriter) error
  • func (m *Message) ReadEncoding() binding.Encoding
  • func (m *Message) ReadStructured(ctx context.Context, encoder binding.StructuredWriter) error
• type Option
  • func WithConnOpt(opt amqp.ConnOption) Option
  • func WithConnSASLPlain(username, password string) Option
  • func WithReceiverLinkOption(opt amqp.LinkOption) Option
  • func WithSenderLinkOption(opt amqp.LinkOption) Option
  • func WithSessionOpt(opt amqp.SessionOption) Option
• type Protocol
  • func New(server, queue string, opts ...Option) (*Protocol, error)
  • func (t *Protocol) Close() error
  • func (t *Protocol) HasTracePropagation() bool
  • func (t *Protocol) OpenInbound(ctx context.Context) error
  • func (t *Protocol) Receive(ctx context.Context) (binding.Message, error)
  • func (t *Protocol) Send(ctx context.Context, in binding.Message) error
• type SenderOptionFunc
  • func WithTransformer(transformer binding.TransformerFactory) SenderOptionFunc

Functions

func NewReceiver

func NewReceiver(amqp *amqp.Receiver) protocol.Receiver

Create a new Receiver which wraps an amqp.Receiver in a binding.Receiver

func NewSender

func NewSender(amqpSender *amqp.Sender, options ...SenderOptionFunc) protocol.Sender

Create a new Sender which wraps an amqp.Sender in a binding.Sender

func WriteMessage

func WriteMessage(ctx context.Context, m binding.Message, amqpMessage *amqp.Message, transformers ...binding.TransformerFactory) error

Fill the provided amqpMessage with the message m. Using context you can tweak the encoding processing (more details on binding.Write documentation).

Types

type Message

type Message struct {
	AMQP *amqp.Message
	// contains filtered or unexported fields
}

Message implements binding.Message by wrapping an *amqp.Message. This message *can* be read several times safely

func NewMessage

func NewMessage(message *amqp.Message) *Message

Wrap an *amqp.Message in a binding.Message. The returned message *can* be read several times safely

func (*Message) Finish

func (m *Message) Finish(err error) error

func (*Message) ReadBinary

func (m *Message) ReadBinary(ctx context.Context, encoder binding.BinaryWriter) error

func (*Message) ReadEncoding

func (m *Message) ReadEncoding() binding.Encoding

func (*Message) ReadStructured

func (m *Message) ReadStructured(ctx context.Context, encoder binding.StructuredWriter) error

type Option

type Option func(*Protocol) error

Option is the function signature required to be considered an amqp.Option.

func WithConnOpt

func WithConnOpt(opt amqp.ConnOption) Option

WithConnOpt sets a connection option for amqp

func WithConnSASLPlain

func WithConnSASLPlain(username, password string) Option

WithConnSASLPlain sets SASLPlain connection option for amqp

func WithReceiverLinkOption

func WithReceiverLinkOption(opt amqp.LinkOption) Option

WithReceiverLinkOption sets a link option for amqp

func WithSenderLinkOption

func WithSenderLinkOption(opt amqp.LinkOption) Option

WithSenderLinkOption sets a link option for amqp

func WithSessionOpt

func WithSessionOpt(opt amqp.SessionOption) Option

WithSessionOpt sets a session option for amqp

type Protocol

type Protocol struct {

	// AMQP
	Client  *amqp.Client
	Session *amqp.Session
	Node    string

	// Sender
	Sender                  protocol.Sender
	SenderContextDecorators []func(context.Context) context.Context

	// Receiver
	Receiver protocol.Receiver
	// contains filtered or unexported fields
}

func New

func New(server, queue string, opts ...Option) (*Protocol, error)

New creates a new amqp transport.

func (*Protocol) Close

func (t *Protocol) Close() error

func (*Protocol) HasTracePropagation

func (t *Protocol) HasTracePropagation() bool

HasTracePropagation implements Protocol.HasTracePropagation

func (*Protocol) OpenInbound

func (t *Protocol) OpenInbound(ctx context.Context) error

StartReceiver implements Protocol.StartReceiver NOTE: This is a blocking call.

func (*Protocol) Receive

func (t *Protocol) Receive(ctx context.Context) (binding.Message, error)

func (*Protocol) Send

func (t *Protocol) Send(ctx context.Context, in binding.Message) error

type SenderOptionFunc

type SenderOptionFunc func(sender *sender)

amqp.Sender options

func WithTransformer

func WithTransformer(transformer binding.TransformerFactory) SenderOptionFunc

Add a transformer, which Sender uses while encoding a binding.Message to an amqp.Message