amqp

type Transport ¶

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

Transport implements a usrv transport using AMQP. The transport operates in both a client and a server mode and utilizes reference counting to share a single transport instance with multiple clients and servers. The transport maintains and reuses an open connection until both reference counters reach zero. It is therefore important that the application always closes the transport before exiting to ensure that the AMQP logs do not fill up with "client unexpectedly closed TCP connection" wanrings.

In both modes, the transport declares a direct amqp exchange called usrv. All usrv messages are routed through this particular exchange. The transport supoprts endpoint versioning and generates AMQP routing keys with format "$version/$service/$endpoint".

When operating in server mode, the transport allocates a private queue and for each defined endpoint it binds its routing key to the private queue. This allows the transport to use a single consumer channel for processing incoming requests. For each incoming request, the transport uses the routing key to figure out which handler it should invoke and spawns a go-routine to handle the request.

When operating in client mode, the transport allocates a private queue for receiving responses. Whenever the client sends an outgoing request it populates the following AMQP message fields:

• AppId: set to the the outgoing message Sender() value.
• Type: set to the outgoing message SenderEndpoint() value.
• ReplyTo: set to the private queue name for receiving responses.
• CorrelationId: set to the outgoing message ID() value.

Since the transport handles responses asynchronously, the correlation ID serves as a unique ID for matching pending requests to their responses.

The client also listens for failed deliveries. This allows the transport to fail pending requests if no servers are available or if the broker cannot route the request.

The transport receives its broker connection URL from a parameter with name: "transport/amqp/uri". It's default value is set to "amqp://guest:guest@localhost:5672/" which corresponds to a rabbitMQ instance running on localhost. At the moment, TLS connections to the broker are not supported.

As with other usrv transports, once connected, the AMQP transport monitors the URI config for changes and automatically attempts to re-dial the connection whenever its value changes.

func New() *Transport

func (t *Transport) Bind(version, service, endpoint string, handler transport.Handler) error

func (t *Transport) Close(mode transport.Mode) error

func (t *Transport) Dial(mode transport.Mode) error

func (t *Transport) Request(msg transport.Message) <-chan transport.ImmutableMessage

func (t *Transport) Unbind(version, service, endpoint string)