Channel Types - III SDK

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.

TypeScript
Python
Rust

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

ChannelWriter

required

Local writer for sending data through the channel.

ChannelReader

required

Local reader for receiving data from the channel.

StreamChannelRef

required

Serializable reference to the writer endpoint. Pass this to other workers to allow them to write to this channel.

StreamChannelRef

required

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.

TypeScript
Python
Rust

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

string

required

Unique identifier for the channel.

string

required

Secret access key for authenticating to the channel endpoint.

'read' | 'write'

required

Whether this reference is for reading from or writing to the channel.

ChannelWriter

Writer endpoint for sending data through a channel.

TypeScript
Python
Rust

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

WritableStream

required

Node.js Writable stream interface for sending binary data.

Methods

function

Write binary data to the channel. Data is automatically chunked into frames.

function

Send a text message through the channel (separate from binary data stream).

function

Close the writer endpoint and signal completion to the reader.

ChannelReader

Reader endpoint for receiving data from a channel.

TypeScript
Python
Rust

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

ReadableStream

required

Node.js Readable stream interface for receiving binary data.

Methods

function

Read the next binary chunk from the channel. Returns None when stream is closed.

function

Read the entire stream into a single buffer.

function

Close the reader endpoint.

Usage Examples

Basic Channel Communication

TypeScript
Python
Rust

// 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

TypeScript
Python
Rust

// 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

TypeScript
Python
Rust

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:

TypeScript
Python
Rust

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

Default: 64 messages The buffer size determines how many messages can be queued in the channel before backpressure is applied to the writer.

​Channel

​Fields

​StreamChannelRef

​Fields

​ChannelWriter

​Properties

​Methods

​ChannelReader

​Properties

​Methods

​Usage Examples

​Basic Channel Communication

​Receiving Channel Reference

​Streaming Large Files

​Configuration

​Buffer Size

​Related

Channel

Fields

StreamChannelRef

Fields

ChannelWriter

Properties

Methods

ChannelReader

Properties

Methods

Usage Examples

Basic Channel Communication

Receiving Channel Reference

Streaming Large Files

Configuration

Buffer Size

Related