Struct Sender

Source

pub struct Sender { /* private fields */ }

Expand description

High-level sender for transmitting messages over the network.

The Sender provides a convenient interface for sending messages through Thubo’s multi-priority transmission pipeline. It handles message serialization, batching, fragmentation, and QoS enforcement automatically.

§Cloning

Sender is cheaply cloneable (uses Arc internally). Multiple clones can send messages concurrently from different tasks.

§Quality of Service

Each sender has configurable QoS settings that control:

Priority: Message importance (8 levels from Control to Background)
Congestion Control: Behavior when buffers are full (Block or Drop)
Express Mode: Whether to batch messages or send immediately

§Examples

// Configure sender for high-priority, real-time messages
sender.qos(
    QoS::default()
        .with_priority(Priority::High)
        .with_congestion_control(CongestionControl::Block)
        .with_express(true),
);
let msg = Bytes::from(vec![1, 2, 3, 4]);
sender.send(&msg).await?;

// Check if the sender is experiencing congestion
if sender.is_congested() {
    println!("Warning: sender is congested");
}

Implementations§

Source §

impl Sender

Source

pub fn qos(&mut self, qos: QoS) -> &mut Self

Configures the Quality of Service settings for this sender.

The QoS settings determine how messages are prioritized, batched, and handled when the pipeline is congested. This affects all subsequent send() calls.

Source

pub fn timeout(&mut self, timeout: Duration) -> &mut Self

Sets the overall deadline for the entire send operation.

This is the maximum total time that send() will attempt to serialize and queue a message. If the message cannot be successfully queued within this time, the operation fails with SendError::Timeout.

The timeout is checked periodically based on the timeout_progress interval. When no progress can be made (e.g., batch pool is empty), the operation waits up to the progress timeout before checking if the overall deadline has been exceeded.

Default: 10 seconds

§Examples

// Fail if message can't be sent within 5 seconds
sender.timeout(Duration::from_secs(5));

// Method chaining
let msg = Bytes::from(vec![42; 42]);
sender.timeout(Duration::from_secs(5)).send(&msg).await?;

Source

pub fn timeout_progress(&mut self, wait: Duration) -> &mut Self

Sets the progress check interval when the send operation stalls.

When the batch pool is temporarily empty (no batches available for serialization), send() will wait up to this duration for a batch to become available. After this timeout, it checks whether the overall timeout has been exceeded. If not, it waits again for up to this duration.

This creates a progress-checking loop:

Try to acquire a batch
If unavailable, wait up to timeout_progress duration
Check if overall timeout exceeded → return error if yes
Otherwise, repeat from step 1

Tuning guidance:

Too high (e.g., 5s): Send operations become unresponsive to the overall timeout
Too low (e.g., 1ms): Causes busy-waiting and unnecessary CPU usage
Recommended: 100-500ms for most applications

Default: 100ms

§Examples


// Check for progress every 100ms
sender.timeout_progress(Duration::from_millis(100));

// Typical configuration
let msg = Bytes::from(vec![42; 42]);
sender
    .timeout(Duration::from_secs(5)) // Give up after 5 seconds
    .timeout_progress(Duration::from_millis(100)) // Check progress every 100ms
    .send(&msg)
    .await?;

Source

pub fn is_congested(&self) -> bool

Checks if the sender’s priority queue is currently congested.

A queue is considered congested when it cannot keep up with incoming messages and has run out of available buffers. When congested:

Messages with CongestionControl::Drop are immediately dropped
Messages with CongestionControl::Block wait (up to the configured timeout)

§Examples

if sender.is_congested() {
    // Switch to a lower priority or handle congestion
    sender.qos(
        QoS::default()
            .with_priority(Priority::Background)
            .with_congestion_control(CongestionControl::Drop),
    );
}

Source

pub fn is_closed(&self) -> bool

Checks if the sender has been closed.

A sender is closed when stop() has been called or the associated SenderTask has terminated. Once closed, all send() calls will fail.

Source

pub async fn send<T>(&mut self, msg: T) -> Result<(), SendError>
where T: AsRef<Bytes>,

Asynchronously sends a message through the transmission pipeline.

The message is serialized, batched, and queued for transmission according to the configured QoS settings. Large messages are automatically fragmented across multiple batches if needed.

§Errors

SendError::Dropped: Message was dropped due to congestion
SendError::Timeout: Timed out waiting for buffer space
SendError::EncodingFailed: Message serialization failed

§Examples

let msg = Bytes::from(vec![1, 2, 3, 4]);
match sender.send(&msg).await {
    Ok(()) => println!("Message sent successfully"),
    Err(SendError::Dropped) => println!("Message dropped (congestion)"),
    Err(SendError::Timeout) => println!("Timeout waiting for buffers"),
    Err(SendError::EncodingFailed) => println!("Serialization failed"),
    Err(SendError::Closed) => println!("Sender task is stopped"),
    Err(SendError::Internal) => println!("Internal error"),
}

§Performance note

Passing a reference (&Bytes) is significantly more efficient than passing by value and can impact performance by up to few million msg/s in case of very small messages (e.g. 8 bytes payload). References avoid cloning reference counters and allow the message to be reused immediately after the call returns.

However, passing ownership is required when using OnDrop or Boomerang wrappers for buffer lifecycle management (e.g., returning buffers to a pool or tracking when transmission completes).

Source