Channels

Stream large or binary payloads between iii workers.

Goal

Stream a large or binary payload from one function to another without putting the data itself in a JSON function payload.

Use a normal function invocation when the payload is small JSON and can be handled as one request. Use a channel when the payload is large, binary, or naturally stream-shaped.

Good fits for channels:

• File uploads and downloads.
• Images, audio, video, PDFs, and datasets.
• Progress updates during long-running work.
• Producer and consumer pipelines where the data should move as a stream.

For the underlying model, see Channels architecture.

Steps

1. Create a channel

A channel has two local stream objects and two serializable refs:

• writer: local writable stream.
• reader: local readable stream.
• writerRef: serializable token for the writer end.
• readerRef: serializable token for the reader end.

Node / TypeScript

const channel = await iii.createChannel();

// channel.writer
// channel.reader
// channel.writerRef
// channel.readerRef

Python

channel = iii_client.create_channel()

# channel.writer
# channel.reader
# channel.writer_ref
# channel.reader_ref

Rust

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

// channel.writer
// channel.reader
// channel.writer_ref
// channel.reader_ref

2. Write to the channel

Write the stream payload to the local writer and close it when you are done.

Node / TypeScript

const channel = await iii.createChannel();

channel.writer.stream.end(Buffer.from("file contents"));

Python

channel = await iii_client.create_channel_async()

await channel.writer.write(b"file contents")
await channel.writer.close_async()

Rust

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

channel.writer.write(b"file contents").await?;
channel.writer.close().await?;

3. Pass the reader ref to another function

Pass the readerRef / reader_ref as part of a normal function invocation. The receiving function uses that ref to read from the channel.

Node / TypeScript

const result = await iii.trigger({
  function_id: "files::process",
  payload: {
    filename: "report.csv",
    reader: channel.readerRef,
  },
});

Python

result = await iii_client.trigger_async({
    "function_id": "files::process",
    "payload": {
        "filename": "report.csv",
        "reader": channel.reader_ref.model_dump(),
    },
})

Rust

use iii_sdk::TriggerRequest;
use serde_json::json;

let result = iii
    .trigger(TriggerRequest {
        function_id: "files::process".to_string(),
        payload: json!({
            "filename": "report.csv",
            "reader": channel.reader_ref,
        }),
        action: None,
        timeout_ms: None,
    })
    .await?;

4. Read from the channel

Node and Python deserialize channel refs into live channel objects before your handler runs. Rust receives the ref in JSON and reconstructs the reader with ChannelReader::new(...).

Node / TypeScript

import type { ChannelReader } from "iii-sdk";

worker.registerFunction("files::process", async (input: { reader: ChannelReader }) => {
  let bytes = 0;

  for await (const chunk of input.reader.stream) {
    bytes += Buffer.isBuffer(chunk) ? chunk.length : Buffer.byteLength(chunk);
  }

  return { bytes };
});

Python

async def process_file(input: dict) -> dict:
    reader = input["reader"]
    chunks = []

    async for chunk in reader:
        chunks.append(chunk)

    return {"bytes": sum(len(chunk) for chunk in chunks)}

worker.register_function("files::process", process_file)

Rust

use iii_sdk::{ChannelDirection, ChannelReader, IIIError};
use serde_json::json;

let refs = iii_sdk::extract_channel_refs(&input);
let reader_ref = refs
    .iter()
    .find(|(key, ref_)| key == "reader" && matches!(ref_.direction, ChannelDirection::Read))
    .map(|(_, ref_)| ref_.clone())
    .ok_or_else(|| IIIError::Handler("missing reader channel ref".to_string()))?;

let reader = ChannelReader::new(iii.address(), &reader_ref);
let mut bytes = 0;

while let Some(chunk) = reader.next_binary().await? {
    bytes += chunk.len();
}

Ok(json!({ "bytes": bytes }))

Result

The caller passes only a small ref through trigger(). The stream payload travels over the channel, and the receiving function reads it incrementally.

Related pages

• Channels architecture
• Node SDK channels
• Python SDK channels
• Rust SDK channels