Struct Connection 

pub struct Connection { /* private fields */ }

High-level connection object that manages a single TCP/STOMP connection.

The Connection spawns a background task that maintains the TCP transport, sends/receives STOMP frames using StompCodec, negotiates heartbeats, and performs simple reconnect logic with exponential backoff.

Implementations

impl Connection

pub const NO_HEARTBEAT: &'static str = "0,0"

Heartbeat value that disables heartbeats entirely.

Use this when you don’t want the client or server to send heartbeats. Note that some brokers may still require heartbeats for long-lived connections.

Example

let conn = Connection::connect(
    "localhost:61613",
    "guest",
    "guest",
    Connection::NO_HEARTBEAT,
).await?;

pub const DEFAULT_HEARTBEAT: &'static str = "10000,10000"

Default heartbeat value: 10 seconds for both send and receive.

This is a reasonable default for most applications. The actual heartbeat interval will be negotiated with the server (taking the maximum of client and server preferences).

Example

let conn = Connection::connect(
    "localhost:61613",
    "guest",
    "guest",
    Connection::DEFAULT_HEARTBEAT,
).await?;

pub const DEFAULT_DISCONNECT_TIMEOUT: Duration

How long close waits for the broker to confirm the DISCONNECT before giving up and tearing the socket down anyway.

Override per connection with ConnectOptions::disconnect_timeout.

pub async fn connect( addr: &str, login: &str, passcode: &str, client_hb: &str, ) -> Result<Self, ConnError>

Establish a connection to the STOMP server at addr with the given credentials and heartbeat header string (e.g. “10000,10000”).

This is a convenience wrapper around connect_with_options() that uses default options (STOMP 1.2, host=“/”, no client-id).

If the broker is unreachable, this method retries with exponential backoff (1s → 2s → 4s → … → 30s cap). Authentication errors (ConnError::ServerRejected) fail immediately. See connect_with_options for full details.

Parameters

• addr: TCP address (host:port) of the STOMP server.
• login: login username for STOMP CONNECT.
• passcode: passcode for STOMP CONNECT.
• client_hb: client’s heart-beat header value (“cx,cy” in milliseconds) that will be sent in the CONNECT frame.

Returns a Connection which provides send, send_frame, next_frame, and close helpers. The detailed connection handling (I/O, heartbeats, reconnects) runs on a background task spawned by this method.

pub async fn connect_with_options( addr: &str, login: &str, passcode: &str, client_hb: &str, options: ConnectOptions, ) -> Result<Self, ConnError>

Establish a connection to the STOMP server with custom options.

Use this method when you need to set a custom client-id (for durable subscriptions), specify a virtual host, negotiate different STOMP versions, or add custom CONNECT headers.

Parameters

• addr: TCP address (host:port) of the STOMP server.
• login: login username for STOMP CONNECT.
• passcode: passcode for STOMP CONNECT.
• client_hb: client’s heart-beat header value (“cx,cy” in milliseconds) that will be sent in the CONNECT frame.
• options: custom connection options (version, host, client-id, etc.).

Connection Behavior

If the broker is unreachable, the method retries with exponential backoff (1s → 2s → 4s → … → 30s cap) — the same strategy used for reconnection after a connection drop. This means your application can start before the broker is available and will connect once it comes up.

This retry is unbounded by default. Set ConnectOptions::connect_timeout to cap it — useful for CLI tools and one-shot scripts that must not hang forever on a misconfigured address.

Errors

Returns an error immediately (no retry) if:

• The server rejects the connection, e.g., due to invalid credentials (ConnError::ServerRejected)

All other errors (TCP refused, connection closed mid-handshake, I/O failures) are retried with backoff.

Example

use iridium_stomp::{Connection, ConnectOptions};

// Connect with a client-id for durable subscriptions
let options = ConnectOptions::default()
    .client_id("my-app-instance-1");

let conn = Connection::connect_with_options(
    "localhost:61613",
    "guest",
    "guest",
    Connection::DEFAULT_HEARTBEAT,
    options,
).await?;

pub async fn send( &self, destination: &str, body: impl AsRef<str>, ) -> Result<(), ConnError>

Send a text message to a destination.

This is a convenience wrapper around send_frame for the common case of sending a string payload with no extra headers.

Example

conn.send("/queue/test", "hello").await?;

pub async fn send_frame(&self, frame: Frame) -> Result<(), ConnError>

Send an arbitrary STOMP frame to the broker.

Use this when you need full control over the frame (custom headers, binary body, receipt requests, etc.). For simple text messages, prefer send.

pub async fn send_frame_with_receipt( &self, frame: Frame, ) -> Result<ReceiptHandle, ConnError>

Send a frame with a receipt request and return a handle to its confirmation.

This method adds a unique receipt header to the frame, registers the receipt id for tracking, and returns a ReceiptHandle. Call ReceiptHandle::wait to await the broker’s RECEIPT response.

Any receipt header already present on frame is ignored in favour of the generated id; use ReceiptHandle::receipt_id to read it back.

The returned handle owns the confirmation channel from the moment the frame is queued, so the broker’s response cannot arrive before there is somewhere to put it. Awaiting may be deferred freely - see ReceiptHandle for sending several frames before awaiting any.

Parameters

• frame: the frame to send. A receipt header will be added.

Returns

A ReceiptHandle for the sent frame.

Example

let handle = conn.send_frame_with_receipt(frame).await?;
handle.wait(Duration::from_secs(5)).await?;

pub async fn send_frame_confirmed( &self, frame: Frame, timeout: Duration, ) -> Result<(), ConnError>

Send a frame and wait for server confirmation via RECEIPT.

This is a convenience method that combines send_frame_with_receipt and ReceiptHandle::wait. Use this when you want to ensure a frame was processed by the server before continuing.

Any receipt header already present on frame is ignored in favour of a generated id. Reach for send_frame_with_receipt directly when you need to send several frames before awaiting any of them.

Parameters

• frame: the frame to send.
• timeout: maximum time to wait for the receipt.

Returns

Ok(()) if the frame was sent and receipt confirmed, or an error if sending failed or the receipt timed out.

Example

let frame = Frame::new("SEND")
    .header("destination", "/queue/orders")
    .set_body(b"order data".to_vec());

conn.send_frame_confirmed(frame, Duration::from_secs(5)).await?;
println!("Order sent and confirmed!");

pub async fn subscribe_with_headers( &self, destination: &str, ack: AckMode, extra_headers: Vec<(String, String)>, ) -> Result<Subscription, ConnError>

Subscribe to a destination.

Parameters

• destination: the STOMP destination to subscribe to (e.g. “/queue/foo”).
• ack: acknowledgement mode to request from the server.

Returns a tuple (subscription_id, receiver) where subscription_id is the opaque id assigned locally for this subscription and receiver is a mpsc::Receiver<Frame> which will yield incoming MESSAGE frames for the destination. The caller should read from the receiver to handle messages. Subscribe to a destination using optional extra headers.

This variant accepts additional headers which are stored locally and re-sent on reconnect. Use subscribe as a convenience wrapper when no extra headers are needed.

pub async fn subscribe( &self, destination: &str, ack: AckMode, ) -> Result<Subscription, ConnError>

Convenience wrapper without extra headers.

pub async fn subscribe_with_options( &self, destination: &str, ack: AckMode, options: SubscriptionOptions, ) -> Result<Subscription, ConnError>

Subscribe with a typed SubscriptionOptions structure.

SubscriptionOptions.headers are forwarded to the broker and persisted for automatic resubscribe after reconnect. Durable subscriptions are requested through those headers - see the module docs for SubscriptionOptions.

pub async fn unsubscribe(&self, subscription_id: &str) -> Result<(), ConnError>

Unsubscribe a previously created subscription by its local subscription id.

pub async fn ack( &self, subscription_id: &str, message_id: &str, ) -> Result<(), ConnError>

Acknowledge a message previously received in client or client-individual ack modes.

STOMP ack semantics:

• auto: server considers message delivered immediately; the client should not ack.
• client: cumulative acknowledgements. ACKing message M for subscription S acknowledges all messages delivered to S up to and including M.
• client-individual: only the named message is acknowledged.

Parameters

• subscription_id: the local subscription id returned by Connection::subscribe. This disambiguates which subscription’s pending queue to advance for cumulative ACKs.
• message_id: the message-id header value from the received MESSAGE frame to acknowledge.

Behavior

• The pending queue for subscription_id is searched for message_id. If the subscription used client ack mode, all pending messages up to and including the matched message are removed. If the subscription used client-individual, only the matched message is removed.
• An ACK frame is sent to the server with id=<message_id> and subscription=<subscription_id> headers.

pub async fn nack( &self, subscription_id: &str, message_id: &str, ) -> Result<(), ConnError>

Negative-acknowledge a message (NACK).

Parameters

• subscription_id: the local subscription id the message was delivered under.
• message_id: the message-id header value from the received MESSAGE.

Behavior

• Removes the message from the local pending queue (cumulatively if the subscription used client ack mode, otherwise only the single message). Sends a NACK frame to the server with id and subscription headers.

pub async fn begin(&self, transaction_id: &str) -> Result<(), ConnError>

Begin a transaction.

Parameters

• transaction_id: unique identifier for the transaction. The caller is responsible for ensuring uniqueness within the connection.

Behavior

• Sends a BEGIN frame to the server with transaction:<transaction_id> header. Subsequent SEND, ACK, and NACK frames may include this transaction id to group them into the transaction. The transaction must be finalized with either commit or abort.

pub async fn commit(&self, transaction_id: &str) -> Result<(), ConnError>

Commit a transaction.

Parameters

• transaction_id: the transaction identifier previously passed to begin.

Behavior

• Sends a COMMIT frame to the server with transaction:<transaction_id> header. All operations within the transaction are applied atomically.

pub async fn abort(&self, transaction_id: &str) -> Result<(), ConnError>

Abort a transaction.

Parameters

• transaction_id: the transaction identifier previously passed to begin.

Behavior

• Sends an ABORT frame to the server with transaction:<transaction_id> header. All operations within the transaction are discarded.

pub async fn next_frame(&self) -> Option<ReceivedFrame>

Receive the next frame from the server.

Returns Some(ReceivedFrame::Frame(..)) for normal frames (MESSAGE, etc.), Some(ReceivedFrame::Error(..)) for ERROR frames, or None if the connection has been closed.

Example

use iridium_stomp::ReceivedFrame;

while let Some(received) = conn.next_frame().await {
    match received {
        ReceivedFrame::Frame(frame) => {
            println!("Got {}: {:?}", frame.command, frame.body);
        }
        ReceivedFrame::Error(err) => {
            eprintln!("Server error: {}", err);
            break;
        }
    }
}

pub async fn close(self) -> Result<(), ConnError>

Gracefully shut down the connection.

Performs the STOMP 1.2 shutdown sequence: sends a DISCONNECT frame carrying a receipt header, waits for the broker’s RECEIPT, then stops the background task and closes the socket.

Because frames are written in the order they are submitted, and the broker only answers a DISCONNECT once it has processed what came before, a confirmed close also proves that everything previously sent on this connection reached the broker.

Returns

Ok(()) once the broker has confirmed the DISCONNECT. Err(ConnError::ReceiptTimeout) if it did not answer within the disconnect timeout, or another ConnError if the DISCONNECT could not be submitted or was rejected.

The connection is torn down either way. An error reports that the shutdown was not clean - that the broker may not have run whatever it does on a protocol-level disconnect, such as transactional rollback or durable subscription cleanup - not that the connection is still open. Callers with nothing to do about that may discard the result.

The wait is bounded by ConnectOptions::disconnect_timeout, defaulting to DEFAULT_DISCONNECT_TIMEOUT, so an unresponsive broker cannot make close hang.

Example

// Report an unclean shutdown.
conn.close().await?;

// Or shut down best-effort.
let _ = conn.close().await;

Trait Implementations

impl Clone for Connection

fn clone(&self) -> Connection

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.