How to Implement Custom Adapters for Queue and State Storage in iii

To implement custom adapters for queue and state storage in iii, implement the QueueAdapter or StateAdapter async traits, provide a factory function returning AdapterFuture, and register your adapter using the register_adapter! macro for runtime discovery.

The iii workflow engine decouples its core execution logic from storage backends through a trait-based adapter system. By implementing these asynchronous traits, you can integrate any message broker (Kafka, NATS, RabbitMQ) or state store (Redis, PostgreSQL, DynamoDB) without modifying the engine's internal code. This guide walks through the exact implementation patterns found in the iii-hq/iii repository, referencing specific source files and production-ready examples.

Understanding the Core Adapter Traits

iii defines two primary adapter interfaces in the engine's worker subsystem. Both traits use async_trait and require Send + Sync + 'static bounds to support concurrent execution across worker threads.

QueueAdapter Trait

The QueueAdapter trait in engine/src/workers/queue/mod.rs (lines 49-72) defines the interface for event streaming and function-queue transport:

#[async_trait]
pub trait QueueAdapter: Send + Sync + 'static {
    async fn enqueue(&self, topic: &str, event_data: Value) -> anyhow::Result<()>;
    async fn subscribe(&self, topic: &str, id: &str, function_id: &str) -> anyhow::Result<()>;
    async fn unsubscribe(&self, topic: &str, id: &str) -> anyhow::Result<()>;
    // DLQ handling and topic discovery methods...
}

StateAdapter Trait

The StateAdapter trait in engine/src/workers/state/adapters/mod.rs (lines 15-29) handles key-value persistence:

#[async_trait]
pub trait StateAdapter: Send + Sync + 'static {
    async fn set(&self, scope: &str, key: &str, value: Value) -> anyhow::Result<SetResult>;
    async fn get(&self, scope: &str, key: &str) -> anyhow::Result<Option<Value>>;
    async fn delete(&self, scope: &str, key: &str) -> anyhow::Result<()>;
    async fn update(&self, scope: &str, key: &str, ops: Vec<UpdateOp>) -> anyhow::Result<UpdateResult>;
    async fn list(&self, scope: &str, prefix: &str) -> anyhow::Result<Vec<String>>;
    async fn list_groups(&self) -> anyhow::Result<Vec<String>>;
    async fn destroy(&self) -> anyhow::Result<()>;
}

Implementing a Custom Queue Adapter

To create a custom queue backend, you must implement the trait methods and register the adapter with the engine's inventory system. Reference the BuiltinQueueAdapter in engine/src/workers/queue/adapters/builtin/adapter.rs for the canonical implementation pattern.

Step 1: Define the Adapter Structure

Create a struct that holds your client connection and any necessary state:

use std::sync::Arc;
use tokio::sync::RwLock as TokioRwLock;
use std::collections::HashMap;
use serde_json::Value;
use iii::Engine;

pub struct InMemoryQueueAdapter {
    subscribers: Arc<TokioRwLock<HashMap<String, Vec<(String, String)>>>>,
    engine: Arc<Engine>,
}

Step 2: Implement the QueueAdapter Trait

Provide concrete logic for event routing. This example from engine/examples/custom_queue_adapter.rs demonstrates an in-memory implementation that invokes engine functions directly:

use async_trait::async_trait;

#[async_trait]
impl CustomQueueAdapter for InMemoryQueueAdapter {
    async fn enqueue(&self, topic: &str, event_data: Value) {
        let subs = self.subscribers.read().await;
        if let Some(list) = subs.get(topic) {
            for (_, function_id) in list {
                let _ = self.engine.call(function_id, event_data.clone()).await;
            }
        }
    }
    
    async fn subscribe(&self, topic: &str, id: &str, function_id: &str) {
        self.subscribers.write().await
            .entry(topic.to_string())
            .or_default()
            .push((id.to_string(), function_id.to_string()));
    }
    
    async fn unsubscribe(&self, topic: &str, id: &str) {
        if let Some(list) = self.subscribers.write().await.get_mut(topic) {
            list.retain(|(sub_id, _)| sub_id != id);
        }
    }
}

Step 3: Compose with Logging Wrappers

Adapters compose naturally through Arc-wrapping. This pattern from the same example adds observability without modifying the underlying adapter:

pub struct LoggingQueueAdapter {
    inner: Arc<dyn CustomQueueAdapter>,
}

#[async_trait]
impl CustomQueueAdapter for LoggingQueueAdapter {
    async fn enqueue(&self, topic: &str, data: Value) {
        tracing::info!(topic = %topic, data = %data, "LoggingQueueAdapter enqueue");
        self.inner.enqueue(topic, data).await;
    }
    // subscribe/unsubscribe forward with logging...
}

Step 4: Register the Factory

The registration macro creates a global registry entry keyed by name:

use iii::register_adapter;

fn make_inmemory_adapter(engine: Arc<Engine>, cfg: Option<Value>) -> AdapterFuture<dyn CustomQueueAdapter> {
    Box::pin(async move {
        let adapter = InMemoryQueueAdapter {
            subscribers: Arc::new(TokioRwLock::new(HashMap::new())),
            engine,
        };
        Ok(Arc::new(adapter) as _)
    })
}

register_adapter!(<CustomQueueAdapterRegistration> name: "my::InMemoryQueueAdapter", make_inmemory_adapter);

Implementing a Custom State Adapter

State adapters follow an identical pattern but persist key-value data rather than routing events. The repository provides two reference implementations: the BuiltinKvStore (engine/src/workers/state/adapters/kv_store.rs) for in-process storage and the BridgeAdapter (engine/src/workers/state/adapters/bridge.rs) for remote delegation via WebSocket.

Redis State Adapter Example

This skeleton demonstrates connecting to an external Redis instance:

use async_trait::async_trait;
use iii::StateAdapter;
use iii_sdk::{SetResult, UpdateOp, UpdateResult};
use redis::AsyncCommands;
use serde_json::Value;

pub struct RedisStateAdapter {
    client: redis::Client,
}

impl RedisStateAdapter {
    pub async fn new(config: Option<Value>) -> anyhow::Result<Self> {
        let url = config
            .and_then(|c| c.get("url"))
            .and_then(|v| v.as_str())
            .unwrap_or("redis://127.0.0.1/");
        let client = redis::Client::open(url)?;
        Ok(Self { client })
    }
}

#[async_trait]
impl StateAdapter for RedisStateAdapter {
    async fn set(&self, scope: &str, key: &str, value: Value) -> anyhow::Result<SetResult> {
        let mut conn = self.client.get_async_connection().await?;
        let full_key = format!("{}:{}", scope, key);
        let raw = serde_json::to_string(&value)?;
        redis::cmd("SET")
            .arg(full_key)
            .arg(raw)
            .query_async(&mut conn)
            .await?;
        Ok(SetResult { updated: true })
    }
    
    async fn get(&self, scope: &str, key: &str) -> anyhow::Result<Option<Value>> {
        let mut conn = self.client.get_async_connection().await?;
        let full_key = format!("{}:{}", scope, key);
        let val: Option<String> = redis::cmd("GET")
            .arg(full_key)
            .query_async(&mut conn)
            .await?;
        Ok(val.map(|s| serde_json::from_str(&s)).transpose()?)
    }
    // Implement delete, update, list, list_groups, destroy...
}

Factory and Registration

fn make_redis_adapter(_engine: Arc<Engine>, cfg: Option<Value>) -> StateAdapterFuture {
    Box::pin(async move {
        let adapter = RedisStateAdapter::new(cfg).await?;
        Ok(Arc::new(adapter) as Arc<dyn StateAdapter>)
    })
}

register_adapter!(<StateAdapterRegistration> name: "redis", make_redis_adapter);

Registering and Configuring Adapters

The EngineBuilder resolves adapter factories at runtime using the inventory pattern. When parsing configuration, it looks up the registered name in the global AdapterRegistry (engine/src/workers/registry.rs).

Configuration Structure

Reference your custom adapter in config.yaml:

modules:
  - class: my::RedisStateModule
    config:
      adapter:
        name: redis
        config:
          url: redis://localhost:6379
          
  - class: my::CustomQueueModule
    config:
      adapter:
        name: my::LoggingQueueAdapter
        config:
          inner_adapter: my::InMemoryQueueAdapter

Programmatic Registration

For library users, register workers explicitly:

use iii::EngineBuilder;

EngineBuilder::new()
    .register_worker::<CustomQueueModule>("my::CustomQueueModule")
    .add_worker(
        "my::CustomQueueModule",
        Some(json!({
            "adapter": {
                "name": "my::InMemoryQueueAdapter",
                "config": null
            }
        }))
    )
    .build()
    .await?;

Summary

• Trait Implementation: Implement QueueAdapter for event streaming or StateAdapter for key-value storage, both defined in engine/src/workers/ subdirectories.
• Factory Pattern: Provide a factory function returning Box::pin(async move { ... }) (typed as AdapterFuture or StateAdapterFuture) that constructs your adapter.
• Macro Registration: Use register_adapter!(<RegistrationType> name: "unique_name", factory_fn) to expose the adapter to the engine's inventory system.
• Configuration: Select adapters via YAML configuration using the adapter.name key, passing custom parameters through adapter.config.
• Composition: Wrap existing adapters (like the logging wrapper example) to add cross-cutting concerns without rewriting storage logic.

Frequently Asked Questions

How does iii discover custom adapters at runtime?

The engine uses the inventory crate to collect all adapters registered via the register_adapter! macro into a global registry. When the EngineBuilder encounters an adapter.name in configuration, it queries this registry for the corresponding factory function, which it calls with the engine instance and configuration JSON. This occurs before any workers begin processing, ensuring the adapter is initialized and ready.

Can I use different adapters for different modules in the same engine?

Yes. Each module configuration specifies its own adapter stanza, allowing one module to use a Redis-backed state store while another uses the built-in KV adapter. The engine maintains separate Arc<dyn StateAdapter> or Arc<dyn QueueAdapter> instances per module based on these configurations, ensuring complete isolation between different storage backends.

What error handling pattern should custom adapters follow?

All trait methods return anyhow::Result<T> to allow flexible error propagation without forcing a specific error type. Your adapter should use ? to propagate IO errors, serialization errors, or connection failures. The engine treats any error result as a transient or permanent failure depending on the context (e.g., state operation failures may trigger function retries depending on your workflow configuration).

How do I test a custom adapter without external dependencies?

Follow the pattern used by BuiltinQueueAdapter and BuiltinKvStore: implement an in-memory version using Arc<TokioRwLock<HashMap<...>>> for state and VecDeque or HashMap for queue backends. These implementations live alongside your production adapters in #[cfg(test)] modules or separate test files, allowing you to run #[tokio::test] async tests without requiring Docker containers or external services.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}