# How to Implement Custom Adapters for Queue and State Storage in iii > Learn to implement custom adapters for queue and state storage in iii by implementing async traits, providing factory functions, and registering with the register_adapter! macro for runtime discovery. - Repository: [iii/iii](https://github.com/iii-hq/iii) - Tags: how-to-guide - Published: 2026-05-28 --- **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`](https://github.com/iii-hq/iii/blob/main/engine/src/workers/queue/mod.rs) (lines 49-72) defines the interface for event streaming and function-queue transport: ```rust #[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`](https://github.com/iii-hq/iii/blob/main/engine/src/workers/state/adapters/mod.rs) (lines 15-29) handles key-value persistence: ```rust #[async_trait] pub trait StateAdapter: Send + Sync + 'static { async fn set(&self, scope: &str, key: &str, value: Value) -> anyhow::Result; async fn get(&self, scope: &str, key: &str) -> anyhow::Result>; async fn delete(&self, scope: &str, key: &str) -> anyhow::Result<()>; async fn update(&self, scope: &str, key: &str, ops: Vec) -> anyhow::Result; async fn list(&self, scope: &str, prefix: &str) -> anyhow::Result>; async fn list_groups(&self) -> anyhow::Result>; 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`](https://github.com/iii-hq/iii/blob/main/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: ```rust 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>>>, engine: Arc, } ``` ### Step 2: Implement the QueueAdapter Trait Provide concrete logic for event routing. This example from [`engine/examples/custom_queue_adapter.rs`](https://github.com/iii-hq/iii/blob/main/engine/examples/custom_queue_adapter.rs) demonstrates an in-memory implementation that invokes engine functions directly: ```rust 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: ```rust pub struct LoggingQueueAdapter { inner: Arc, } #[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: ```rust use iii::register_adapter; fn make_inmemory_adapter(engine: Arc, cfg: Option) -> AdapterFuture { Box::pin(async move { let adapter = InMemoryQueueAdapter { subscribers: Arc::new(TokioRwLock::new(HashMap::new())), engine, }; Ok(Arc::new(adapter) as _) }) } register_adapter!( 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`](https://github.com/iii-hq/iii/blob/main/engine/src/workers/state/adapters/kv_store.rs)) for in-process storage and the **BridgeAdapter** ([`engine/src/workers/state/adapters/bridge.rs`](https://github.com/iii-hq/iii/blob/main/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: ```rust 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) -> anyhow::Result { 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 { 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> { let mut conn = self.client.get_async_connection().await?; let full_key = format!("{}:{}", scope, key); let val: Option = 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 ```rust fn make_redis_adapter(_engine: Arc, cfg: Option) -> StateAdapterFuture { Box::pin(async move { let adapter = RedisStateAdapter::new(cfg).await?; Ok(Arc::new(adapter) as Arc) }) } register_adapter!( 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`](https://github.com/iii-hq/iii/blob/main/engine/src/workers/registry.rs)). ### Configuration Structure Reference your custom adapter in [`config.yaml`](https://github.com/iii-hq/iii/blob/main/config.yaml): ```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: ```rust use iii::EngineBuilder; EngineBuilder::new() .register_worker::("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!( 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` or `Arc` 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` 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>>` 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.