Channels

Stream data between workers using channels

Overview

Channels provide WebSocket-backed streams for worker-to-worker data transfer. They enable streaming large payloads, real-time data processing, and bidirectional communication.

Creating Channels

create_channel

Create a streaming channel pair with a writer and reader.

channel = await iii.create_channel(buffer_size=64)

print(channel.writer)      # ChannelWriter instance
print(channel.reader)      # ChannelReader instance
print(channel.writer_ref)  # Serializable StreamChannelRef
print(channel.reader_ref)  # Serializable StreamChannelRef

buffer_size · int · default: 64

Optional buffer size for the channel

channel · Channel

Channel object containing writer, reader, and their serializable references

Channel Type

@dataclass
class Channel:
    writer: ChannelWriter  # Write binary data
    reader: ChannelReader  # Read binary data
    writer_ref: StreamChannelRef  # Pass to other functions
    reader_ref: StreamChannelRef  # Pass to other functions

ChannelWriter

WebSocket-backed writer for streaming binary data and text messages.

write

Write binary data to the channel.

writer = channel.writer

await writer.write(b"Hello, world!")
await writer.write(b"More data...")

data · bytes · required

Binary data to write. Large payloads are automatically chunked (64KB per frame)

send_message_async

Send a text message to the channel.

await writer.send_message_async('{"type": "metadata", "value": 123}')

msg · str · required

Text message to send

send_message

Fire-and-forget text message (non-async).

writer.send_message('{"type": "status", "value": "processing"}')

msg · str · required

Text message to send

close_async

Close the writer connection.

await writer.close_async()

close

Fire-and-forget close (non-async).

writer.close()

stream Property

Access the writer as a WritableStream interface:

stream = writer.stream
stream.write(b"data")  # Fire-and-forget write
stream.end(b"final data")  # Write and close

ChannelReader

WebSocket-backed reader for streaming binary data and text messages.

Async Iteration

Read binary chunks using async iteration:

reader = channel.reader

async for chunk in reader:
    print(f"Received {len(chunk)} bytes")
    # Process chunk...

read_all

Read the entire stream into a single bytes object:

data = await reader.read_all()
print(f"Total size: {len(data)} bytes")

data · bytes

All data from the stream concatenated together

on_message

Register a callback for text messages:

def handle_message(msg: str):
    print(f"Message: {msg}")

reader.on_message(handle_message)

# Now iterate over binary data
async for chunk in reader:
    # handle_message is called for text messages
    # this loop only yields binary data
    process(chunk)

callback · Callable[[str], Any] · required

Function called for each text message received

close_async

Close the reader connection.

await reader.close_async()

stream Property

Access the reader as a ReadableStream interface:

stream = reader.stream

async for chunk in stream:
    print(chunk)

Passing Channels Between Functions

Channels can be passed between functions using their serializable references:

async def producer(data):
    # Create a channel
    channel = await iii.create_channel()
    
    # Pass the reader reference to a consumer
    iii.call_void("consumer", {"reader": channel.reader_ref})
    
    # Write data
    for i in range(10):
        await channel.writer.write(f"Chunk {i}\n".encode())
        await asyncio.sleep(0.1)
    
    await channel.writer.close_async()
    return {"status": "sent"}

async def consumer(data):
    # The reader_ref is automatically resolved to a ChannelReader
    reader = data["reader"]
    
    chunks = []
    async for chunk in reader:
        chunks.append(chunk.decode())
    
    print(f"Received: {''.join(chunks)}")
    return {"status": "received", "chunks": len(chunks)}

iii.register_function("producer", producer)
iii.register_function("consumer", consumer)

# Start the producer
await iii.call("producer", {})

Streaming Patterns

Producer-Consumer

async def stream_logs(data):
    channel = await iii.create_channel()
    
    # Start consumer in background
    iii.call_void("process_logs", {"reader": channel.reader_ref})
    
    # Stream log lines
    with open("/var/log/app.log", "rb") as f:
        while chunk := f.read(8192):
            await channel.writer.write(chunk)
    
    await channel.writer.close_async()
    return {"status": "streaming_complete"}

async def process_logs(data):
    reader = data["reader"]
    
    line_buffer = b""
    async for chunk in reader:
        line_buffer += chunk
        while b"\n" in line_buffer:
            line, line_buffer = line_buffer.split(b"\n", 1)
            # Process line
            print(f"Log: {line.decode()}")
    
    return {"status": "processed"}

Bidirectional Communication

async def chat_handler(data):
    # Create two channels for bidirectional communication
    to_client = await iii.create_channel()
    from_client = await iii.create_channel()
    
    # Pass both channels to client
    iii.call_void("chat_client", {
        "input": to_client.reader_ref,
        "output": from_client.writer_ref
    })
    
    # Send messages to client
    await to_client.writer.write(b"Welcome!\n")
    
    # Read responses from client
    async for msg in from_client.reader:
        print(f"Client said: {msg.decode()}")
        # Echo back
        await to_client.writer.write(b"Echo: " + msg)
    
    await to_client.writer.close_async()
    return {"status": "chat_ended"}

async def chat_client(data):
    input_reader = data["input"]
    output_writer = data["output"]
    
    # Read from input, write to output
    async for msg in input_reader:
        print(f"Server said: {msg.decode()}")
        response = input("You: ").encode() + b"\n"
        await output_writer.write(response)
    
    await output_writer.close_async()
    return {"status": "done"}

File Upload

async def upload_file(data):
    channel = await iii.create_channel(buffer_size=128)
    
    # Pass writer to uploader
    upload_task = iii.call("file_uploader", {
        "filename": "data.bin",
        "reader": channel.reader_ref
    })
    
    # Stream file contents
    with open("local_file.bin", "rb") as f:
        while chunk := f.read(64 * 1024):  # 64KB chunks
            await channel.writer.write(chunk)
    
    await channel.writer.close_async()
    
    # Wait for upload confirmation
    result = await upload_task
    return result

async def file_uploader(data):
    filename = data["filename"]
    reader = data["reader"]
    
    total_bytes = 0
    with open(f"/uploads/{filename}", "wb") as f:
        async for chunk in reader:
            f.write(chunk)
            total_bytes += len(chunk)
    
    return {
        "status": "uploaded",
        "filename": filename,
        "size": total_bytes
    }

Transform Stream

async def transform_data(data):
    input_channel = await iii.create_channel()
    output_channel = await iii.create_channel()
    
    # Chain transformations
    iii.call_void("uppercase_transform", {
        "input": input_channel.reader_ref,
        "output": output_channel.writer_ref
    })
    
    # Send data
    await input_channel.writer.write(b"hello world\n")
    await input_channel.writer.write(b"streaming data\n")
    await input_channel.writer.close_async()
    
    # Read transformed results
    result = await output_channel.reader.read_all()
    return {"transformed": result.decode()}

async def uppercase_transform(data):
    reader = data["input"]
    writer = data["output"]
    
    async for chunk in reader:
        transformed = chunk.upper()
        await writer.write(transformed)
    
    await writer.close_async()
    return {"status": "complete"}

HTTP Response Streaming

Use channels with HTTP triggers for streaming responses:

from iii import HttpResponse

async def stream_logs_http(data):
    # data contains an embedded response channel
    reader = data["request_body"]
    writer = data["response"]
    
    response = HttpResponse(writer)
    await response.status(200)
    await response.headers({"Content-Type": "text/plain"})
    
    # Stream log lines
    with open("/var/log/app.log", "rb") as f:
        while line := f.readline():
            await response.stream.write(line)
    
    response.close()
    return {"status": "streamed"}

WritableStream

Node.js-style writable stream interface:

class WritableStream:
    def write(self, data: bytes) -> None:
        """Fire-and-forget binary write."""
    
    def end(self, data: bytes | None = None) -> None:
        """Write optional final data and close the stream."""

Usage

writer = channel.writer
stream = writer.stream

stream.write(b"chunk 1")
stream.write(b"chunk 2")
stream.end(b"final chunk")

ReadableStream

Node.js-style readable stream interface:

class ReadableStream:
    async def __aiter__(self) -> AsyncIterator[bytes]:
        """Async iteration over binary chunks."""

Usage

reader = channel.reader
stream = reader.stream

async for chunk in stream:
    print(chunk)

StreamChannelRef

Serializable reference to a channel endpoint:

class StreamChannelRef(BaseModel):
    channel_id: str
    access_key: str
    direction: Literal["read", "write"]

These references are automatically resolved to ChannelReader or ChannelWriter when passed in function invocations.

Error Handling

try:
    channel = await iii.create_channel()
    await channel.writer.write(b"data")
    await channel.writer.close_async()
except Exception as e:
    print(f"Channel error: {e}")

Best Practices

1. Always close channels: Call close_async() or close() when done writing
2. Handle errors: Wrap channel operations in try-except blocks
3. Use appropriate buffer sizes: Larger buffers for high-throughput scenarios
4. Chunking: Large payloads are automatically chunked at 64KB per frame
5. Text vs Binary: Use write() for binary data, send_message() for text

Example: Image Processing Pipeline

import asyncio
from iii import III

iii = III("ws://localhost:49134")

async def download_image(data):
    url = data["url"]
    channel = await iii.create_channel(buffer_size=256)
    
    # Start processing pipeline
    process_task = iii.call("process_image", {
        "reader": channel.reader_ref,
        "format": "thumbnail"
    })
    
    # Simulate downloading image
    image_data = b"...binary image data..."
    await channel.writer.write(image_data)
    await channel.writer.close_async()
    
    # Wait for processed result
    result = await process_task
    return result

async def process_image(data):
    reader = data["reader"]
    format_type = data["format"]
    
    # Read entire image
    image_data = await reader.read_all()
    
    # Process image (resize, compress, etc.)
    processed = process_image_data(image_data, format_type)
    
    # Create output channel
    output_channel = await iii.create_channel()
    
    # Upload processed image
    iii.call_void("upload_image", {
        "reader": output_channel.reader_ref,
        "filename": f"thumbnail.{format_type}"
    })
    
    await output_channel.writer.write(processed)
    await output_channel.writer.close_async()
    
    return {"status": "processed", "format": format_type}

async def upload_image(data):
    reader = data["reader"]
    filename = data["filename"]
    
    image_data = await reader.read_all()
    
    # Upload to storage
    url = save_to_storage(filename, image_data)
    
    return {"url": url, "size": len(image_data)}

iii.register_function("download_image", download_image)
iii.register_function("process_image", process_image)
iii.register_function("upload_image", upload_image)

async def main():
    await iii.connect()
    
    result = await iii.call("download_image", {
        "url": "https://example.com/image.jpg"
    })
    print(f"Image processed: {result}")

if __name__ == "__main__":
    asyncio.run(main())