...
The manageable categories are arranged into a tree structure. SystemConfig is at the root and has a single descendent Broker. The Broker itself has children: Port, AuthenticationProvider, VirtualHostNode amongst others. VirtualHostNode has a child VirtualHost. The children of the VirtualHost are categories that directly involved in messaging such as Queue. The diagram below illustrates the category hierarchy but many categories are elided for brevity. The model tree structure is codified in the BrokerModel class.
draw.io Diagram | ||||||||||||||||||
---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
|
...
AMQPConnection_0_8Impl#received ultimately delegates to methods such as AMQPConnection_0_8Impl#receiveConnectionStartOk
- AMQPConnection_0_10Impl#received ultimately delegates to delegate ServerConnectionDelegate#connectionStartOk
- AMQPConnection_1_0Impl#received ultimately delegates to AMQPConnection_1_0Impl#receiveOpen
Producing bytes
AMQP As already said, the transport pulls tasks from the protocol engine. These tasks produce bytes. To do this, the transport calls the pending iterator which provides a stream of tasks that generate bytes for the wire. The transport keeps pulling until the output exceeds the buffer. It then tries to write the buffered bytes to the wire. If it writes more than half to the wire it continues to pull more tasks from the engine. The cycle continues until the transport cannot take more bytes (back pressure at the TCP/IP layer, or the pending iterator yields no more tasks. This arrangement always means that the transport retains control of backlog of bytes to be written to the wire.
The protocol engines' pending iterator are responsible for maintaining fairness within the connection. They do this by maintaining state between invocations. For instance if a connection had sessions A, B, C, all with tasks to producer and on this output cycle, the network stopped accepting bytes after A's tasks, on the next output cycle. B would be considered first, even if A had subsequently had more work. This fairness patten is repeated through each layer of the protocol.
Queues
<todo>
Management
The Broker exposes two management layers:
- AMQP management
- HTTP management
The management layers esentially exposes the Broker model over the management interface. The management layer has little knowledge of the model itself.
AMQP management
AMQP management is defined by the AMQP Management specification which is currently a Working Draft at Oasis. It defines a standard way to identify objects within the Broker, to invoke operations and pass arguments and get results, all over AMQP itself. When interacting with management, you send management messages to a special address ($management).
For Broker-J:
- To manage a virtualhost, you connect with AMQP to the virtualhost as normal, then send management messages to $management.
In AMQP management, you form an AMQP connection to the Broker.
- To manage an individual virtualhost, you connect to the virtualhost itself then
In AMQP management, objects have a name identifying the type of the object. This is defined using an annotation ManagedObject#amqpName.
In AMQP management
HTTP management
HTTP, REST and Web Management
AMQP Management
Terms
Queue model objects provide the messaging queues. There are several specialisations of Queue
- StandardQueue which provides a FIFO behaviour
- PriorityQueue which provides queuing ordered by a message's priority
- LVQQueue which provides a last-value or conflation queue.
Internally queues are implemented as a linked list (QueueEntryList) of nodes (QueueEntry). The linked list is implemented from first principals. It uses a thread safe and lock-less algorithm (it uses compare and swap operations).
Enqueueing
When a message is enqueued (using the AbstractQueue#enqueue() method) it adds the message to the tail of the queue and notifies a subscriber (consumer) about the new message. The connection that owns the consumer is then awoken and events proceed as described above in the Producing Bytes. This is described by Consumer-Queue-Interactions
Subscriptions
Each subscription keeps a "pointer" (QueueContext#_lastSeenEntry) into the list denoting the point at which that particular subscription has reached. A subscription will only take a message if it is the next AVAILABLE (MessageInstance.State.AVAILABLE) entry.
The diagram below shows point to point queue with three subscribers attached.
Messages
Each queue node QueueEntry refers to a ServerMessage. The server message encapsulates:
- Message meta-data (loosely the message's headers)
- Message payload
- Original routing information,
Many QueueEntries may refer to the same ServerMessage. In the case where a incoming message is routed through an exchange to many queues, the QueueEntry point to the same ServerMessage. This means only one copy of the message exists in the Broker, regardless of however many queues refer to it. This is important for topics where the same message may be sent to hundreds of subscribers.
ServerMessage uses a Reference counting system to control its lifecycle. When the reference reaches zero, it knows no one references it and it can safely delete itself.
The ServerMessage refers to StoredMessage. The StoredMessage the backs the underlying message storage. It provides methods that get the content and the metadata. This might return cached copies, or it might cause store operations to fetch the data from the disk.
StoredMessage can be flowed to disk. The Broker (FlowToDiskCheckingTask) responds to memory pressure by flowing messages that are in-memory only (i.e. transient messages) to disk and freeing the cached copies of persistent messages from memory. This approach frees up memory for messages.
Message and Configuration Store
Messages are written to the MessageStore and configuration to the DurableConfigurationStore. It is possible to back these with the same underlying provider or use a different provider for configuration and messages.
There are several store provider implementations:
- JSON - Configuration Store only
- Berkeley BDB JE - Durable Configuration and/or Message Store
- Derby - Durable Configuration and/or Message Store
- JDBC - Durable Configuration and/or Message Store
These interfaces are pluggable.
Management
The Broker exposes two management layers:
- AMQP management
- HTTP management
The management layers esentially exposes the Broker model over the management interface. The management layers almost no knowledge of the model itself.
AMQP management
AMQP management is defined by the AMQP Management specification which is currently a Working Draft at Oasis. It defines a standard way to identify objects within the Broker, to invoke operations (CRUD operations or arbitrary operations such as clearQueue) and pass arguments and get results, all over AMQP itself. When interacting with management, you send management messages to a special node ($management) which support a reply-to address.
It also defines a standard way to discover the objects that an device managed by AMQP exposes. One idea here is that is should be possible to produce a generic console which is capable of managing (at least at a basic level) any device managed by AMQP.
For Broker-J:
- To manage a virtualhost, you connect with AMQP to the virtualhost as normal, then send management messages to $management.
- To manage the Broker as a whole, you connect with AMQP to a synthetic virtualhost $management and then send management messages to $management.
ManagementNode provides the implementation.
The model (objects, attributes, operations) are simply exposed verbatim over AMQP Management. In AMQP management, objects have a name identifying the type of the object. This is defined using an annotation ManagedObject#amqpName.
HTTP management
The Broker's model is exposed as a REST API. This allows simple tools such as cURL to be an effective way to both manage and monitor the Broker.
The URI for the REST API is /api/latest or /api/v<version>. Currently the Broker supports only a single version of the API. It is envisages that in future a mapping layer might maintain support for n older versions, thus allowing a smooth upgrade for those migrating to new Broker versions.
The URI is mapped to the object's within the tree. The form of the URI is /api/latest/<category type>/<name1>/<name2>/.../<namen> where the names describe a path to an object starting at the root. It is always possible to refer to objects by ids.
- POST/PUT - create or update
- DELETE - delete an object
- GET - get an object or a collection of objects.
The Broker embeds Jetty to provide the HTTP management interface. HttpManagment configures Jetty according to the configuration provided to by the Port/KeyStore/TrustStore model objects.
The embedded server also provides a Web Management Console. This is written using the Dojo framework. It uses the REST API to interact with the Broker.