# How the iii-stream Worker Integrates with Redis for Real-Time Streaming Patterns

> Discover how the iii-stream worker integrates with Redis to manage real-time streaming patterns. Learn about durable storage and low-latency event distribution.

- Repository: [iii/iii](https://github.com/iii-hq/iii)
- Tags: how-to-guide
- Published: 2026-05-28

---

**The iii-stream worker uses a pluggable Redis adapter that persists stream items as Redis hashes while distributing real-time events through Redis Pub/Sub channels, enabling durable storage and low-latency fan-out to WebSocket clients.**

The **iii-stream** worker is the built-in streaming primitive of the iii engine (iii-hq/iii). By configuring the worker to use the Redis adapter, you gain ACID-compliant persistence alongside a pub/sub backbone that powers live data streaming to connected clients.

## Selecting the Redis Adapter

The stream worker loads adapters dynamically based on the `adapter.name` field in `StreamModuleConfig`. When set to `"redis"`, the engine invokes the factory registered in [`engine/src/workers/stream/adapters/redis_adapter.rs`](https://github.com/iii-hq/iii/blob/main/engine/src/workers/stream/adapters/redis_adapter.rs):

```rust
crate::register_adapter!(<StreamAdapterRegistration> name: "redis", make_adapter);

```

The `make_adapter` function constructs a `RedisAdapter` instance by reading the optional `redis_url` from configuration (defaulting to `redis://localhost:6379`):

```rust
fn make_adapter(_engine: Arc<Engine>, config: Option<Value>) -> StreamAdapterFuture {
    Box::pin(async move {
        let redis_url = config
            .as_ref()
            .and_then(|c| c.get("redis_url"))
            .and_then(|v| v.as_str())
            .unwrap_or("redis://localhost:6379")
            .to_string();
        Ok(Arc::new(RedisAdapter::new(redis_url).await?) as Arc<dyn StreamAdapter>)
    })
}

```

## Connection Architecture

The `RedisAdapter` maintains separate Redis clients for publishing and subscribing. In [`engine/src/workers/stream/adapters/redis_adapter.rs`](https://github.com/iii-hq/iii/blob/main/engine/src/workers/stream/adapters/redis_adapter.rs), the `new` constructor establishes:

- A **connection manager** for publishing operations (wrapped in a `Mutex` for thread-safe access)
- A dedicated **async client** for Pub/Sub subscriptions

```rust
let client = Client::open(redis_url.as_str())?;
let manager = timeout(DEFAULT_REDIS_CONNECTION_TIMEOUT, client.get_connection_manager())
    .await?
    .map_err(|e| anyhow::anyhow!("Failed to connect to Redis at {}: {}", redis_url, e))?;
let publisher = Arc::new(Mutex::new(manager));
let subscriber = Arc::new(client);

```

This dual-client design isolates command throughput from event listening, preventing subscription latency from blocking storage writes.

## Persistent Storage with Redis Hashes

All stream operations map to Redis hash commands keyed by `stream:<stream_name>:<group_id>`. The adapter implements atomic operations using Lua scripts to ensure ACID semantics without client-side locks.

### Atomic Set and Delete

The `set` method uses a Lua script to retrieve the old value before writing the new JSON payload:

```rust
let script = redis::Script::new(
    r#"
        local old_value = redis.call('HGET', KEYS[1], ARGV[1])
        redis.call('HSET', KEYS[1], ARGV[1], ARGV[2])
        return old_value
    "#,
);

```

Similarly, `delete` atomically returns the removed value before deleting the hash field, ensuring clients receive the previous state even under concurrent load.

### Partial Updates with JSON Operations

The `update` method applies JSON Patch operations in-place using a pre-defined Lua script (`JSON_UPDATE_SCRIPT`). This script parses a list of `UpdateOp` structs, applies them to the stored JSON, and returns both the previous and current values without requiring a full read-modify-write cycle from the client.

### Group Queries

For listing streams and groups, the adapter uses non-blocking `HSCAN` and `SCAN` commands rather than `KEYS`. All Redis commands lock the `publisher` mutex only for the duration of network I/O, maximizing concurrency across multiple stream workers.

## Real-Time Event Distribution

Beyond persistence, the Redis adapter powers the real-time streaming pattern through Pub/Sub.

### Publishing Events

Every mutation (`set`, `delete`, `update`) calls `emit_event`, which serializes a `StreamWrapperMessage` and publishes it to the `stream::events` channel:

```rust
let event_json = serde_json::to_string(&message)?;
conn.publish::<_, _, ()>(&STREAM_TOPIC, &event_json).await?;

```

### Fan-Out to WebSocket Clients

The adapter runs a background `watch_events` task that subscribes to the same channel, deserializes messages once, and broadcasts them to all active WebSocket connections tracked in a `RwLock<HashMap<String, Arc<dyn StreamConnection>>>`:

```rust
let mut pubsub = self.subscriber.get_async_pubsub().await?;
pubsub.subscribe(&STREAM_TOPIC).await?;
while let Some(msg) = msg.next().await {
    let msg: StreamWrapperMessage = serde_json::from_str(&payload)?;
    for connection in connections.values() {
        connection.handle_stream_message(&msg).await?;
    }
}

```

The `StreamWorker` spawns this watcher at boot time in [`engine/src/workers/stream/stream.rs`](https://github.com/iii-hq/iii/blob/main/engine/src/workers/stream/stream.rs):

```rust
let watch_handle = tokio::spawn(async move { adapter.watch_events().await });

```

When a client connects via WebSocket to the worker's HTTP server, the adapter registers the connection in the shared map, ensuring subsequent Redis events reach that socket immediately.

## Configuration and Usage Examples

### Enabling Redis in Engine Configuration

Configure the worker in [`engine/config.yaml`](https://github.com/iii-hq/iii/blob/main/engine/config.yaml) to use the Redis adapter:

```yaml
workers:
  - id: iii-stream
    config:
      port: 3112
      host: "0.0.0.0"
      adapter:
        name: "redis"
        config:
          redis_url: "redis://redis.example.com:6379"

```

Setting `adapter.name` to `"redis"` triggers the factory that instantiates `RedisAdapter` with the provided URL.

### Writing Stream Items via SDK

Using the TypeScript SDK to persist and update data:

```typescript
import { iii } from "iii-sdk";

await iii.callFunction({
  function_id: "stream::set",
  payload: {
    stream_name: "chat",
    group_id: "room-42",
    item_id: "msg-123",
    data: { author: "alice", text: "Hello world!" }
  }
});

await iii.callFunction({
  function_id: "stream::update",
  payload: {
    stream_name: "chat",
    group_id: "room-42",
    item_id: "msg-123",
    ops: [{ op: "append", path: "reactions", value: "thumbs_up" }]
  }
});

```

### Subscribing to Live Updates

Connect via WebSocket to receive real-time events broadcast by the Redis watcher:

```typescript
const socket = new WebSocket("ws://localhost:3112/");

socket.addEventListener("message", (ev) => {
  const msg = JSON.parse(ev.data);
  if (msg.type === "stream") {
    console.log("Stream event:", msg);
  }
});

```

When any mutation occurs, the Redis adapter publishes the change to the `stream::events` channel, the worker's `watch_events` task forwards it, and connected clients receive the payload immediately.

## Summary

- The **iii-stream worker** integrates with Redis through a configurable adapter system located in [`engine/src/workers/stream/adapters/redis_adapter.rs`](https://github.com/iii-hq/iii/blob/main/engine/src/workers/stream/adapters/redis_adapter.rs).
- **Persistent storage** uses Redis hashes keyed by `stream:<stream_name>:<group_id>`, with Lua scripts ensuring atomic set, delete, and JSON update operations.
- **Real-time distribution** relies on Redis Pub/Sub via the `stream::events` channel, with the adapter acting as both publisher (on writes) and subscriber (for WebSocket fan-out).
- The adapter maintains **dual Redis clients**—one connection manager for commands and one async client for Pub/Sub—to optimize throughput and latency.
- Configuration requires setting `adapter.name` to `"redis"` and optionally specifying a `redis_url` in the worker configuration.

## Frequently Asked Questions

### What Redis data structures does the iii-stream worker use?

The worker uses **Redis hashes** to store stream items, where each hash key follows the pattern `stream:<stream_name>:<group_id>` and fields represent individual item IDs. For real-time notifications, it uses **Redis Pub/Sub** with a single channel named `stream::events` to broadcast changes to all connected WebSocket clients.

### How does the iii-stream worker handle concurrent writes to the same stream item?

The `RedisAdapter` uses **Lua scripts** for all write operations (set, delete, update). These scripts execute atomically on the Redis server, ensuring that read-modify-write cycles are race-condition free without requiring client-side locks or transactions.

### Can I use a Redis Cluster with the iii-stream worker?

The current implementation in [`redis_adapter.rs`](https://github.com/iii-hq/iii/blob/main/redis_adapter.rs) uses `redis::Client::open` with a single connection URL. For Redis Cluster support, you would need to modify the `make_adapter` factory to initialize a `redis::cluster::ClusterClient` instead of the standard client, though this is not implemented in the current iii-hq/iii codebase.

### What is the default Redis URL if not specified in configuration?

If the `redis_url` key is omitted from the adapter configuration, the factory defaults to `redis://localhost:6379`. This is hardcoded in the `make_adapter` function within [`engine/src/workers/stream/adapters/redis_adapter.rs`](https://github.com/iii-hq/iii/blob/main/engine/src/workers/stream/adapters/redis_adapter.rs).