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 enable streaming data transfer between workers in a distributed III application. They provide bidirectional communication with reader and writer endpoints.
Channel
A streaming channel pair created by createChannel(), containing both local handles and serializable references.
type Channel = {
writer: ChannelWriter
reader: ChannelReader
writerRef: StreamChannelRef
readerRef: StreamChannelRef
}
@dataclass
class Channel:
writer: ChannelWriter
reader: ChannelReader
writer_ref: StreamChannelRef
reader_ref: StreamChannelRef
pub struct Channel {
pub writer: ChannelWriter,
pub reader: ChannelReader,
pub writer_ref: StreamChannelRef,
pub reader_ref: StreamChannelRef,
}
Fields
Local writer for sending data through the channel.
Local reader for receiving data from the channel.
Serializable reference to the writer endpoint. Pass this to other workers to allow them to write to this channel.
Serializable reference to the reader endpoint. Pass this to other workers to allow them to read from this channel.
StreamChannelRef
A serializable reference to a channel endpoint that can be passed in function invocation data.
type StreamChannelRef = {
channel_id: string
access_key: string
direction: 'read' | 'write'
}
class StreamChannelRef(BaseModel):
channel_id: str
access_key: str
direction: Literal["read", "write"]
pub struct StreamChannelRef {
pub channel_id: String,
pub access_key: String,
pub direction: ChannelDirection,
}
pub enum ChannelDirection {
Read,
Write,
}
Fields
Unique identifier for the channel.
Secret access key for authenticating to the channel endpoint.
Whether this reference is for reading from or writing to the channel.
ChannelWriter
Writer endpoint for sending data through a channel.
class ChannelWriter {
readonly stream: Writable
sendMessage(msg: string): void
close(): void
}
class ChannelWriter:
stream: WritableStream
async def write(self, data: bytes) -> None
def send_message(self, msg: str) -> None
async def send_message_async(self, msg: str) -> None
def close(self) -> None
async def close_async(self) -> None
pub struct ChannelWriter {
// Internal fields
}
impl ChannelWriter {
pub async fn write(&self, data: &[u8]) -> Result<(), IIIError>
pub async fn send_message(&self, msg: &str) -> Result<(), IIIError>
pub async fn close(&self) -> Result<(), IIIError>
}
Properties
Node.js Writable stream interface for sending binary data.
Methods
Write binary data to the channel. Data is automatically chunked into frames.
Send a text message through the channel (separate from binary data stream).
Close the writer endpoint and signal completion to the reader.
ChannelReader
Reader endpoint for receiving data from a channel.
class ChannelReader {
readonly stream: Readable
onMessage(callback: (msg: string) => void): void
}
class ChannelReader:
stream: ReadableStream
def on_message(self, callback: Callable[[str], Any]) -> None
async def __aiter__(self) -> AsyncIterator[bytes]
async def read_all(self) -> bytes
async def close_async(self) -> None
pub struct ChannelReader {
// Internal fields
}
impl ChannelReader {
pub async fn on_message<F>(&self, callback: F)
where
F: Fn(String) + Send + Sync + 'static
pub async fn next_binary(&self) -> Result<Option<Vec<u8>>, IIIError>
pub async fn read_all(&self) -> Result<Vec<u8>, IIIError>
pub async fn close(&self) -> Result<(), IIIError>
}
Properties
Node.js Readable stream interface for receiving binary data.
Methods
Register a callback to receive text messages (separate from binary data stream).
Read the next binary chunk from the channel. Returns None when stream is closed.
Read the entire stream into a single buffer.
Close the reader endpoint.
Usage Examples
Basic Channel Communication
// Worker A: Create and send channel
const channel = await iii.createChannel()
// Pass writer ref to another worker
await iii.trigger('worker-b::process', {
input: 'some data',
output_channel: channel.writerRef
})
// Read results from the channel
for await (const chunk of channel.reader.stream) {
console.log('Received:', chunk.toString())
}
# Worker A: Create and send channel
channel = await iii.create_channel()
# Pass writer ref to another worker
await iii.trigger('worker-b::process', {
'input': 'some data',
'output_channel': channel.writer_ref
})
# Read results from the channel
async for chunk in channel.reader:
print(f'Received: {chunk.decode()}')
// Worker A: Create and send channel
let channel = iii.create_channel(None).await?;
// Pass writer ref to another worker
iii.trigger(
"worker-b::process",
json!({
"input": "some data",
"output_channel": channel.writer_ref
})
).await?;
// Read results from the channel
while let Some(chunk) = channel.reader.next_binary().await? {
println!("Received: {}", String::from_utf8_lossy(&chunk));
}
Receiving Channel Reference
// Worker B: Receive channel ref and write to it
type ProcessInput = {
input: string
output_channel: StreamChannelRef
}
iii.registerFunction(
{ id: 'worker-b::process' },
async (data: ProcessInput) => {
const writer = new ChannelWriter(engineUrl, data.output_channel)
// Write results to the channel
writer.stream.write('Processing...\n')
const result = await processData(data.input)
writer.stream.write(`Result: ${result}\n`)
writer.close()
return { success: true }
}
)
# Worker B: Receive channel ref and write to it
@dataclass
class ProcessInput:
input: str
output_channel: StreamChannelRef
async def process_data(data: ProcessInput):
writer = ChannelWriter(engine_url, data.output_channel)
# Write results to the channel
await writer.write(b'Processing...\n')
result = await process_data(data.input)
await writer.write(f'Result: {result}\n'.encode())
writer.close()
return {'success': True}
iii.register_function('worker-b::process', process_data)
// Worker B: Receive channel ref and write to it
#[derive(Deserialize)]
struct ProcessInput {
input: String,
output_channel: StreamChannelRef,
}
iii.register_function("worker-b::process", |input: Value| {
Box::pin(async move {
let data: ProcessInput = serde_json::from_value(input)?;
let writer = ChannelWriter::new(&engine_url, &data.output_channel);
// Write results to the channel
writer.write(b"Processing...\n").await?;
let result = process_data(&data.input).await?;
writer.write(format!("Result: {}\n", result).as_bytes()).await?;
writer.close().await?;
Ok(json!({ "success": true }))
})
});
Streaming Large Files
import { createReadStream } from 'fs'
import { pipeline } from 'stream/promises'
const channel = await iii.createChannel()
// Stream file to another worker
await iii.trigger('worker-b::save-file', {
filename: 'large-file.dat',
data_channel: channel.readerRef
})
// Pipe file data through the channel
const fileStream = createReadStream('large-file.dat')
await pipeline(fileStream, channel.writer.stream)
channel = await iii.create_channel()
# Stream file to another worker
await iii.trigger('worker-b::save-file', {
'filename': 'large-file.dat',
'data_channel': channel.reader_ref
})
# Stream file data through the channel
async with aiofiles.open('large-file.dat', 'rb') as f:
while chunk := await f.read(64 * 1024):
await channel.writer.write(chunk)
await channel.writer.close_async()
use tokio::fs::File;
use tokio::io::AsyncReadExt;
let channel = iii.create_channel(None).await?;
// Stream file to another worker
iii.trigger(
"worker-b::save-file",
json!({
"filename": "large-file.dat",
"data_channel": channel.reader_ref
})
).await?;
// Stream file data through the channel
let mut file = File::open("large-file.dat").await?;
let mut buffer = vec![0; 64 * 1024];
loop {
let n = file.read(&mut buffer).await?;
if n == 0 { break; }
channel.writer.write(&buffer[..n]).await?;
}
channel.writer.close().await?;
Configuration
Buffer Size
When creating a channel, you can optionally specify a buffer size:
const channel = await iii.createChannel(128) // 128 message buffer
channel = await iii.create_channel(buffer_size=128)
let channel = iii.create_channel(Some(128)).await?;