The Architecture

If you’ve ever wondered how messages actually travel from one application to another across the digital landscape, you’re about to get a fascinating peek under the hood. Let’s dive into Rhea’s architecture.

Building Blocks: A Layered Approach

Imagine building a bridge between two cities. You’d need foundations, support structures, road surfaces, and traffic management systems. Rhea’s architecture works similarly, with each layer handling a specific responsibility in the messaging journey.

Rhea Component Architecture

The library divides responsibilities across distinct layers:

┌─────────────────────────────────────┐
│ Application Code                    │
└─────────────┬───────────────────────┘
              ▼
┌─────────────────────────────────────┐
│ Container                           │
└─────────────┬───────────────────────┘
              ▼
┌─────────────────────────────────────┐
│ Connection                          │
└─────────────┬───────────────────────┘
              ▼
┌─────────────┴───────────────────────┐
│ Session                             │
└┬────────────────────────────────────┘
 │
 ├─────────────┐         ┌─────────────┐
 ▼             ▼         ▼             │
┌─────────────────┐    ┌─────────────────┐
│ Sender          │    │ Receiver        │
└─────────────────┘    └─────────────────┘

This separation isn’t just for code organization - it mirrors the AMQP protocol itself, letting each component focus on what it does best. Let’s meet these components and see what makes them tick.

The Key Players: Deep Technical Analysis

Container: Command Central

The Container is where everything begins. It’s the first component you’ll interact with when using Rhea:

const rhea = require("rhea"); // This gives you the container

// Set up global options
rhea.on("connection_open", (context) => {
  console.log(`Connection established to ${context.connection.hostname}`);
});

// Create a connection
const connection = rhea.connect({ host: "message-broker.example.com" });

Technical Deep Dive:

The Container is implemented as a singleton in Rhea, following the module pattern in Node.js. When you require('rhea'), you’re getting a pre-instantiated Container object. The Container extends Node’s EventEmitter and implements a hierarchical event propagation system:

Container.prototype.dispatch = function (name) {
  log.events("[%s] Container got event: " + name, this.id);
  EventEmitter.prototype.emit.apply(this, arguments);
  if (this.listeners(name).length) {
    return true;
  } else {
    return false;
  }
};

This allows events to “bubble up” from lower components (Connections, Sessions, Links) to the Container if they’re not handled at a lower level.

Behind the scenes, the Container maintains internal registries:

A unique container ID (used in AMQP Open frames)
A registry of active connections
SASL mechanism factories for authentication

Connection: The Communications Pipeline

The Connection manages the actual network socket between your application and the message broker. It’s responsible for:

Socket lifecycle (connect, disconnect, reconnect)
Protocol handshake and negotiation
Authentication
Heartbeating to detect connection failures
Channel allocation for sessions

Technical Deep Dive:

In AMQP 1.0, the Connection represents a single TCP/TLS socket connection between two peers. It’s the foundational transport layer for all communication. Rhea’s Connection implementation is complex, handling:

State Management: Each Connection has an EndpointState object that tracks its lifecycle:

this.state = new EndpointState();

This state machine tracks transitions between states like:

Closed (initial state)
Locally Opened (outgoing Open sent)
Remotely Opened (incoming Open received)
Fully Open (both sides Open)
Locally Closed (outgoing Close sent)
Remotely Closed (incoming Close received)
Fully Closed (both sides Close)

Frame Processing: The Connection handles the reading and dispatching of AMQP frames:

Connection.prototype.on_begin = function (frame) {
  var session;
  if (frame.performative.remote_channel === null) {
    // Peer initiated session
    session = this.create_session();
    session.local.begin.remote_channel = frame.channel;
  } else {
    session = this.local_channel_map[frame.performative.remote_channel];
  }
  session.on_begin(frame);
  this.remote_channel_map[frame.channel] = session;
};

Channel Multiplexing: A single AMQP connection can support multiple independent Sessions, each identified by a channel number. The Connection maintains bidirectional mappings:

this.local_channel_map = {}; // Local channel ID → Session
this.remote_channel_map = {}; // Remote channel ID → Session

This multiplexing is a critical performance optimization - it allows many logical communication pathways to share a single TCP connection, reducing overhead.

Heartbeat Management: Connections implement a heartbeat mechanism using empty frames to detect network failures:

// Set up heartbeat timer
if (this.is_open() && this.remote.open.idle_time_out) {
  this.heartbeat_out = setTimeout(
    this._write_frame.bind(this),
    this.remote.open.idle_time_out / 2
  );
}

// Check for received heartbeats
if (this.local.open.idle_time_out) {
  this.heartbeat_in = setTimeout(
    this.idle.bind(this),
    this.local.open.idle_time_out * 2
  );
}

Technical Constraints: A single Connection is restricted by:

The negotiated max_frame_size - limiting individual frame size
The negotiated channel_max - limiting the number of parallel sessions
The idle_time_out - determining heartbeat frequency

Session: The Traffic Manager

Within a Connection, a Session provides an independent channel for communication. Think of it like having multiple phone calls over a single phone line - each session keeps its messages separate.

Technical Deep Dive:

In AMQP, Sessions provide isolation between different message flows over the same Connection. Each Session:

Has its own Sequence Space: Each Session maintains independent sequence numbering for transfers:

this.outgoing = new Outgoing(connection);
this.incoming = new Incoming();
this.outgoing.next_transfer_id = 0;
this.incoming.next_transfer_id = 0;

This allows parallel message streams to operate independently without interfering with each other.

Implements Flow Control: Sessions use a sliding window mechanism for flow control:

// Inside Incoming.prototype.on_flow
this.next_transfer_id = fields.next_outgoing_id;
this.remote_next_transfer_id = fields.next_outgoing_id;
this.remote_window = fields.outgoing_window;

This window-based approach allows efficient batch processing of messages while preventing buffer overflows.

Tracks Message Settlements: Sessions maintain records of message deliveries until they’re fully settled:

// Inside Outgoing
this.deliveries.pop_if(function (d) {
  return d.settled && d.remote_settled;
});

This ensures reliable delivery even during network disruptions.

Manages Link Handles: Each Session assigns and tracks numeric handles for Links:

Session.prototype.create_link = function (name, constructor, opts) {
  var i = 0;
  while (this.local.handles[i]) i++;
  var l = new constructor(this, name, i, opts);
  this.links[name] = l;
  this.local.handles[i] = l;
  return l;
};

Technical Rationale for Multiple Sessions:

AMQP architects designed the protocol to support multiple Sessions per Connection for several compelling reasons:

Fault Isolation: If one Session encounters an error (e.g., a transfer sequence violation), it can be closed without affecting other Sessions on the same Connection.
Parallel Processing: Different application components can use different Sessions, allowing them to process messages independently without coordination.
Quality of Service Differentiation: Critical and non-critical traffic can use separate Sessions with different window sizes.
Transaction Boundaries: Each Session can have its own transactional context.
Independent Sequencing: Each Session maintains its own transfer ID sequence, simplifying message tracking.

In Rhea, the Session implementation maintains internal circular buffers for both incoming and outgoing deliveries:

// Session's circular buffers for tracking deliveries
this.outgoing.deliveries = new CircularBuffer(capacity);
this.incoming.deliveries = new CircularBuffer(capacity);

These buffers efficiently track the state of messages in transit, allowing for performance optimizations like batch acknowledgments.

Links: The Message Endpoints

Links are where the real action happens. They come in two flavors:

Sender - For pushing messages to a destination.
Receiver - For pulling messages from a source.

Technical Deep Dive:

In AMQP, Links are the most granular endpoints for message transfer, and they have a rich set of capabilities:

Source and Target Addressing: Links connect a source to a target, each with addressing information and capabilities:

const sender = connection.open_sender({
  target: {
    address: "orders",
    durable: 2, // Durable (none=0, configuration=1, unsettled-state=2)
    expiry_policy: "never",
    capabilities: ["topic"],
  },
});

This addressing model provides extreme flexibility, supporting different routing patterns (queues, topics, direct) through the same consistent Link API.

Settlement Modes: Links can operate with different reliability guarantees:

// At-most-once delivery (fire and forget)
const fastSender = connection.open_sender({
  snd_settle_mode: 0, // Settled (pre-settled by sender)
  rcv_settle_mode: 0, // First (receiver settles first)
});

// At-least-once delivery (acknowledged)
const reliableSender = connection.open_sender({
  snd_settle_mode: 1, // Unsettled (sender doesn't pre-settle)
  rcv_settle_mode: 0, // First (receiver settles first)
});

// Exactly-once delivery (transactional)
const transactionalSender = connection.open_sender({
  snd_settle_mode: 1, // Unsettled
  rcv_settle_mode: 1, // Second (sender settles after receiver)
});

These settlement modes give applications precise control over delivery guarantees, trading off between reliability and performance.

Credit-Based Flow Control: Links implement a credit-based flow control system:

// In Sender.prototype.sendable
return Boolean(this.credit && this.session.outgoing.available());

// In Receiver.prototype.flow
if (credit > 0) {
  this.credit += credit;
  this.issue_flow = true;
  this.connection._register();
}

Credit represents permission to send messages. A sender can only send if it has both credit (granted by the receiver) and delivery capacity in the session.

Link Recovery: Links can recover after disconnections:

link._reconnect = function () {
  this.state.reconnect();
  this.remote = { handle: undefined };
  this.delivery_count = 0;
  this.credit = 0;
};

This allows applications to maintain logical communication patterns even when the underlying connection is disrupted.

Technical Hierarchy and Scoping Rules:

The AMQP component hierarchy enforces specific scoping rules:

Container scope:
- Global configuration
- Factory for Connections
- Application-wide event handlers
Connection scope:
- Socket management
- Authentication context
- Session creation and tracking
- Channel number allocation
- Frame multiplexing
Session scope:
- Transfer ID sequence space
- Flow control windows
- Message settlement
- Link handle allocation
- Transaction boundaries
Link scope:
- Message sending/receiving
- Credit-based flow control
- Delivery tracking
- Settlement notification
- Source/target addressing

This hierarchy creates natural isolation boundaries. For example:

If a Connection fails, all its Sessions and Links fail
If a Session fails, only its Links fail (other Sessions continue working)
If a Link fails, only that specific communication path is affected

These isolation properties are crucial for building resilient systems.

The Message Journey

Now that we’ve met the players, let’s follow a message as it travels through the system:

Message Flow Through Rhea Components

Sending: From JavaScript to Bytes

Your code calls sender.send()

You provide a plain JavaScript object:

sender.send({
  orderId: "A1234",
  customer: "Acme Corp",
  items: [
    { product: "Anvil", quantity: 1 },
    { product: "Rocket", quantity: 3 },
  ],
});

The Sender prepares the message

The Sender converts your JavaScript object to an AMQP message format:
```
// Inside the sender.send method
return this.session.send(this, tag, message.encode(msg));
```
The Session tracks the delivery

The Session assigns a delivery ID and records the pending message:
```
// Inside session.send
var d = this.outgoing.send(sender, tag, data, format);
```

The Connection creates a transfer frame

The Connection wraps the message in an AMQP transfer frame:

// Inside Connection
this._write_frame(
  session.channel,
  frames.transfer({
    handle: link.handle,
    delivery_id: delivery_id,
    delivery_tag: tag,
    // ...other fields
  }),
  payload
);

The Transport encodes the frame

The Transport layer converts the frame to binary format according to the AMQP spec:
```
// Inside Transport
var buffer = frames.write_frame(frame);
socket.write(buffer);
```
The Socket sends the data

Finally, the raw bytes travel across the network to the message broker.

Receiving: From Bytes to JavaScript

On the receiving side, the process works in reverse:

Data arrives on the Socket

socket.on("data", (buffer) => {
  connection.input(buffer);
});

The Transport reads and decodes frames

// Inside Transport
var frame = frames.read_frame(buffer);

The Connection dispatches to the right Session

// Inside Connection
var session = this.remote_channel_map[frame.channel];
session.on_transfer(frame);

The Session routes to the right Receiver

// Inside Session
this._get_link(frame).on_transfer(frame);

The Receiver assembles the message

// Inside Receiver
var message_data = Buffer.concat(current.frames);
var message = message.decode(message_data);

Your code gets notified

receiver.on("message", (context) => {
  // Your handler gets called with the decoded message
  console.log(context.message);
});

Technical Constraints and Limitations

Each layer in the AMQP architecture imposes specific constraints:

Connection Constraints

Max Frame Size: Negotiated during connection establishment, typically 512K-1MB. Larger messages must be split into multiple frames.
Channel Maximum: Limits the number of concurrent sessions (often 65535).
Idle Timeout: Affects heartbeat frequency. Too short causes unnecessary traffic; too long delays failure detection.

Session Constraints

Incoming Window: Controls how many incoming transfers can be received before acknowledgment.
Outgoing Window: Controls how many outgoing transfers can be sent before acknowledgment.
Handle Maximum: Limits the number of links per session (default: 4,294,967,295).

Link Constraints

Credit Window: Controls the number of messages that can be sent before receiver grants more credit.
Target Capabilities: Determine what messaging patterns the target supports.
Filters: Can restrict which messages are received (e.g., selector expressions).

Performance Optimizations

Rhea employs several clever optimizations to ensure efficient messaging:

Buffer Management

For large messages, Rhea fragments the data into multiple frames:

// In Outgoing.prototype.send
if (max_payload && data.length > max_payload) {
  var start = 0;
  while (start < data.length) {
    var end = Math.min(start + max_payload, data.length);
    fragments.push(data.slice(start, end));
    start = end;
  }
}

This allows it to handle messages of any size without exceeding frame size limitations.

Connection Multiplexing

One of the most powerful optimizations is the ability to run multiple independent sessions over a single network connection:

// Create multiple sessions on one connection
const ordersSession = connection.create_session();
const notificationsSession = connection.create_session();

// Each can have its own links
const orderSender = ordersSession.open_sender("orders");
const alertReceiver = notificationsSession.open_receiver("alerts");

This dramatically reduces the overhead of maintaining multiple TCP connections.

Zero-Copy Reading

When possible, Rhea uses buffer slicing rather than copying:

// Inside the reader
return buffer.slice(current, this.position);

This avoids unnecessary data copying for improved performance with large messages.

Batched Acknowledgments

Rhea can batch disposition updates for multiple deliveries into a single frame:

function write_dispositions(deliveries) {
  var first, last, next_id, i, delivery;
  // Group consecutive deliveries with same state
  for (i = 0; i < deliveries.length; i++) {
    delivery = deliveries[i];
    if (first === undefined) {
      first = delivery;
      last = delivery;
      next_id = delivery.id;
    }

    if (
      (first !== last &&
        !message.are_outcomes_equivalent(last.state, delivery.state)) ||
      last.settled !== delivery.settled ||
      next_id !== delivery.id
    ) {
      first.link.session.output(
        frames.disposition({
          role: first.link.is_receiver(),
          first: first.id,
          last: last.id,
          state: first.state,
          settled: first.settled,
        })
      );
      first = delivery;
      last = delivery;
      next_id = delivery.id;
    } else {
      if (last.id !== delivery.id) {
        last = delivery;
      }
      next_id++;
    }
  }
  // Send final batch
  if (first !== undefined && last !== undefined) {
    first.link.session.output(
      frames.disposition({
        role: first.link.is_receiver(),
        first: first.id,
        last: last.id,
        state: first.state,
        settled: first.settled,
      })
    );
  }
}

This optimization significantly reduces the number of frames needed for high-volume messaging.

Advanced Technical Features

Rhea implements several advanced AMQP features that are worth understanding:

Dynamic Link Addressing

Links can use dynamic addressing, where the broker assigns an address:

// Request a dynamic (temporary) address
const tempReceiver = connection.open_receiver({
  source: {
    dynamic: true,
  },
});

// The address is available after attachment
tempReceiver.on("receiver_open", () => {
  console.log("Assigned address:", tempReceiver.source.address);
});

This is useful for request-response patterns and temporary subscriptions.

Delivery State and Outcomes

AMQP provides rich delivery outcomes beyond simple acknowledgment:

// Accept the message (success)
delivery.accept();

// Reject the message (permanent failure)
delivery.reject({ error: "Invalid format" });

// Release the message (temporary failure, can be redelivered)
delivery.release();

// Modify the message (e.g., for dead-letter queues)
delivery.modified({
  delivery_failed: true,
  undeliverable_here: true,
  message_annotations: { "x-retry-count": 3 },
});

These outcomes allow sophisticated message handling patterns.

Message Annotations

AMQP messages can include annotations that don’t affect the business payload:

sender.send({
  body: actualData,
  message_annotations: {
    "x-routing-key": "special-handling",
    "x-priority": 1,
  },
});

Annotations are useful for routing, tracking, and debugging.

Practical Applications of Rhea’s Architecture

Understanding Rhea’s architecture opens up interesting possibilities:

Multiple Parallel Workflows

You can create different sessions for different types of messaging:

// Critical business transactions
const criticalSession = connection.create_session();
criticalSession.open_sender("orders.critical");

// Non-critical logging
const loggingSession = connection.create_session();
loggingSession.open_sender("logs");

If the logging session gets backed up, it won’t affect your critical order processing.

Sophisticated Flow Control

You can implement adaptive credit management:

// Grant more credit when system load is low
const CREDIT_LOW_LOAD = 1000;
const CREDIT_HIGH_LOAD = 10;

function updateCredit() {
  const systemLoad = getSystemLoad();
  const credit = systemLoad > 0.8 ? CREDIT_HIGH_LOAD : CREDIT_LOW_LOAD;
  receiver.flow(credit - receiver.credit);
}

This lets your application adapt to changing conditions.

Custom Message Routing

You can implement sophisticated routing patterns:

// Create a fanout pattern
function fanout(message, destinations) {
  destinations.forEach((dest) => {
    const sender = connection.open_sender(dest);
    sender.send(message);
  });
}

Wrapping Up

Rhea’s layered architecture provides a clean mapping between the AMQP protocol and JavaScript programming models. Each component handles specific responsibilities, working together to provide reliable messaging.

What makes this design particularly elegant is how it balances complexity and usability:

The high-level API is simple and intuitive
The underlying components handle complex protocol details
Each layer can be tested and optimized independently
The event-driven design fits naturally with Node.js

By understanding these components, you can better leverage Rhea’s capabilities to build robust messaging systems that connect different parts of your application ecosystem.

Whether you’re building microservices, event-driven architectures, or integrating with enterprise systems, Rhea’s architecture provides the foundation for reliable, efficient, and scalable messaging in JavaScript.

Reinventing the wheel

Rhea: The Architecture of a JavaScript AMQP Client (pt. 2)

The Architecture

Building Blocks: A Layered Approach

The Key Players: Deep Technical Analysis

Container: Command Central

Connection: The Communications Pipeline

Session: The Traffic Manager

Links: The Message Endpoints

The Message Journey

Sending: From JavaScript to Bytes

Receiving: From Bytes to JavaScript

Technical Constraints and Limitations

Connection Constraints

Session Constraints

Link Constraints

Performance Optimizations

Buffer Management

Connection Multiplexing

Zero-Copy Reading

Batched Acknowledgments

Advanced Technical Features

Dynamic Link Addressing

Delivery State and Outcomes

Message Annotations

Practical Applications of Rhea’s Architecture

Multiple Parallel Workflows

Sophisticated Flow Control

Custom Message Routing

Wrapping Up