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.

pub async fn create_channel(&self, buffer_size: Option<usize>) -> Result<Channel, IIIError>

buffer_size · Option<usize>

Internal buffer size for the channel (default determined by engine)

Channel · Result<Channel, IIIError>

A channel with writer, reader, and their serializable references

Example:

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.

pub async fn write(&self, data: &[u8]) -> Result<(), IIIError>

data · &[u8] · required

Binary data to write to the channel

Example:

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?;

Data larger than 64KB is automatically chunked into multiple WebSocket frames.

send_message

Send a text message to the channel.

pub async fn send_message(&self, msg: &str) -> Result<(), IIIError>

msg · &str · required

Text message to send

Example:

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.

pub async fn close(&self) -> Result<(), IIIError>

Example:

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.

pub async fn next_binary(&self) -> Result<Option<Vec<u8>>, IIIError>

Option<Vec<u8>> · Result<Option<Vec<u8>>, IIIError>

The next binary chunk, or None when the channel is closed

Example:

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");

Text messages are not returned by next_binary(). Register a callback with on_message() to handle text messages.

read_all

Read the entire stream into a single buffer.

pub async fn read_all(&self) -> Result<Vec<u8>, IIIError>

Vec<u8> · Result<Vec<u8>, IIIError>

All binary data from the channel concatenated into a single buffer

Example:

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.

pub async fn on_message<F>(&self, callback: F)
where
    F: Fn(String) + Send + Sync + 'static

callback · F · required

Callback invoked for each text message received

Example:

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.

pub async fn close(&self) -> Result<(), IIIError>

Passing Channels to Functions

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

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:

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.

pub fn is_channel_ref(value: &Value) -> bool

Example:

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.

pub fn extract_channel_refs(data: &Value) -> Vec<(String, StreamChannelRef)>

data · &Value · required

JSON value to search for channel references

Vec<(String, StreamChannelRef)> · Vec<(String, StreamChannelRef)>

List of (field path, channel reference) tuples

Example:

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.

pub struct Channel {
    pub writer: ChannelWriter,
    pub reader: ChannelReader,
    pub writer_ref: StreamChannelRef,
    pub reader_ref: StreamChannelRef,
}

StreamChannelRef

Serializable reference to a channel endpoint.

pub struct StreamChannelRef {
    pub channel_id: String,
    pub access_key: String,
    pub direction: ChannelDirection,
}

ChannelDirection

Direction of a channel reference.

pub enum ChannelDirection {
    Read,
    Write,
}

Complete Example

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

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 - Pass channels to functions
• Streaming API - Stream updates and atomic operations
• Invocation API - Call functions with channel arguments