Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/iii-hq/iii/llms.txt

Use this file to discover all available pages before exploring further.

The QueueAdapter trait defines the interface for queue implementations in the iii framework. Adapters handle message enqueueing, topic subscriptions, and dead-letter queue management.

​
Location

use iii::modules::queue::QueueAdapter;
Defined in src/modules/queue/mod.rs

​
Trait Definition

#[async_trait::async_trait]
pub trait QueueAdapter: Send + Sync + 'static {
    async fn enqueue(
        &self,
        topic: &str,
        data: Value,
        traceparent: Option<String>,
        baggage: Option<String>,
    );
    
    async fn subscribe(
        &self,
        topic: &str,
        id: &str,
        function_id: &str,
        condition_function_id: Option<String>,
        queue_config: Option<SubscriberQueueConfig>,
    );
    
    async fn unsubscribe(&self, topic: &str, id: &str);
    
    async fn redrive_dlq(&self, topic: &str) -> anyhow::Result<u64>;
    
    async fn dlq_count(&self, topic: &str) -> anyhow::Result<u64>;
}

​
Required Methods

​
enqueue

async fn enqueue(
    &self,
    topic: &str,
    data: Value,
    traceparent: Option<String>,
    baggage: Option<String>,
)
Enqueues a message to a topic with optional OpenTelemetry tracing context.
​
topic
&str
required
The topic name to publish the message to
​
data
Value
required
The message payload as a JSON value
​
traceparent
Option<String>
W3C traceparent header for distributed tracing
​
baggage
Option<String>
W3C baggage header for trace context propagation

​
subscribe

async fn subscribe(
    &self,
    topic: &str,
    id: &str,
    function_id: &str,
    condition_function_id: Option<String>,
    queue_config: Option<SubscriberQueueConfig>,
)
Subscribes to a topic with a handler function and optional filtering.
​
topic
&str
required
The topic name to subscribe to
​
id
&str
required
Unique identifier for this subscription
​
function_id
&str
required
The ID of the function to invoke when messages arrive
​
condition_function_id
Option<String>
Optional filter function ID to conditionally process messages
​
queue_config
Option<SubscriberQueueConfig>
Optional queue configuration (retries, concurrency, timeouts, etc.)

​
unsubscribe

async fn unsubscribe(&self, topic: &str, id: &str)
Removes a subscription from a topic.
​
topic
&str
required
The topic name to unsubscribe from
​
id
&str
required
The subscription ID to remove

​
redrive_dlq

async fn redrive_dlq(&self, topic: &str) -> anyhow::Result<u64>
Redrives (retries) all messages from the dead-letter queue for a topic.
​
topic
&str
required
The topic whose dead-letter queue to redrive
​
return
anyhow::Result<u64>
The number of messages redriven, or an error

​
dlq_count

async fn dlq_count(&self, topic: &str) -> anyhow::Result<u64>
Returns the number of messages in the dead-letter queue for a topic.
​
topic
&str
required
The topic whose dead-letter queue to count
​
return
anyhow::Result<u64>
The message count, or an error

​
Implementation Example

use std::sync::Arc;
use async_trait::async_trait;
use redis::{Client, AsyncCommands, aio::ConnectionManager};
use serde_json::Value;
use tokio::sync::{Mutex, RwLock};
use iii::{
    engine::Engine,
    modules::queue::{QueueAdapter, SubscriberQueueConfig},
};

pub struct RedisAdapter {
    publisher: Arc<Mutex<ConnectionManager>>,
    subscriber: Arc<Client>,
    subscriptions: Arc<RwLock<HashMap<String, SubscriptionInfo>>>,
    engine: Arc<Engine>,
}

impl RedisAdapter {
    pub async fn new(redis_url: String, engine: Arc<Engine>) -> anyhow::Result<Self> {
        let client = Client::open(redis_url.as_str())?;
        let manager = client.get_connection_manager().await?;
        
        Ok(Self {
            publisher: Arc::new(Mutex::new(manager)),
            subscriber: Arc::new(client),
            subscriptions: Arc::new(RwLock::new(HashMap::new())),
            engine,
        })
    }
}

#[async_trait]
impl QueueAdapter for RedisAdapter {
    async fn enqueue(
        &self,
        topic: &str,
        data: Value,
        traceparent: Option<String>,
        baggage: Option<String>,
    ) {
        let topic = topic.to_string();
        let publisher = Arc::clone(&self.publisher);
        
        tokio::spawn(async move {
            let mut conn = publisher.lock().await;
            let message = serde_json::json!({
                "data": data,
                "traceparent": traceparent,
                "baggage": baggage,
            });
            let _: Result<(), _> = conn.publish(topic, message.to_string()).await;
        });
    }

    async fn subscribe(
        &self,
        topic: &str,
        id: &str,
        function_id: &str,
        condition_function_id: Option<String>,
        queue_config: Option<SubscriberQueueConfig>,
    ) {
        // Implementation details...
    }

    async fn unsubscribe(&self, topic: &str, id: &str) {
        // Implementation details...
    }

    async fn redrive_dlq(&self, topic: &str) -> anyhow::Result<u64> {
        // Implementation details...
        Ok(0)
    }

    async fn dlq_count(&self, topic: &str) -> anyhow::Result<u64> {
        // Implementation details...
        Ok(0)
    }
}

​
SubscriberQueueConfig

pub struct SubscriberQueueConfig {
    pub queue_mode: Option<String>,
    pub max_retries: Option<u32>,
    pub concurrency: Option<u32>,
    pub visibility_timeout: Option<u64>,
    pub delay_seconds: Option<u64>,
    pub backoff_type: Option<String>,
    pub backoff_delay_ms: Option<u64>,
}
Configuration for queue subscriber behavior:
  • queue_mode - Queue type (e.g., “fifo”, “standard”)
  • max_retries - Maximum retry attempts before moving to DLQ
  • concurrency - Number of concurrent message processors
  • visibility_timeout - Time (ms) before message becomes visible again
  • delay_seconds - Initial delay before processing
  • backoff_type - Retry backoff strategy (e.g., “exponential”)
  • backoff_delay_ms - Base delay between retries

​
Registration Example

use iii::modules::queue::QueueCoreModule;

// Register a custom queue adapter
QueueCoreModule::add_adapter("my_queue", |engine, config| async move {
    let adapter = MyQueueAdapter::new(engine, config).await?;
    Ok(Arc::new(adapter) as Arc<dyn QueueAdapter>)
}).await?;

​
See Also

Build docs developers (and LLMs) love

Get started for freeTalk to us