Struct QueueManager 

pub struct QueueManager { /* private fields */ }

Queue Manager interface

Uses StreamableHTTP protocol for all operations.

Implementations

impl QueueManager

pub async fn create_queue( &self, queue_name: &str, max_depth: Option<usize>, ack_deadline_secs: Option<u64>, ) -> Result<()>

Create a new queue

Arguments

• queue_name - Name of the queue
• max_depth - Maximum queue depth (optional)
• ack_deadline_secs - ACK deadline in seconds (optional)

pub async fn publish( &self, queue_name: &str, payload: &[u8], priority: Option<u8>, max_retries: Option<u32>, ) -> Result<String>

Publish a message to a queue

Example

client.queue().publish("tasks", b"process-video", Some(9), None).await?;

pub async fn consume( &self, queue_name: &str, consumer_id: &str, ) -> Result<Option<Message>>

Consume a message from a queue

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

Acknowledge a message

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

Negative acknowledge a message (requeue)

pub async fn stats(&self, queue_name: &str) -> Result<QueueStats>

Get queue statistics

pub async fn list(&self) -> Result<Vec<String>>

List all queues

pub async fn delete_queue(&self, queue_name: &str) -> Result<()>

Delete a queue

impl QueueManager

pub fn observe_messages( &self, queue_name: impl Into<String>, consumer_id: impl Into<String>, poll_interval: Duration, ) -> (impl Stream<Item = Message> + 'static, SubscriptionHandle)

Observe messages from a queue reactively

Returns a Stream of messages that can be processed asynchronously. The stream will poll the queue at the specified interval.

Arguments

• queue_name - Name of the queue
• consumer_id - Unique consumer identifier
• poll_interval - How often to poll for new messages

Example

use futures::StreamExt;
use synap_sdk::{SynapClient, SynapConfig};
use std::time::Duration;

let (mut stream, handle) = client.queue()
    .observe_messages("tasks", "worker-1", Duration::from_millis(100));

// Process messages reactively
while let Some(message) = stream.next().await {
    println!("Received: {:?}", message);
    // ACK handled automatically
}

// Stop consuming
handle.unsubscribe();

pub fn process_messages<F, Fut>( &self, queue_name: impl Into<String>, consumer_id: impl Into<String>, poll_interval: Duration, handler: F, ) -> SubscriptionHandle

where F: Fn(Message) -> Fut + Send + 'static, Fut: Future<Output = Result<()>> + Send,

Process messages from a queue with automatic ACK/NACK

Automatically acknowledges successfully processed messages and requeues failed ones.

Example

use synap_sdk::{SynapClient, SynapConfig};
use std::time::Duration;

let handle = client.queue().process_messages(
    "tasks",
    "worker-1",
    Duration::from_millis(100),
    |message| async move {
        // Process the message
        println!("Processing: {:?}", message.id);
        Ok(()) // Success = ACK, Err = NACK
    }
);

// Stop processing
handle.unsubscribe();

Trait Implementations

impl Clone for QueueManager

fn clone(&self) -> QueueManager

Returns a duplicate of the value.

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

Performs copy-assignment from source.