> ## 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
> 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.
```python theme={null}
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
```
Optional buffer size for the channel
Channel object containing writer, reader, and their serializable references
### Channel Type
```python theme={null}
@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.
```python theme={null}
writer = channel.writer
await writer.write(b"Hello, world!")
await writer.write(b"More data...")
```
Binary data to write. Large payloads are automatically chunked (64KB per frame)
### send\_message\_async
Send a text message to the channel.
```python theme={null}
await writer.send_message_async('{"type": "metadata", "value": 123}')
```
Text message to send
### send\_message
Fire-and-forget text message (non-async).
```python theme={null}
writer.send_message('{"type": "status", "value": "processing"}')
```
Text message to send
### close\_async
Close the writer connection.
```python theme={null}
await writer.close_async()
```
### close
Fire-and-forget close (non-async).
```python theme={null}
writer.close()
```
### stream Property
Access the writer as a `WritableStream` interface:
```python theme={null}
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:
```python theme={null}
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:
```python theme={null}
data = await reader.read_all()
print(f"Total size: {len(data)} bytes")
```
All data from the stream concatenated together
### on\_message
Register a callback for text messages:
```python theme={null}
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)
```
Function called for each text message received
### close\_async
Close the reader connection.
```python theme={null}
await reader.close_async()
```
### stream Property
Access the reader as a `ReadableStream` interface:
```python theme={null}
stream = reader.stream
async for chunk in stream:
print(chunk)
```
## Passing Channels Between Functions
Channels can be passed between functions using their serializable references:
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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:
```python theme={null}
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:
```python theme={null}
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
```python theme={null}
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:
```python theme={null}
class ReadableStream:
async def __aiter__(self) -> AsyncIterator[bytes]:
"""Async iteration over binary chunks."""
```
### Usage
```python theme={null}
reader = channel.reader
stream = reader.stream
async for chunk in stream:
print(chunk)
```
## StreamChannelRef
Serializable reference to a channel endpoint:
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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())
```