# Channel API

The Channel API provides efficient bidirectional message passing between Erlang and Python. Channels use `enif_ioq` for zero-copy buffering and integrate with Python's asyncio for non-blocking operations.

## Overview

Channels are faster than the Reactor pattern for message passing scenarios:

| Message Size | Channel | Reactor | Speedup |
|-------------|---------|---------|---------|
| 64 bytes    | 6.2M ops/s | 772K ops/s | **8x** |
| 1KB         | 3.8M ops/s | 734K ops/s | **5x** |
| 16KB        | 1.1M ops/s | 576K ops/s | **2x** |

Use channels when you need:
- High-throughput message streaming
- Bidirectional Erlang-Python communication
- Asyncio integration
- Backpressure support

## Quick Start

### Erlang Side

```erlang
%% Create a channel
{ok, Ch} = py_channel:new(),

%% Send messages with sender PID for replies
ok = py_channel:send(Ch, {request, self(), <<"data">>}),

%% Wait for response
receive
    {response, Result} ->
        io:format("Got result: ~p~n", [Result])
end,

%% Close when done
py_channel:close(Ch).
```

### Python Side (Sync)

```python
from erlang.channel import Channel, reply

def process_messages(channel_ref):
    ch = Channel(channel_ref)

    for msg in ch:
        # Extract sender PID from message
        _, sender_pid, data = msg

        # Process and reply
        result = process(data)
        reply(sender_pid, ('response', result))
```

### Python Side (Async)

```python
from erlang.channel import Channel, reply

async def process_messages(channel_ref):
    ch = Channel(channel_ref)

    async for msg in ch:
        # Extract sender PID from message
        _, sender_pid, data = msg

        # Process and reply
        result = await process(data)
        reply(sender_pid, ('response', result))
```

## Erlang API

### `py_channel:new/0,1`

Create a new channel.

```erlang
%% Unbounded channel
{ok, Ch} = py_channel:new().

%% Channel with backpressure (max 10KB queued)
{ok, Ch} = py_channel:new(#{max_size => 10000}).
```

**Options:**
- `max_size` - Maximum queue size in bytes. When exceeded, `send/2` returns `busy`.

### `py_channel:send/2`

Send an Erlang term to Python.

```erlang
ok = py_channel:send(Ch, Term).
```

**Returns:**
- `ok` - Message queued successfully
- `busy` - Queue full (backpressure)
- `{error, closed}` - Channel was closed

### `py_channel:close/1`

Close the channel. Python receivers will get `StopIteration`.

```erlang
ok = py_channel:close(Ch).
```

### `py_channel:info/1`

Get channel status.

```erlang
Info = py_channel:info(Ch).
%% #{size => 1024, max_size => 10000, closed => false}
```

## Python API

### `Channel` class

Wrapper for receiving messages from Erlang.

```python
from erlang.channel import Channel

ch = Channel(channel_ref)
```

#### `receive()`

Blocking receive. Blocks Python execution until a message is available.

```python
msg = ch.receive()  # Blocks until message available
```

**Behavior:**
- If the channel has data, returns immediately
- If empty, suspends the Erlang process via `receive`, releasing the dirty scheduler
- Other Erlang processes can run while waiting for data

**Raises:** `ChannelClosed` when the channel is closed.

#### `try_receive()`

Non-blocking receive. Returns immediately.

```python
msg = ch.try_receive()  # Returns None if empty
```

**Returns:** Message or `None` if empty.
**Raises:** `ChannelClosed` when the channel is closed.

#### `async_receive()`

Asyncio-compatible receive. Yields to other coroutines while waiting.

```python
msg = await ch.async_receive()
```

**Raises:** `ChannelClosed` when the channel is closed.

#### Iteration

```python
# Sync iteration
for msg in channel:
    process(msg)

# Async iteration
async for msg in channel:
    process(msg)
```

### `reply(pid, term)`

Send a message to an Erlang process.

```python
from erlang.channel import reply

# Reply to the sender
reply(sender_pid, {"status": "ok", "result": data})
```

### `ChannelClosed` exception

Raised when receiving from a closed channel.

```python
from erlang.channel import Channel, ChannelClosed

try:
    msg = ch.receive()
except ChannelClosed:
    print("Channel closed")
```

## Backpressure

Channels support backpressure to prevent unbounded memory growth.

### Erlang Side

```erlang
{ok, Ch} = py_channel:new(#{max_size => 10000}),

case py_channel:send(Ch, LargeData) of
    ok ->
        continue;
    busy ->
        %% Queue is full, wait before retrying
        timer:sleep(10),
        retry
end.
```

### Monitoring Queue Size

```erlang
#{size := Size, max_size := MaxSize} = py_channel:info(Ch),
Utilization = Size / MaxSize.
```

## Examples

### Request-Response Pattern

```erlang
%% Erlang: Send request, receive response
{ok, Ch} = py_channel:new(),
ok = py_channel:send(Ch, {request, self(), <<"compute">>}),
receive
    {response, Result} -> Result
end.
```

```python
from erlang.channel import Channel, reply

def handle_requests(channel_ref):
    ch = Channel(channel_ref)

    for msg in ch:
        if msg[0] == 'request':
            _, sender_pid, data = msg
            result = compute(data)
            reply(sender_pid, ('response', result))
```

### Streaming Data

```erlang
%% Erlang: Stream data to Python
{ok, Ch} = py_channel:new(),
lists:foreach(fun(Item) ->
    ok = py_channel:send(Ch, Item)
end, large_list()),
py_channel:close(Ch).
```

```python
async def process_stream(channel_ref):
    ch = Channel(channel_ref)
    results = []

    async for item in ch:
        results.append(process(item))

    return results
```

### Worker Pool Pattern

```erlang
%% Erlang: Distribute work across Python workers
{ok, Ch} = py_channel:new(#{max_size => 100000}),

%% Start multiple Python workers on the channel
[spawn_python_worker(Ch) || _ <- lists:seq(1, 4)],

%% Send work items
[py_channel:send(Ch, {work, Item}) || Item <- WorkItems],

%% Signal completion
py_channel:close(Ch).
```

```python
import asyncio
from erlang.channel import Channel

async def worker(channel_ref, worker_id):
    ch = Channel(channel_ref)

    async for msg in ch:
        if msg[0] == 'work':
            _, item = msg
            await process_item(item)
            print(f"Worker {worker_id} processed {item}")
```

## Performance Tips

1. **Use async iteration** for high-throughput scenarios - it allows other coroutines to run while waiting.

2. **Set appropriate `max_size`** to prevent memory issues while maintaining throughput.

3. **Batch messages** when possible - sending fewer larger messages is more efficient than many small ones.

4. **Avoid `try_receive` polling** - use blocking `receive()` or async `async_receive()` instead.

## Architecture

```
Erlang                              Python
──────                              ──────

py_channel:new() ─────────────────▶ Channel created

py_channel:send(Ch, Term)
       │
       ▼
  enif_term_to_binary()
       │
       ▼
  enif_ioq_enq_binary() ──────────▶ channel.receive()
                                          │
                                          ▼
                                    enif_ioq_peek()
                                          │
                                          ▼
                                    enif_binary_to_term()
                                          │
                                          ▼
                                    Python term

py_channel:close() ───────────────▶ StopIteration
```

## See Also

- [Reactor](reactor.md) - FD-based protocol handling for sockets
- [Asyncio](asyncio.md) - Erlang-native asyncio event loop
- [Getting Started](getting-started.md) - Basic usage guide