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.
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>
Internal buffer size for the channel (default determined by engine)
Channel
Result<Channel, IIIError>
A channel with writer, reader, and their serializable references
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>
Binary data to write to the channel
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>
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>
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
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
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 invoked for each text message received
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
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");
}
pub fn extract_channel_refs(data: &Value) -> Vec<(String, StreamChannelRef)>
JSON value to search for channel references
Vec<(String, StreamChannelRef)>
Vec<(String, StreamChannelRef)>
List of (field path, channel reference) tuples
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