> ## Documentation Index
> Fetch the complete documentation index at: https://mintlify.com/iii-hq/sdk/llms.txt
> Use this file to discover all available pages before exploring further.

# Channels

> Streaming data channels for worker-to-worker communication

## Overview

Channels provide WebSocket-backed streaming communication between workers. They enable efficient transfer of large binary data and real-time messaging without going through the function invocation protocol.

## Creating Channels

### create\_channel

Create a bidirectional streaming channel.

```rust theme={null}
pub async fn create_channel(&self, buffer_size: Option<usize>) -> Result<Channel, IIIError>
```

<ParamField path="buffer_size" type="Option<usize>">
  Internal buffer size for the channel (default determined by engine)
</ParamField>

<ResponseField name="Channel" type="Result<Channel, IIIError>">
  A channel with writer, reader, and their serializable references
</ResponseField>

**Example:**

```rust theme={null}
use iii_sdk::III;

let iii = III::new("ws://localhost:49134");
iii.connect().await?;

let channel = iii.create_channel(Some(1000)).await?;

// Channel contains:
// - channel.writer: ChannelWriter
// - channel.reader: ChannelReader
// - channel.writer_ref: StreamChannelRef (serializable)
// - channel.reader_ref: StreamChannelRef (serializable)
```

## Writing to Channels

### ChannelWriter

The writer side of a channel supports writing binary data and text messages.

#### write

Write binary data to the channel.

```rust theme={null}
pub async fn write(&self, data: &[u8]) -> Result<(), IIIError>
```

<ParamField path="data" type="&[u8]" required>
  Binary data to write to the channel
</ParamField>

**Example:**

```rust theme={null}
let channel = iii.create_channel(None).await?;

// Write binary data
let data = vec![1, 2, 3, 4, 5];
channel.writer.write(&data).await?;

// Write larger data (automatically chunked at 64KB)
let large_data = vec![0u8; 1_000_000];
channel.writer.write(&large_data).await?;
```

<Note>
  Data larger than 64KB is automatically chunked into multiple WebSocket frames.
</Note>

#### send\_message

Send a text message to the channel.

```rust theme={null}
pub async fn send_message(&self, msg: &str) -> Result<(), IIIError>
```

<ParamField path="msg" type="&str" required>
  Text message to send
</ParamField>

**Example:**

```rust theme={null}
use serde_json::json;

// Send JSON message
let message = json!({
    "type": "status",
    "progress": 50
});
channel.writer.send_message(&message.to_string()).await?;
```

#### close

Close the writer side of the channel.

```rust theme={null}
pub async fn close(&self) -> Result<(), IIIError>
```

**Example:**

```rust theme={null}
channel.writer.write(&data).await?;
channel.writer.close().await?;
```

## Reading from Channels

### ChannelReader

The reader side of a channel supports reading binary data and receiving text messages.

#### next\_binary

Read the next binary chunk from the channel.

```rust theme={null}
pub async fn next_binary(&self) -> Result<Option<Vec<u8>>, IIIError>
```

<ResponseField name="Option<Vec<u8>>" type="Result<Option<Vec<u8>>, IIIError>">
  The next binary chunk, or `None` when the channel is closed
</ResponseField>

**Example:**

```rust theme={null}
let channel = iii.create_channel(None).await?;

// Read binary data
while let Some(chunk) = channel.reader.next_binary().await? {
    println!("Received {} bytes", chunk.len());
    // Process chunk
}

println!("Channel closed");
```

<Note>
  Text messages are not returned by `next_binary()`. Register a callback with `on_message()` to handle text messages.
</Note>

#### read\_all

Read the entire stream into a single buffer.

```rust theme={null}
pub async fn read_all(&self) -> Result<Vec<u8>, IIIError>
```

<ResponseField name="Vec<u8>" type="Result<Vec<u8>, IIIError>">
  All binary data from the channel concatenated into a single buffer
</ResponseField>

**Example:**

```rust theme={null}
let channel = iii.create_channel(None).await?;

// Read all data at once
let all_data = channel.reader.read_all().await?;
println!("Received {} total bytes", all_data.len());
```

#### on\_message

Register a callback for text messages.

```rust theme={null}
pub async fn on_message<F>(&self, callback: F)
where
    F: Fn(String) + Send + Sync + 'static
```

<ParamField path="callback" type="F" required>
  Callback invoked for each text message received
</ParamField>

**Example:**

```rust theme={null}
use serde_json::Value;

let channel = iii.create_channel(None).await?;

// Register message handler
channel.reader.on_message(|msg| {
    if let Ok(json) = serde_json::from_str::<Value>(&msg) {
        println!("Received message: {:?}", json);
    }
}).await;

// Read binary data (messages are handled by callback)
while let Some(chunk) = channel.reader.next_binary().await? {
    // Process binary data
}
```

#### close

Close the reader side of the channel.

```rust theme={null}
pub async fn close(&self) -> Result<(), IIIError>
```

## Passing Channels to Functions

Channel references are serializable and can be passed as function arguments:

```rust theme={null}
use iii_sdk::III;
use serde_json::json;

let iii = III::new("ws://localhost:49134");
iii.connect().await?;

// Create a channel
let channel = iii.create_channel(None).await?;

// Pass writer to another function
iii.call("data_processor", json!({
    "output": channel.writer_ref
})).await?;

// Read results from the reader
let results = channel.reader.read_all().await?;
```

### Receiving Channel References

In a function handler, extract channel references and create reader/writer instances:

```rust theme={null}
use iii_sdk::{III, ChannelWriter, ChannelReader, extract_channel_refs};
use serde_json::Value;

iii.register_function("process_data", |input: Value| async move {
    // Extract channel references from input
    let refs = extract_channel_refs(&input);
    
    for (path, channel_ref) in refs {
        match channel_ref.direction {
            ChannelDirection::Write => {
                let writer = ChannelWriter::new(
                    "ws://localhost:49134",
                    &channel_ref
                );
                writer.write(b"processed data").await?;
                writer.close().await?;
            }
            ChannelDirection::Read => {
                let reader = ChannelReader::new(
                    "ws://localhost:49134",
                    &channel_ref
                );
                let data = reader.read_all().await?;
                println!("Received {} bytes", data.len());
            }
        }
    }
    
    Ok(Value::Null)
});
```

## Channel Utilities

### is\_channel\_ref

Check if a JSON value is a channel reference.

```rust theme={null}
pub fn is_channel_ref(value: &Value) -> bool
```

**Example:**

```rust theme={null}
use iii_sdk::is_channel_ref;
use serde_json::json;

let value = json!({
    "channel_id": "ch-123",
    "access_key": "key-abc",
    "direction": "write"
});

if is_channel_ref(&value) {
    println!("This is a channel reference");
}
```

### extract\_channel\_refs

Extract all channel references from a JSON value.

```rust theme={null}
pub fn extract_channel_refs(data: &Value) -> Vec<(String, StreamChannelRef)>
```

<ParamField path="data" type="&Value" required>
  JSON value to search for channel references
</ParamField>

<ResponseField name="Vec<(String, StreamChannelRef)>" type="Vec<(String, StreamChannelRef)>">
  List of (field path, channel reference) tuples
</ResponseField>

**Example:**

```rust theme={null}
use iii_sdk::extract_channel_refs;
use serde_json::json;

let input = json!({
    "output": {
        "channel_id": "ch-123",
        "access_key": "key-abc",
        "direction": "write"
    },
    "input": {
        "channel_id": "ch-456",
        "access_key": "key-def",
        "direction": "read"
    }
});

let refs = extract_channel_refs(&input);
for (path, channel_ref) in refs {
    println!("Found channel at: {}", path);
    println!("  Direction: {:?}", channel_ref.direction);
}
```

## Types

### Channel

A bidirectional streaming channel.

```rust theme={null}
pub struct Channel {
    pub writer: ChannelWriter,
    pub reader: ChannelReader,
    pub writer_ref: StreamChannelRef,
    pub reader_ref: StreamChannelRef,
}
```

### StreamChannelRef

Serializable reference to a channel endpoint.

```rust theme={null}
pub struct StreamChannelRef {
    pub channel_id: String,
    pub access_key: String,
    pub direction: ChannelDirection,
}
```

### ChannelDirection

Direction of a channel reference.

```rust theme={null}
pub enum ChannelDirection {
    Read,
    Write,
}
```

## Complete Example

Here's a complete example showing data streaming between two workers:

```rust theme={null}
use iii_sdk::{III, ChannelWriter, ChannelReader, extract_channel_refs};
use serde_json::{json, Value};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let iii = III::new("ws://localhost:49134");
    
    // Worker 1: Data producer
    iii.register_function("generate_data", |input: Value| async move {
        let refs = extract_channel_refs(&input);
        let (_, writer_ref) = &refs[0];
        
        let writer = ChannelWriter::new("ws://localhost:49134", writer_ref);
        
        // Stream data in chunks
        for i in 0..10 {
            let data = format!("Chunk {}", i).into_bytes();
            writer.write(&data).await?;
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        }
        
        writer.close().await?;
        Ok(json!({ "status": "complete" }))
    });
    
    iii.connect().await?;
    
    // Create channel
    let channel = iii.create_channel(Some(100)).await?;
    
    // Call producer function with writer reference
    let producer_task = tokio::spawn({
        let iii = iii.clone();
        let writer_ref = channel.writer_ref.clone();
        async move {
            iii.call("generate_data", json!({
                "output": writer_ref
            })).await
        }
    });
    
    // Read all data
    let data = channel.reader.read_all().await?;
    println!("Received {} bytes", data.len());
    
    producer_task.await??;
    
    Ok(())
}
```

## See Also

* [Functions API](/rust/api/functions) - Pass channels to functions
* [Streaming API](/rust/api/streaming) - Stream updates and atomic operations
* [Invocation API](/rust/api/invocation) - Call functions with channel arguments