> ## 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.

# StreamAdapter

> Trait for implementing stream adapters in the iii framework

The `StreamAdapter` trait defines the interface for real-time stream storage and synchronization in the iii framework. Adapters handle CRUD operations, event broadcasting, and WebSocket subscriptions.

## Location

```rust theme={null}
use iii::modules::stream::adapters::StreamAdapter;
```

Defined in `src/modules/stream/adapters/mod.rs`

## Trait Definition

```rust theme={null}
#[async_trait]
pub trait StreamAdapter: Send + Sync {
    async fn set(
        &self,
        stream_name: &str,
        group_id: &str,
        item_id: &str,
        data: Value,
    ) -> anyhow::Result<SetResult>;

    async fn get(
        &self,
        stream_name: &str,
        group_id: &str,
        item_id: &str,
    ) -> anyhow::Result<Option<Value>>;

    async fn delete(
        &self,
        stream_name: &str,
        group_id: &str,
        item_id: &str,
    ) -> anyhow::Result<DeleteResult>;

    async fn get_group(
        &self,
        stream_name: &str,
        group_id: &str
    ) -> anyhow::Result<Vec<Value>>;

    async fn list_groups(
        &self,
        stream_name: &str
    ) -> anyhow::Result<Vec<String>>;

    async fn list_all_stream(&self) -> anyhow::Result<Vec<StreamMetadata>>;

    async fn emit_event(
        &self,
        message: StreamWrapperMessage
    ) -> anyhow::Result<()>;

    async fn subscribe(
        &self,
        id: String,
        connection: Arc<dyn StreamConnection>,
    ) -> anyhow::Result<()>;

    async fn unsubscribe(&self, id: String) -> anyhow::Result<()>;

    async fn watch_events(&self) -> anyhow::Result<()>;

    async fn destroy(&self) -> anyhow::Result<()>;

    async fn update(
        &self,
        stream_name: &str,
        group_id: &str,
        item_id: &str,
        ops: Vec<UpdateOp>,
    ) -> anyhow::Result<UpdateResult>;
}
```

## Required Methods

### set

```rust theme={null}
async fn set(
    &self,
    stream_name: &str,
    group_id: &str,
    item_id: &str,
    data: Value,
) -> anyhow::Result<SetResult>
```

Sets or creates an item in the stream and broadcasts a create/update event.

<ParamField path="stream_name" type="&str" required>
  The stream identifier
</ParamField>

<ParamField path="group_id" type="&str" required>
  The group/collection within the stream
</ParamField>

<ParamField path="item_id" type="&str" required>
  The unique item identifier
</ParamField>

<ParamField path="data" type="Value" required>
  The item data as a JSON value
</ParamField>

<ParamField path="return" type="anyhow::Result<SetResult>">
  Result indicating whether item was created or updated
</ParamField>

### get

```rust theme={null}
async fn get(
    &self,
    stream_name: &str,
    group_id: &str,
    item_id: &str,
) -> anyhow::Result<Option<Value>>
```

Retrieves a single item from the stream.

<ParamField path="stream_name" type="&str" required>
  The stream identifier
</ParamField>

<ParamField path="group_id" type="&str" required>
  The group/collection within the stream
</ParamField>

<ParamField path="item_id" type="&str" required>
  The unique item identifier
</ParamField>

<ParamField path="return" type="anyhow::Result<Option<Value>>">
  The item data if found, None otherwise
</ParamField>

### delete

```rust theme={null}
async fn delete(
    &self,
    stream_name: &str,
    group_id: &str,
    item_id: &str,
) -> anyhow::Result<DeleteResult>
```

Deletes an item from the stream and broadcasts a delete event.

<ParamField path="stream_name" type="&str" required>
  The stream identifier
</ParamField>

<ParamField path="group_id" type="&str" required>
  The group/collection within the stream
</ParamField>

<ParamField path="item_id" type="&str" required>
  The unique item identifier
</ParamField>

<ParamField path="return" type="anyhow::Result<DeleteResult>">
  Result indicating success or failure
</ParamField>

### get\_group

```rust theme={null}
async fn get_group(
    &self,
    stream_name: &str,
    group_id: &str
) -> anyhow::Result<Vec<Value>>
```

Retrieves all items in a group.

<ParamField path="stream_name" type="&str" required>
  The stream identifier
</ParamField>

<ParamField path="group_id" type="&str" required>
  The group/collection within the stream
</ParamField>

<ParamField path="return" type="anyhow::Result<Vec<Value>>">
  Vector of all items in the group
</ParamField>

### list\_groups

```rust theme={null}
async fn list_groups(
    &self,
    stream_name: &str
) -> anyhow::Result<Vec<String>>
```

Lists all group IDs within a stream.

<ParamField path="stream_name" type="&str" required>
  The stream identifier
</ParamField>

<ParamField path="return" type="anyhow::Result<Vec<String>>">
  Vector of group identifiers
</ParamField>

### list\_all\_stream

```rust theme={null}
async fn list_all_stream(&self) -> anyhow::Result<Vec<StreamMetadata>>
```

Lists all available streams with their metadata.

<ParamField path="return" type="anyhow::Result<Vec<StreamMetadata>>">
  Vector of stream metadata objects
</ParamField>

### emit\_event

```rust theme={null}
async fn emit_event(
    &self,
    message: StreamWrapperMessage
) -> anyhow::Result<()>
```

Broadcasts an event to all subscribed connections.

<ParamField path="message" type="StreamWrapperMessage" required>
  The event message to broadcast
</ParamField>

<ParamField path="return" type="anyhow::Result<()>">
  Returns Ok(()) on success or an error
</ParamField>

### subscribe

```rust theme={null}
async fn subscribe(
    &self,
    id: String,
    connection: Arc<dyn StreamConnection>,
) -> anyhow::Result<()>
```

Registers a WebSocket connection for stream events.

<ParamField path="id" type="String" required>
  Unique connection identifier
</ParamField>

<ParamField path="connection" type="Arc<dyn StreamConnection>" required>
  The WebSocket connection handler
</ParamField>

<ParamField path="return" type="anyhow::Result<()>">
  Returns Ok(()) on success or an error
</ParamField>

### unsubscribe

```rust theme={null}
async fn unsubscribe(&self, id: String) -> anyhow::Result<()>
```

Removes a WebSocket connection subscription.

<ParamField path="id" type="String" required>
  Connection identifier to remove
</ParamField>

<ParamField path="return" type="anyhow::Result<()>">
  Returns Ok(()) on success or an error
</ParamField>

### watch\_events

```rust theme={null}
async fn watch_events(&self) -> anyhow::Result<()>
```

Starts watching for stream events (typically from Redis pub/sub).

<ParamField path="return" type="anyhow::Result<()>">
  Returns Ok(()) on success or an error
</ParamField>

### destroy

```rust theme={null}
async fn destroy(&self) -> anyhow::Result<()>
```

Cleans up adapter resources during shutdown.

<ParamField path="return" type="anyhow::Result<()>">
  Returns Ok(()) on success or an error
</ParamField>

### update

```rust theme={null}
async fn update(
    &self,
    stream_name: &str,
    group_id: &str,
    item_id: &str,
    ops: Vec<UpdateOp>,
) -> anyhow::Result<UpdateResult>
```

Atomically updates an item using JSON operations.

<ParamField path="stream_name" type="&str" required>
  The stream identifier
</ParamField>

<ParamField path="group_id" type="&str" required>
  The group/collection within the stream
</ParamField>

<ParamField path="item_id" type="&str" required>
  The unique item identifier
</ParamField>

<ParamField path="ops" type="Vec<UpdateOp>" required>
  Vector of atomic update operations to apply
</ParamField>

<ParamField path="return" type="anyhow::Result<UpdateResult>">
  The updated value or an error
</ParamField>

## Implementation Example

```rust theme={null}
use std::{collections::HashMap, sync::Arc};
use async_trait::async_trait;
use redis::{Client, AsyncCommands, aio::ConnectionManager};
use tokio::sync::{Mutex, RwLock};
use serde_json::Value;
use iii::{
    modules::stream::{
        adapters::{StreamAdapter, StreamConnection},
        StreamMetadata, StreamWrapperMessage,
    },
};
use iii_sdk::{
    UpdateOp, UpdateResult,
    types::{DeleteResult, SetResult},
};

pub struct RedisAdapter {
    publisher: Arc<Mutex<ConnectionManager>>,
    subscriber: Arc<Client>,
    connections: Arc<RwLock<HashMap<String, Arc<dyn StreamConnection>>>>,
}

impl RedisAdapter {
    pub async fn new(redis_url: String) -> 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),
            connections: Arc::new(RwLock::new(HashMap::new())),
        })
    }
}

#[async_trait]
impl StreamAdapter for RedisAdapter {
    async fn set(
        &self,
        stream_name: &str,
        group_id: &str,
        item_id: &str,
        data: Value,
    ) -> anyhow::Result<SetResult> {
        let mut conn = self.publisher.lock().await;
        let key = format!("stream:{}:{}", stream_name, group_id);
        
        let exists: bool = conn.hexists(&key, item_id).await?;
        conn.hset(&key, item_id, data.to_string()).await?;
        
        Ok(if exists { SetResult::Updated } else { SetResult::Created })
    }

    async fn get(
        &self,
        stream_name: &str,
        group_id: &str,
        item_id: &str,
    ) -> anyhow::Result<Option<Value>> {
        let mut conn = self.publisher.lock().await;
        let key = format!("stream:{}:{}", stream_name, group_id);
        
        let result: Option<String> = conn.hget(&key, item_id).await?;
        Ok(result.and_then(|s| serde_json::from_str(&s).ok()))
    }

    async fn delete(
        &self,
        stream_name: &str,
        group_id: &str,
        item_id: &str,
    ) -> anyhow::Result<DeleteResult> {
        let mut conn = self.publisher.lock().await;
        let key = format!("stream:{}:{}", stream_name, group_id);
        
        let deleted: i32 = conn.hdel(&key, item_id).await?;
        Ok(if deleted > 0 { DeleteResult::Deleted } else { DeleteResult::NotFound })
    }

    async fn get_group(
        &self,
        stream_name: &str,
        group_id: &str,
    ) -> anyhow::Result<Vec<Value>> {
        // Implementation...
        Ok(vec![])
    }

    async fn list_groups(&self, stream_name: &str) -> anyhow::Result<Vec<String>> {
        // Implementation...
        Ok(vec![])
    }

    async fn list_all_stream(&self) -> anyhow::Result<Vec<StreamMetadata>> {
        // Implementation...
        Ok(vec![])
    }

    async fn emit_event(&self, message: StreamWrapperMessage) -> anyhow::Result<()> {
        // Implementation...
        Ok(())
    }

    async fn subscribe(
        &self,
        id: String,
        connection: Arc<dyn StreamConnection>,
    ) -> anyhow::Result<()> {
        let mut connections = self.connections.write().await;
        connections.insert(id, connection);
        Ok(())
    }

    async fn unsubscribe(&self, id: String) -> anyhow::Result<()> {
        let mut connections = self.connections.write().await;
        connections.remove(&id);
        Ok(())
    }

    async fn watch_events(&self) -> anyhow::Result<()> {
        // Implementation...
        Ok(())
    }

    async fn destroy(&self) -> anyhow::Result<()> {
        Ok(())
    }

    async fn update(
        &self,
        stream_name: &str,
        group_id: &str,
        item_id: &str,
        ops: Vec<UpdateOp>,
    ) -> anyhow::Result<UpdateResult> {
        // Implementation with atomic operations...
        Ok(UpdateResult { value: Value::Null })
    }
}
```

## Related Types

### StreamMetadata

```rust theme={null}
pub struct StreamMetadata {
    pub id: String,
    pub groups: Vec<String>,
}
```

Metadata describing a stream and its groups.

### StreamWrapperMessage

```rust theme={null}
pub struct StreamWrapperMessage {
    pub event_type: String,
    pub timestamp: i64,
    pub stream_name: String,
    pub group_id: String,
    pub id: Option<String>,
    pub event: StreamOutboundMessage,
}
```

Wrapper for stream events broadcast to WebSocket connections.

### StreamConnection

```rust theme={null}
#[async_trait]
pub trait StreamConnection: Subscriber + Send + Sync {
    async fn cleanup(&self);
    async fn handle_stream_message(&self, msg: &StreamWrapperMessage) -> anyhow::Result<()>;
}
```

Trait for WebSocket connection handlers that receive stream events.

### UpdateOp

From the `iii_sdk` crate - represents atomic JSON operations:

* Set a field
* Delete a field
* Append to array
* Remove from array
* Increment/decrement numbers

## Registration Example

```rust theme={null}
use iii::modules::stream::StreamCoreModule;

// Register a custom stream adapter
StreamCoreModule::add_adapter("my_stream", |engine, config| async move {
    let adapter = MyStreamAdapter::new(engine, config).await?;
    Ok(Arc::new(adapter) as Arc<dyn StreamAdapter>)
}).await?;
```

## See Also

* [Module](/api/traits/module) - Base module trait
* [QueueAdapter](/api/traits/queue-adapter) - Trait for queue adapters
* [Stream Module](/modules/stream) - Stream module implementation