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

> Bidirectional streaming channels for worker-to-worker data transfer with ChannelWriter and ChannelReader

## Overview

Channels provide bidirectional streaming for transferring large datasets between workers without loading everything into memory. They support both binary data streaming and text message passing, making them ideal for processing files, media, logs, and real-time data feeds.

## Channel Architecture

Channels are created by the Engine and consist of:

<Steps>
  <Step title="Writer Endpoint">
    Sends data to the channel. Multiple chunks can be written sequentially.
  </Step>

  <Step title="Engine Buffer">
    Stores data temporarily (configurable buffer size, default 64 items)
  </Step>

  <Step title="Reader Endpoint">
    Receives data from the channel. Supports Node.js Readable stream interface.
  </Step>
</Steps>

```mermaid theme={null}
sequenceDiagram
    participant W1 as Worker 1
    participant E as Engine
    participant W2 as Worker 2
    
    W1->>E: createChannel()
    E->>W1: {writer, reader} refs
    W1->>E: call('process', {reader: readerRef})
    E->>W2: InvokeFunction {reader: ChannelReader}
    W1->>E: writer.stream.write(chunk1)
    E->>W2: Binary data chunk1
    W1->>E: writer.stream.write(chunk2)
    E->>W2: Binary data chunk2
    W1->>E: writer.stream.end()
    E->>W2: Stream complete
    W2->>E: InvocationResult {result}
    E->>W1: result
```

## Creating Channels

### Basic Channel Creation

```typescript theme={null}
const channel = await iii.createChannel()

console.log(channel.writer)    // ChannelWriter instance (local)
console.log(channel.reader)    // ChannelReader instance (local)
console.log(channel.writerRef) // StreamChannelRef (serializable)
console.log(channel.readerRef) // StreamChannelRef (serializable)
```

**Channel structure:**

```typescript theme={null}
// Source: packages/node/iii/src/types.ts:215-220
type Channel = {
  writer: ChannelWriter        // Local writer instance
  reader: ChannelReader        // Local reader instance
  writerRef: StreamChannelRef  // Pass to other functions
  readerRef: StreamChannelRef  // Pass to other functions
}
```

**StreamChannelRef structure:**

```typescript theme={null}
// Source: packages/node/iii/src/iii-types.ts:198-202
type StreamChannelRef = {
  channel_id: string     // Unique channel identifier
  access_key: string     // Authentication key
  direction: 'read' | 'write'
}
```

### Custom Buffer Size

```typescript theme={null}
const channel = await iii.createChannel(128)  // Buffer 128 items
```

<Note>
  Buffer size determines how many chunks can be queued before writers block. Larger buffers improve throughput but use more memory.
</Note>

**Implementation:**

```typescript theme={null}
// Source: packages/node/iii/src/iii.ts:275-287
creatChannel = async (bufferSize?: number): Promise<Channel> => {
  const result = await this.call<
    { buffer_size?: number },
    { writer: StreamChannelRef; reader: StreamChannelRef }
  >('engine::channels::create', { buffer_size: bufferSize })
  
  return {
    writer: new ChannelWriter(this.address, result.writer),
    reader: new ChannelReader(this.address, result.reader),
    writerRef: result.writer,
    readerRef: result.reader
  }
}
```

## ChannelWriter

### Writing Binary Data

ChannelWriter provides a Node.js Writable stream interface:

```typescript theme={null}
const channel = await iii.createChannel()

// Write data chunks
const data = Buffer.from('Hello, World!')
channel.writer.stream.write(data, (err) => {
  if (err) console.error('Write failed:', err)
})

// End the stream when done
channel.writer.stream.end()
```

### Automatic Chunking

Large buffers are automatically split into 64KB frames:

```typescript theme={null}
// Source: packages/node/iii/src/channels.ts:82-101
private sendChunked(data: Buffer, callback: (err?: Error | null) => void): void {
  let offset = 0
  const sendNext = (err?: Error | null): void => {
    if (err) {
      callback(err)
      return
    }
    
    if (offset >= data.length) {
      callback(null)
      return
    }
    
    const end = Math.min(offset + ChannelWriter.FRAME_SIZE, data.length)
    const part = data.subarray(offset, end)
    offset = end
    this.sendRaw(part, sendNext)
  }
  sendNext(null)
}
```

<Tip>
  The 64KB frame size (`ChannelWriter.FRAME_SIZE`) balances WebSocket message overhead with memory efficiency.
</Tip>

### Piping Streams

Pipe Node.js readable streams directly:

```typescript theme={null}
import { createReadStream } from 'fs'
import { pipeline } from 'stream/promises'

const channel = await iii.createChannel()

const fileStream = createReadStream('/path/to/large-file.bin')
await pipeline(fileStream, channel.writer.stream)

console.log('File uploaded to channel')
```

### Sending Text Messages

Send out-of-band text messages (separate from binary stream):

```typescript theme={null}
channel.writer.sendMessage(JSON.stringify({
  type: 'progress',
  percent: 50
}))
```

**Implementation:**

```typescript theme={null}
// Source: packages/node/iii/src/channels.ts:66-71
sendMessage(msg: string): void {
  this.ensureConnected()
  this.sendRaw(msg, err => {
    if (err) this.stream.destroy(err)
  })
}
```

<Info>
  Text messages and binary chunks use the same WebSocket but are distinguished by message type. Text messages don't affect the binary stream.
</Info>

### Closing Channels

```typescript theme={null}
// Graceful close (after pending writes)
channel.writer.close()

// Or end the stream
channel.writer.stream.end()
```

**ChannelWriter class structure:**

```typescript theme={null}
// Source: packages/node/iii/src/channels.ts:5-41
class ChannelWriter {
  private static readonly FRAME_SIZE = 64 * 1024
  private ws: WebSocket | null = null
  private wsReady = false
  private readonly pendingMessages: {
    data: Buffer | string
    callback: (err?: Error | null) => void
  }[] = []
  public readonly stream: Writable
  private readonly url: string
  
  constructor(engineWsBase: string, ref: StreamChannelRef) {
    this.url = buildChannelUrl(engineWsBase, ref.channel_id, ref.access_key, 'write')
    
    this.stream = new Writable({
      write: (chunk: Buffer, _encoding, callback) => {
        const buf = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)
        this.sendChunked(buf, callback)
      },
      final: callback => {
        if (!this.ws) {
          callback()
          return
        }
        if (this.wsReady) {
          this.ws.close(1000, 'stream_complete')
        } else {
          this.ws.on('open', () => this.ws?.close(1000, 'stream_complete'))
        }
        callback()
      },
      destroy: (err, callback) => {
        if (this.ws) this.ws.terminate()
        callback(err)
      }
    })
  }
}
```

Location: `packages/node/iii/src/channels.ts:5-111`

## ChannelReader

### Reading Binary Data

ChannelReader provides a Node.js Readable stream interface:

```typescript theme={null}
const channel = await iii.createChannel()

// Read chunks as they arrive
for await (const chunk of channel.reader.stream) {
  const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)
  console.log('Received chunk:', buffer.length, 'bytes')
  // Process chunk...
}

console.log('Stream complete')
```

### Collecting Full Stream

```typescript theme={null}
const chunks: Buffer[] = []
for await (const chunk of channel.reader.stream) {
  chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk))
}

const fullData = Buffer.concat(chunks)
console.log('Total size:', fullData.length, 'bytes')
```

### Piping to Destination

```typescript theme={null}
import { createWriteStream } from 'fs'
import { pipeline } from 'stream/promises'

const channel = await iii.createChannel()
const outputFile = createWriteStream('/path/to/output.bin')

await pipeline(channel.reader.stream, outputFile)
console.log('File downloaded from channel')
```

### Receiving Text Messages

Listen for out-of-band text messages:

```typescript theme={null}
channel.reader.onMessage((msg) => {
  const message = JSON.parse(msg)
  console.log('Progress:', message.percent + '%')
})

// Then read binary stream normally
for await (const chunk of channel.reader.stream) {
  // Process binary data...
}
```

**ChannelReader class structure:**

```typescript theme={null}
// Source: packages/node/iii/src/channels.ts:113-174
class ChannelReader {
  private ws: WebSocket | null = null
  private connected = false
  private readonly messageCallbacks: Array<(msg: string) => void> = []
  public readonly stream: Readable
  private readonly url: string
  
  constructor(engineWsBase: string, ref: StreamChannelRef) {
    this.url = buildChannelUrl(engineWsBase, ref.channel_id, ref.access_key, 'read')
    
    const self = this
    this.stream = new Readable({
      read() {
        self.ensureConnected()
        if (self.ws) self.ws.resume()
      },
      destroy(err, callback) {
        if (self.ws && self.ws.readyState !== WebSocket.CLOSED) {
          self.ws.terminate()
        }
        self.ws = null
        callback(err)
      }
    })
  }
  
  private ensureConnected(): void {
    if (this.connected) return
    this.connected = true
    this.ws = new WebSocket(this.url)
    
    this.ws.on('open', () => {
      (this.ws as unknown as { binaryType: string }).binaryType = 'nodebuffer'
    })
    
    this.ws.on('message', (data: Buffer, isBinary: boolean) => {
      if (isBinary) {
        if (!this.stream.push(data)) {
          this.ws?.pause()
        }
      } else {
        const msg = data.toString('utf-8')
        for (const cb of this.messageCallbacks) {
          cb(msg)
        }
      }
    })
    
    this.ws.on('close', () => {
      this.ws = null
      if (!this.stream.destroyed) this.stream.push(null)
    })
    
    this.ws.on('error', err => {
      this.stream.destroy(err)
    })
  }
  
  onMessage(callback: (msg: string) => void): void {
    this.messageCallbacks.push(callback)
  }
}
```

Location: `packages/node/iii/src/channels.ts:113-174`

## Channel References in Function Calls

Channel references are serializable and can be passed as function arguments:

```typescript theme={null}
type ProcessInput = {
  label: string
  dataStream: ChannelReader
}

iii.registerFunction(
  { id: 'data::process' },
  async (input: ProcessInput) => {
    const { label, dataStream } = input
    
    const chunks: Buffer[] = []
    for await (const chunk of dataStream.stream) {
      chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk))
    }
    
    const data = JSON.parse(Buffer.concat(chunks).toString('utf-8'))
    return { label, recordCount: data.length }
  }
)

// Caller
const channel = await iii.createChannel()

// Send channel reader reference
const resultPromise = iii.call<ProcessInput, any>('data::process', {
  label: 'batch-1',
  dataStream: channel.readerRef  // Serializable reference
})

// Write data
const payload = JSON.stringify([{ a: 1 }, { b: 2 }])
channel.writer.stream.end(Buffer.from(payload))

// Await result
const result = await resultPromise
console.log(result)  // { label: 'batch-1', recordCount: 2 }
```

**Automatic resolution:**

The Engine automatically resolves channel references to `ChannelReader`/`ChannelWriter` instances:

```typescript theme={null}
// Source: packages/node/iii/src/iii.ts:769-786
private resolveChannelValue(value: unknown): unknown {
  if (isChannelRef(value)) {
    return value.direction === 'read'
      ? new ChannelReader(this.address, value)
      : new ChannelWriter(this.address, value)
  }
  if (Array.isArray(value)) {
    return value.map(item => this.resolveChannelValue(item))
  }
  if (value !== null && typeof value === 'object') {
    const out: Record<string, unknown> = {}
    for (const [k, v] of Object.entries(value)) {
      out[k] = this.resolveChannelValue(v)
    }
    return out
  }
  return value
}
```

<Warning>
  Channel references are resolved recursively in objects and arrays. Nested channel refs work automatically.
</Warning>

## Real-World Examples

### Example 1: Data Processing Pipeline

```typescript theme={null}
// Source: packages/node/iii/tests/data-channels.test.ts:6-84
type Record = { name: string; value: number }

const processor = iii.registerFunction(
  { id: 'test.data.processor' },
  async (input: { label: string; reader: ChannelReader }) => {
    const chunks: Buffer[] = []
    for await (const chunk of input.reader.stream) {
      chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk))
    }
    
    const records: Record[] = JSON.parse(Buffer.concat(chunks).toString('utf-8'))
    
    const sum = records.reduce((acc, r) => acc + r.value, 0)
    const max = Math.max(...records.map(r => r.value))
    const min = Math.min(...records.map(r => r.value))
    
    return {
      label: input.label,
      messages: [
        { type: 'stat', key: 'count', value: records.length },
        { type: 'stat', key: 'sum', value: sum },
        { type: 'stat', key: 'average', value: sum / records.length },
        { type: 'stat', key: 'min', value: min },
        { type: 'stat', key: 'max', value: max }
      ]
    }
  }
)

const sender = iii.registerFunction(
  { id: 'test.data.sender' },
  async (input: { records: Record[] }) => {
    const channel = await iii.createChannel()
    
    const writePromise = new Promise<void>((resolve, reject) => {
      const payload = Buffer.from(JSON.stringify(input.records))
      channel.writer.stream.end(payload, (err?: Error | null) => {
        if (err) reject(err)
        else resolve()
      })
    })
    
    const result = await iii.call('test.data.processor', {
      label: 'metrics-batch',
      reader: channel.readerRef
    })
    
    await writePromise
    return result
  }
)

const records: Record[] = [
  { name: 'cpu_usage', value: 72 },
  { name: 'memory_mb', value: 2048 },
  { name: 'disk_iops', value: 340 },
  { name: 'network_mbps', value: 95 },
  { name: 'latency_ms', value: 12 }
]

const result = await iii.call('test.data.sender', { records })
// result.messages: [{type:'stat', key:'count', value:5}, ...]
```

### Example 2: Bidirectional Streaming with Progress

```typescript theme={null}
// Source: packages/node/iii/tests/data-channels.test.ts:86-153 (partial)
const worker = iii.registerFunction(
  { id: 'test.stream.worker' },
  async (input: { reader: ChannelReader; writer: ChannelWriter }) => {
    const { reader, writer } = input
    const chunks: Buffer[] = []
    let chunkCount = 0
    
    for await (const chunk of reader.stream) {
      chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk))
      chunkCount++
      writer.sendMessage(JSON.stringify({
        type: 'progress',
        chunks_received: chunkCount
      }))
    }
    
    const fullData = Buffer.concat(chunks)
    writer.sendMessage(JSON.stringify({
      type: 'complete',
      total_bytes: fullData.length
    }))
    writer.close()
    
    return { receivedBytes: fullData.length }
  }
)

const coordinator = iii.registerFunction(
  { id: 'test.stream.coordinator' },
  async (input: { size: number }) => {
    const inputChannel = await iii.createChannel()
    const outputChannel = await iii.createChannel()
    
    outputChannel.reader.onMessage((msg) => {
      const message = JSON.parse(msg)
      console.log('Worker message:', message)
    })
    
    const resultPromise = iii.call('test.stream.worker', {
      reader: inputChannel.readerRef,
      writer: outputChannel.writerRef
    })
    
    // Send data
    const data = Buffer.alloc(input.size)
    inputChannel.writer.stream.end(data)
    
    return await resultPromise
  }
)
```

### Example 3: File Upload/Download

```typescript theme={null}
import { createReadStream, createWriteStream } from 'fs'
import { pipeline } from 'stream/promises'

iii.registerFunction(
  { id: 'storage::upload' },
  async (input: { filename: string; data: ChannelReader }) => {
    const filepath = `/uploads/${input.filename}`
    await pipeline(
      input.data.stream,
      createWriteStream(filepath)
    )
    return { filepath, status: 'uploaded' }
  }
)

// Upload a file
const channel = await iii.createChannel()
const resultPromise = iii.call('storage::upload', {
  filename: 'report.pdf',
  data: channel.readerRef
})

const fileStream = createReadStream('./local-report.pdf')
await pipeline(fileStream, channel.writer.stream)

const result = await resultPromise
console.log(result)  // { filepath: '/uploads/report.pdf', status: 'uploaded' }
```

## Channel URL Format

Channels connect via dedicated WebSocket endpoints:

```typescript theme={null}
// Source: packages/node/iii/src/channels.ts:176-184
function buildChannelUrl(
  engineWsBase: string,
  channelId: string,
  accessKey: string,
  direction: 'read' | 'write'
): string {
  const base = engineWsBase.replace(/\/$/, '')
  return `${base}/ws/channels/${channelId}?key=${encodeURIComponent(accessKey)}&dir=${direction}`
}
```

**Example URL:**

```
ws://localhost:8080/ws/channels/a3f8b2c1-4d5e-6f7a-8b9c-0d1e2f3a4b5c?key=secret123&dir=read
```

## Error Handling

### Writer Errors

```typescript theme={null}
channel.writer.stream.on('error', (err) => {
  console.error('Write error:', err)
})

channel.writer.stream.write(data, (err) => {
  if (err) {
    console.error('Chunk write failed:', err)
  }
})
```

### Reader Errors

```typescript theme={null}
channel.reader.stream.on('error', (err) => {
  console.error('Read error:', err)
})

try {
  for await (const chunk of channel.reader.stream) {
    // Process chunk
  }
} catch (error) {
  console.error('Stream read failed:', error)
}
```

### Connection Errors

```typescript theme={null}
// Automatic reconnection on WebSocket errors
// Source: packages/node/iii/src/channels.ts:55-57
this.ws.on('error', err => {
  this.stream.destroy(err)
})
```

<Note>
  Channel WebSockets do not automatically reconnect. If the connection drops, the stream will emit an error and close.
</Note>

## Performance Considerations

### Frame Size

The 64KB frame size balances:

* **Smaller frames**: Lower latency, higher overhead
* **Larger frames**: Higher throughput, more memory

```typescript theme={null}
// Source: packages/node/iii/src/channels.ts:6
private static readonly FRAME_SIZE = 64 * 1024
```

### Backpressure Handling

Node.js streams automatically handle backpressure:

```typescript theme={null}
// Reader pauses when consumer is slow
if (!this.stream.push(data)) {
  this.ws?.pause()
}

// Writer blocks when buffer is full
stream.write(chunk, callback)  // Callback fires when buffer has space
```

### Buffer Size Tuning

Adjust channel buffer size based on use case:

```typescript theme={null}
// Low-latency (small buffer)
const channel = await iii.createChannel(16)

// High-throughput (large buffer)
const channel = await iii.createChannel(256)

// Default (balanced)
const channel = await iii.createChannel()  // 64 items
```

## Next Steps

<CardGroup cols={2}>
  <Card title="Streaming" icon="water" href="/concepts/streaming">
    Learn about the Stream API for real-time data
  </Card>

  <Card title="Functions" icon="function" href="/concepts/functions">
    Pass channels as function arguments
  </Card>
</CardGroup>