---
title: "Channels"
format:
  html:
    toc: true
vignette: >
  %\VignetteIndexEntry{Channels}
  %\VignetteEngine{quarto::html}
  %\VignetteEncoding{UTF-8}
---

The main function is `mizu_channel()`.
It creates a two-way shared-memory channel, with one lock-free ring per direction, and spawns a child R process connected to the other end.
The child evaluates a quoted expression with `ch` bound to its side of the channel.
Only a short join token is passed to the child on its command line; all data crosses through the rings.

``` r
library(mizu)

ch <- mizu_channel(quote(
  repeat {
    x <- mizu_recv(ch, timeout = 30)
    if (inherits(x, "mizu_sentinel")) break
    mizu_send(ch, x * 2)
  }
))

mizu_send(ch, 21)
mizu_recv(ch, timeout = 5)
#> [1] 42
```

`mizu_send()` publishes a message to the peer, and it is visible as soon as the call returns.
`mizu_recv()` returns the next message.
`mizu_send_batch()` and `mizu_recv_batch()` move a whole list of messages in one call.
Use them at rates where the per-call overhead of R itself starts to matter.

A full ring, a timeout, an orderly close, and peer death are returned as class-tagged sentinel values, not errors.
A receive loop tests for them with `inherits(x, "mizu_sentinel")`, or with the specific classes `mizu_full`, `mizu_timeout`, `mizu_closed`, `mizu_peer_gone`.
`mizu_is_sentinel()` is the stricter test.
It returns `TRUE` only for the sentinel objects that mizu itself returned, since a class test cannot tell a terminal state from a look-alike payload (a sentinel forwarded as a message arrives as an ordinary copy).
No call needs an error handler.
If the process at the other end dies, receives first drain the messages that it already published, then report `mizu_peer_gone`.
`mizu_alive(ch)` reports at any time whether the peer still runs, without touching the rings.
`mizu_channel_info(ch)` returns a read-only snapshot of the channel’s state: ring geometry and watermarks, the free-list and open caches, and the notify signals sent.

`mizu_close()` performs an orderly shutdown.
It waits for the peer to finish reading its messages before it releases the shared resources:

``` r
mizu_close(ch)
```

If the peer’s code raises an error, the error crosses as data: a receive returns a `mizu_error_remote` condition object rather than signalling it.
`mizu_is_remote_error()` tests for one, `conditionMessage()` leads with the remote type, and `stop(x)` signals it in your own session:

``` r
ch <- mizu_channel(quote(stop("boom")))
x <- mizu_recv(ch, timeout = 5)
mizu_is_remote_error(x)
#> [1] TRUE
conditionMessage(x)
#> [1] "simpleError: boom"
mizu_close(ch)
```

The peer also prints the error to its standard error, which appears on your console by default.
The `launcher` argument controls how the peer is spawned.
The default `mizu_launcher()` takes `stdout` and `stderr` arguments, so `launcher = mizu_launcher(stderr = "peer.log")` writes the error to a file instead.
The same launcher spawns the R workers of `mizu_pool()`.

## Promises and handlers

In an event-driven program (a Shiny app, an `httpuv` server, anything built around an event loop), the channel can wait on a notify endpoint instead of blocking.
When the peer publishes, the endpoint wakes the event loop, and `later` delivers the message on the main thread.
There is no polling and no extra thread.
This needs the `later` package, and `promises` for the promise forms.

`mizu_recv_async()` returns a promise without blocking.
It resolves with the next message, the same object a blocking `mizu_recv()` would return, and rejects on close, peer death, or its own `timeout` with a classed error that carries the sentinel’s class name.
`mizu_recv_handler()` is the stream form.
It installs a function that the event loop calls with each burst of messages, and `batch = TRUE` delivers each burst as one list.
Delivery is per burst, not per message.
A promise costs tens of microseconds of R work against about a microsecond of transport, so use the batch handler for high message volumes.
Every main-thread drain is bounded per event-loop turn and re-scheduled, so a saturated ring cannot starve the loop.

``` r
library(promises)
ch <- mizu_channel(quote(mizu_send(ch, mizu_recv(ch))))
mizu_recv_async(ch) %...>% print()
mizu_send(ch, 42L)
while (!later::loop_empty()) later::run_now(1) # pump the loop until the resolution lands
#> [1] 42
mizu_close(ch)
```

A channel can have only one of the two consumer forms at a time.
Installing a handler while promises are pending, or requesting a promise while a handler is installed, is an error.