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

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

<ParamField path="bufferSize" type="number" default="64">
  Optional buffer size for the channel
</ParamField>

<ResponseField name="channel" type="Channel">
  Channel object with writer and reader

  <Expandable title="Channel properties">
    <ResponseField name="writer" type="ChannelWriter">
      Writer for sending data through the channel
    </ResponseField>

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

    <ResponseField name="writerRef" type="StreamChannelRef">
      Serializable writer reference to pass to other functions
    </ResponseField>

    <ResponseField name="readerRef" type="StreamChannelRef">
      Serializable reader reference to pass to other functions
    </ResponseField>
  </Expandable>
</ResponseField>

### Example: Basic Channel Usage

```typescript theme={null}
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

<ParamField path="stream" type="Writable">
  Node.js Writable stream for sending data

  ```typescript theme={null}
  // Write data
  writer.stream.write(Buffer.from('data'))

  // End stream
  writer.stream.end()

  // Write and end
  writer.stream.end(Buffer.from('final data'))
  ```
</ParamField>

### Methods

#### sendMessage()

Send a text message through the channel.

```typescript theme={null}
writer.sendMessage(msg: string): void
```

<ParamField path="msg" type="string" required>
  Text message to send
</ParamField>

#### close()

Close the channel writer.

```typescript theme={null}
writer.close(): void
```

### Example: Writing Data

```typescript theme={null}
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

<ParamField path="stream" type="Readable">
  Node.js Readable stream for receiving data

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

  // Pipe to another stream
  reader.stream.pipe(outputStream)
  ```
</ParamField>

### Methods

#### onMessage()

Register a callback for text messages.

```typescript theme={null}
reader.onMessage(callback: (msg: string) => void): void
```

<ParamField path="callback" type="(msg: string) => void" required>
  Function called for each text message received
</ParamField>

### Example: Reading Data

```typescript theme={null}
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

```typescript theme={null}
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

```typescript theme={null}
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

<AccordionGroup>
  <Accordion title="Always close writers">
    ```typescript theme={null}
    // 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()
      })
    })
    ```
  </Accordion>

  <Accordion title="Handle backpressure">
    ```typescript theme={null}
    // 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))
      }
    }
    ```
  </Accordion>

  <Accordion title="Process chunks incrementally">
    ```typescript theme={null}
    // 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)
    ```
  </Accordion>

  <Accordion title="Use appropriate buffer sizes">
    ```typescript theme={null}
    // Small messages - small buffer
    const channel = await iii.createChannel(16)

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

  <Accordion title="Handle errors properly">
    ```typescript theme={null}
    reader.stream.on('error', (err) => {
      console.error('Reader error:', err)
    })

    writer.stream.on('error', (err) => {
      console.error('Writer error:', err)
    })
    ```
  </Accordion>
</AccordionGroup>

## StreamChannelRef

Channel references can be serialized and passed to other functions.

```typescript theme={null}
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.