Skip to main content

Design

Where ruststream-amqp sits: between the RustStream runtime and the fe2o3-amqp client, a session per subscription and a shared session for publishers.

The service, the RustStream runtime, this crate, the fe2o3-amqp client and the broker, top to bottomYour serviceRustStream coreruststream-amqpfe2o3-amqp: one connection#[subscriber]handlersRustStream::new(..).with_broker(..)Runtime: dispatch, lanes, settlementAmqpSubscriberpump task, channelAmqpPublisher,AmqpTxnPublisherAmqpMessageRequest and replyAmqpBrokerURL, Sasl, containerAmqpAddressqueue, topic, rawA session and a receiverper subscriptionOne session, a senderlink per addressAMQP 1.0 broker
The diagram as text
  1. Your service: #[subscriber] handlers; RustStream::new(..) .with_broker(..)
  2. RustStream core: Runtime: dispatch, lanes, settlement
  3. ruststream-amqp: AmqpSubscriber pump task, channel; AmqpPublisher, AmqpTxnPublisher; AmqpMessage; Request and reply; AmqpBroker URL, Sasl, container; AmqpAddress queue, topic, raw
  4. fe2o3-amqp: one connection: A session and a receiver per subscription; One session, a sender link per address
  5. #[subscriber] handlers → Runtime: dispatch, lanes, settlement: mounted
  6. Runtime: dispatch, lanes, settlement → AmqpSubscriber pump task, channel: polls the stream
  7. AmqpSubscriber pump task, channel → A session and a receiver per subscription
  8. AmqpPublisher, AmqpTxnPublisher → One session, a sender link per address
  9. A session and a receiver per subscription → AMQP 1.0 broker
  10. One session, a sender link per address → AMQP 1.0 broker

The crate implements the core's broker traits over fe2o3-amqp. The runtime above it is the same for every broker. What is particular to AMQP 1.0 is in this crate: the address descriptor, what an ack means, where a retry copy goes and how a transaction posts.

A connected broker holds one connection. Its publishers share one session, and the first publish to an address attaches a sender link that every later publish to that address reuses. The connection owns those links, so a shutdown can close each one and wait for the broker's answer. Each subscription begins a session of its own and attaches one receiver link on it.

Subscriptions and their sessions start on the runtime the broker connected on, whichever thread opens them. A handler on a thread of its own therefore leaves no part of the connection on that thread.

The crate opens the TCP socket itself, with Nagle's algorithm off. A publish and a request each send a small frame and wait for the answer, which is the pattern Nagle's algorithm delays until the peer's acknowledgement.

A message carries no envelope of the framework's own. The well-known headers ride the properties section, the partition key as group-id, and every other header rides application-properties. A message from another AMQP client arrives with its headers, and one this crate sent reads as a plain AMQP message.

The asyncapi feature describes the addresses with x-ruststream-amqp1 objects, because the specification reserves its amqp1 bindings and leaves them empty.

How the pieces start and stop is on Lifecycle, and the way of one message on Path.

The in-process transport​

With the testing feature, TestApp::start connects the same broker to an in-process transport instead of a server. The descriptors, the policies and the deliveries are the production ones, and a publish goes through the same conversion into AMQP sections. It routes by exact address: the queue receivers of an address compete, and each topic subscription gets a copy. A queue subscription creates its queue, which keeps a message until a receiver takes it. It settles in the subscription's mode, counts a modified delivery in delivery-count, returns an unsettled delivery to the queue when its subscription detaches, and answers a request over a private reply address. A transaction publishes nothing before its commit.

What is the server's own is left to a live run: link credit, how a server creates an address a publish reaches first, the dead-letter policy behind a rejection, the server's own delivery limit and durability. In process a message to an address no queue subscription has attached to is dropped, and a raw address is delivered once, where Artemis may fan it out.

Without the feature the transport is not compiled. A production build's connected broker is one shared pointer to the connection, and the compiler checks that size.