> ## 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.

# Channel Types

> Types for worker-to-worker streaming channels

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.

<Tabs>
  <Tab title="TypeScript">
    ```typescript theme={null}
    type Channel = {
      writer: ChannelWriter
      reader: ChannelReader
      writerRef: StreamChannelRef
      readerRef: StreamChannelRef
    }
    ```
  </Tab>

  <Tab title="Python">
    ```python theme={null}
    @dataclass
    class Channel:
        writer: ChannelWriter
        reader: ChannelReader
        writer_ref: StreamChannelRef
        reader_ref: StreamChannelRef
    ```
  </Tab>

  <Tab title="Rust">
    ```rust theme={null}
    pub struct Channel {
        pub writer: ChannelWriter,
        pub reader: ChannelReader,
        pub writer_ref: StreamChannelRef,
        pub reader_ref: StreamChannelRef,
    }
    ```
  </Tab>
</Tabs>

### Fields

<ResponseField name="writer" type="ChannelWriter" required>
  Local writer for sending data through the channel.
</ResponseField>

<ResponseField name="reader" type="ChannelReader" required>
  Local reader for receiving data from the channel.
</ResponseField>

<ResponseField name="writerRef" type="StreamChannelRef" required>
  Serializable reference to the writer endpoint. Pass this to other workers to allow them to write to this channel.
</ResponseField>

<ResponseField name="readerRef" type="StreamChannelRef" required>
  Serializable reference to the reader endpoint. Pass this to other workers to allow them to read from this channel.
</ResponseField>

## StreamChannelRef

A serializable reference to a channel endpoint that can be passed in function invocation data.

<Tabs>
  <Tab title="TypeScript">
    ```typescript theme={null}
    type StreamChannelRef = {
      channel_id: string
      access_key: string
      direction: 'read' | 'write'
    }
    ```
  </Tab>

  <Tab title="Python">
    ```python theme={null}
    class StreamChannelRef(BaseModel):
        channel_id: str
        access_key: str
        direction: Literal["read", "write"]
    ```
  </Tab>

  <Tab title="Rust">
    ```rust theme={null}
    pub struct StreamChannelRef {
        pub channel_id: String,
        pub access_key: String,
        pub direction: ChannelDirection,
    }

    pub enum ChannelDirection {
        Read,
        Write,
    }
    ```
  </Tab>
</Tabs>

### Fields

<ResponseField name="channel_id" type="string" required>
  Unique identifier for the channel.
</ResponseField>

<ResponseField name="access_key" type="string" required>
  Secret access key for authenticating to the channel endpoint.
</ResponseField>

<ResponseField name="direction" type="'read' | 'write'" required>
  Whether this reference is for reading from or writing to the channel.
</ResponseField>

## ChannelWriter

Writer endpoint for sending data through a channel.

<Tabs>
  <Tab title="TypeScript">
    ```typescript theme={null}
    class ChannelWriter {
      readonly stream: Writable
      sendMessage(msg: string): void
      close(): void
    }
    ```
  </Tab>

  <Tab title="Python">
    ```python theme={null}
    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
    ```
  </Tab>

  <Tab title="Rust">
    ```rust theme={null}
    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>
    }
    ```
  </Tab>
</Tabs>

### Properties

<ResponseField name="stream" type="WritableStream" required>
  Node.js Writable stream interface for sending binary data.
</ResponseField>

### Methods

<ResponseField name="write" type="function">
  Write binary data to the channel. Data is automatically chunked into frames.
</ResponseField>

<ResponseField name="sendMessage" type="function">
  Send a text message through the channel (separate from binary data stream).
</ResponseField>

<ResponseField name="close" type="function">
  Close the writer endpoint and signal completion to the reader.
</ResponseField>

## ChannelReader

Reader endpoint for receiving data from a channel.

<Tabs>
  <Tab title="TypeScript">
    ```typescript theme={null}
    class ChannelReader {
      readonly stream: Readable
      onMessage(callback: (msg: string) => void): void
    }
    ```
  </Tab>

  <Tab title="Python">
    ```python theme={null}
    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
    ```
  </Tab>

  <Tab title="Rust">
    ```rust theme={null}
    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>
    }
    ```
  </Tab>
</Tabs>

### Properties

<ResponseField name="stream" type="ReadableStream" required>
  Node.js Readable stream interface for receiving binary data.
</ResponseField>

### Methods

<ResponseField name="onMessage" type="function">
  Register a callback to receive text messages (separate from binary data stream).
</ResponseField>

<ResponseField name="next_binary" type="function">
  Read the next binary chunk from the channel. Returns `None` when stream is closed.
</ResponseField>

<ResponseField name="read_all" type="function">
  Read the entire stream into a single buffer.
</ResponseField>

<ResponseField name="close" type="function">
  Close the reader endpoint.
</ResponseField>

## Usage Examples

### Basic Channel Communication

<Tabs>
  <Tab title="TypeScript">
    ```typescript theme={null}
    // 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())
    }
    ```
  </Tab>

  <Tab title="Python">
    ```python theme={null}
    # 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()}')
    ```
  </Tab>

  <Tab title="Rust">
    ```rust theme={null}
    // 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));
    }
    ```
  </Tab>
</Tabs>

### Receiving Channel Reference

<Tabs>
  <Tab title="TypeScript">
    ```typescript theme={null}
    // 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 }
      }
    )
    ```
  </Tab>

  <Tab title="Python">
    ```python theme={null}
    # 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)
    ```
  </Tab>

  <Tab title="Rust">
    ```rust theme={null}
    // 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 }))
        })
    });
    ```
  </Tab>
</Tabs>

### Streaming Large Files

<Tabs>
  <Tab title="TypeScript">
    ```typescript theme={null}
    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)
    ```
  </Tab>

  <Tab title="Python">
    ```python theme={null}
    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()
    ```
  </Tab>

  <Tab title="Rust">
    ```rust theme={null}
    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?;
    ```
  </Tab>
</Tabs>

## Configuration

### Buffer Size

When creating a channel, you can optionally specify a buffer size:

<Tabs>
  <Tab title="TypeScript">
    ```typescript theme={null}
    const channel = await iii.createChannel(128) // 128 message buffer
    ```
  </Tab>

  <Tab title="Python">
    ```python theme={null}
    channel = await iii.create_channel(buffer_size=128)
    ```
  </Tab>

  <Tab title="Rust">
    ```rust theme={null}
    let channel = iii.create_channel(Some(128)).await?;
    ```
  </Tab>
</Tabs>

**Default**: 64 messages

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

## Related

* [Streaming Concepts](/concepts/streaming)
* [Channels Concepts](/concepts/channels)
* [HTTP Request/Response Types](/api/types/api-request-response)