Channels

Stream data between functions with bidirectional channels

Channels enable efficient streaming of large datasets between functions without loading everything into memory.

createChannel()

Create a bidirectional channel for streaming data between workers.

const channel = await iii.createChannel(bufferSize?)

bufferSize · number · default: "64"

Optional buffer size for the channel

channel · Channel

Channel object with writer and reader

Show Channel properties

writer · ChannelWriter

Writer for sending data through the channel

reader · ChannelReader

Reader for receiving data from the channel

writerRef · StreamChannelRef

Serializable writer reference to pass to other functions

readerRef · StreamChannelRef

Serializable reader reference to pass to other functions

Example: Basic Channel Usage

import { init, type ChannelReader } from 'iii-sdk'

const iii = init('ws://localhost:49199')

// Receiver function
iii.registerFunction(
  { id: 'processor::consume' },
  async (input: { reader: ChannelReader }) => {
    const chunks: Buffer[] = []
    
    // Read from channel stream
    for await (const chunk of input.reader.stream) {
      chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk))
    }
    
    const data = Buffer.concat(chunks).toString('utf-8')
    return { size: data.length, data }
  }
)

// Sender function
iii.registerFunction(
  { id: 'producer::send' },
  async (input: { message: string }) => {
    const channel = await iii.createChannel()
    
    // Start processing in background
    const resultPromise = iii.call('processor::consume', {
      reader: channel.readerRef
    })
    
    // Write data to channel
    channel.writer.stream.write(Buffer.from(input.message))
    channel.writer.stream.end()
    
    // Wait for result
    return await resultPromise
  }
)

ChannelWriter

Write data to a channel for streaming to another function.

Properties

stream · Writable

Node.js Writable stream for sending data

// Write data
writer.stream.write(Buffer.from('data'))

// End stream
writer.stream.end()

// Write and end
writer.stream.end(Buffer.from('final data'))

Methods

sendMessage()

Send a text message through the channel.

writer.sendMessage(msg: string): void

msg · string · required

Text message to send

close()

Close the channel writer.

writer.close(): void

Example: Writing Data

import { init } from 'iii-sdk'

const iii = init('ws://localhost:49199')

iii.registerFunction(
  { id: 'writer::stream_data' },
  async () => {
    const channel = await iii.createChannel()
    
    // Write chunks
    const chunks = ['Hello', ' ', 'World', '!']
    for (const chunk of chunks) {
      channel.writer.stream.write(Buffer.from(chunk))
    }
    
    // Close the stream
    channel.writer.stream.end()
    
    return { writerRef: channel.writerRef }
  }
)

ChannelReader

Read data from a channel streamed by another function.

Properties

stream · Readable

Node.js Readable stream for receiving data

// Read chunks
for await (const chunk of reader.stream) {
  console.log('Received:', chunk)
}

// Pipe to another stream
reader.stream.pipe(outputStream)

Methods

onMessage()

Register a callback for text messages.

reader.onMessage(callback: (msg: string) => void): void

callback · (msg: string) => void · required

Function called for each text message received

Example: Reading Data

import { init, type ChannelReader } from 'iii-sdk'

const iii = init('ws://localhost:49199')

iii.registerFunction(
  { id: 'reader::consume' },
  async (input: { reader: ChannelReader }) => {
    const chunks: Buffer[] = []
    
    // Read all chunks
    for await (const chunk of input.reader.stream) {
      const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)
      chunks.push(buffer)
    }
    
    // Combine chunks
    const fullData = Buffer.concat(chunks)
    
    return {
      size: fullData.length,
      content: fullData.toString('utf-8')
    }
  }
)

Complete Examples

File Processing Pipeline

import { init, type ChannelReader, type ChannelWriter } from 'iii-sdk'
import * as fs from 'node:fs'
import { pipeline } from 'node:stream/promises'

const iii = init('ws://localhost:49199')

// Worker that processes file chunks
iii.registerFunction(
  { id: 'files::process' },
  async (input: { reader: ChannelReader; writer: ChannelWriter }) => {
    const { reader, writer } = input
    
    // Send progress messages
    let chunkCount = 0
    for await (const chunk of reader.stream) {
      chunkCount++
      
      // Process chunk (e.g., compress, encrypt)
      const processed = processChunk(chunk)
      
      // Write to output channel
      writer.stream.write(processed)
      
      // Send progress message
      writer.sendMessage(JSON.stringify({
        type: 'progress',
        chunks: chunkCount
      }))
    }
    
    writer.stream.end()
    return { chunks_processed: chunkCount }
  }
)

// Coordinator that manages the pipeline
iii.registerFunction(
  { id: 'files::upload' },
  async (input: { filepath: string }) => {
    const inputChannel = await iii.createChannel()
    const outputChannel = await iii.createChannel()
    
    // Listen for progress messages
    outputChannel.reader.onMessage((msg) => {
      const progress = JSON.parse(msg)
      console.log('Progress:', progress)
    })
    
    // Start processing
    const processPromise = iii.call('files::process', {
      reader: inputChannel.readerRef,
      writer: outputChannel.writerRef
    })
    
    // Stream file to input channel
    const fileStream = fs.createReadStream(input.filepath)
    await pipeline(fileStream, inputChannel.writer.stream)
    
    // Collect processed output
    const outputChunks: Buffer[] = []
    for await (const chunk of outputChannel.reader.stream) {
      outputChunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk))
    }
    
    // Wait for completion
    const result = await processPromise
    
    return {
      ...result,
      output_size: Buffer.concat(outputChunks).length
    }
  }
)

function processChunk(chunk: Buffer): Buffer {
  // Example processing: convert to uppercase
  return Buffer.from(chunk.toString('utf-8').toUpperCase())
}

Large Dataset Transfer

import { init, type ChannelReader } from 'iii-sdk'

const iii = init('ws://localhost:49199')

// Processor that analyzes streaming data
iii.registerFunction(
  { id: 'analytics::process' },
  async (input: { reader: ChannelReader }) => {
    let count = 0
    let sum = 0
    
    // Process chunks as they arrive
    for await (const chunk of input.reader.stream) {
      const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)
      const data = JSON.parse(buffer.toString('utf-8'))
      
      for (const item of data) {
        count++
        sum += item.value
      }
    }
    
    return {
      count,
      sum,
      average: count > 0 ? sum / count : 0
    }
  }
)

// Generator that sends large dataset
iii.registerFunction(
  { id: 'data::stream' },
  async (input: { batch_size: number; total: number }) => {
    const channel = await iii.createChannel()
    
    // Start processing in background
    const resultPromise = iii.call('analytics::process', {
      reader: channel.readerRef
    })
    
    // Generate and stream data in batches
    let sent = 0
    while (sent < input.total) {
      const batch = []
      for (let i = 0; i < input.batch_size && sent < input.total; i++, sent++) {
        batch.push({ id: sent, value: Math.random() * 100 })
      }
      
      // Send batch
      channel.writer.stream.write(
        Buffer.from(JSON.stringify(batch))
      )
    }
    
    // Close stream
    channel.writer.stream.end()
    
    // Wait for results
    return await resultPromise
  }
)

Best Practices

Always close writers

// Good
channel.writer.stream.end()

// Or use with promise
await new Promise<void>((resolve, reject) => {
  channel.writer.stream.end((err) => {
    if (err) reject(err)
    else resolve()
  })
})

Handle backpressure

// Check if buffer is full
for (const item of largeArray) {
  const canContinue = writer.stream.write(Buffer.from(item))
  
  if (!canContinue) {
    // Wait for drain event
    await new Promise(resolve => writer.stream.once('drain', resolve))
  }
}

Process chunks incrementally

// Good - process as data arrives
for await (const chunk of reader.stream) {
  processChunk(chunk)
}

// Avoid - loads everything into memory
const chunks: Buffer[] = []
for await (const chunk of reader.stream) {
  chunks.push(chunk)
}
const allData = Buffer.concat(chunks)

Use appropriate buffer sizes

// Small messages - small buffer
const channel = await iii.createChannel(16)

// Large files - large buffer
const channel = await iii.createChannel(256)

Handle errors properly

reader.stream.on('error', (err) => {
  console.error('Reader error:', err)
})

writer.stream.on('error', (err) => {
  console.error('Writer error:', err)
})

StreamChannelRef

Channel references can be serialized and passed to other functions.

interface StreamChannelRef {
  channel_id: string
  access_key: string
  direction: 'read' | 'write'
}

The III SDK automatically converts these references to ChannelReader or ChannelWriter instances when received by function handlers.