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:

1. Writer Endpoint

   Sends data to the channel. Multiple chunks can be written sequentially.

2. Engine Buffer

   Stores data temporarily (configurable buffer size, default 64 items)

3. Reader Endpoint

   Receives data from the channel. Supports Node.js Readable stream interface.

Creating Channels

Basic Channel Creation

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:

// 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:

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

const channel = await iii.createChannel(128)  // Buffer 128 items

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

Implementation:

// 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:

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:

// 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)
}

The 64KB frame size (ChannelWriter.FRAME_SIZE) balances WebSocket message overhead with memory efficiency.

Piping Streams

Pipe Node.js readable streams directly:

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):

channel.writer.sendMessage(JSON.stringify({
  type: 'progress',
  percent: 50
}))

Implementation:

// 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)
  })
}

Text messages and binary chunks use the same WebSocket but are distinguished by message type. Text messages don’t affect the binary stream.

Closing Channels

// Graceful close (after pending writes)
channel.writer.close()

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

ChannelWriter class structure:

// 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:

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

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

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:

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:

// 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:

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:

// 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
}

Channel references are resolved recursively in objects and arrays. Nested channel refs work automatically.

Real-World Examples

Example 1: Data Processing Pipeline

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

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

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:

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

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

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

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

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

Performance Considerations

Frame Size

The 64KB frame size balances:

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

// Source: packages/node/iii/src/channels.ts:6
private static readonly FRAME_SIZE = 64 * 1024

Backpressure Handling

Node.js streams automatically handle backpressure:

// 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:

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

Streaming

Learn about the Stream API for real-time data

Functions

Pass channels as function arguments