rho.async

rho.async is the small asynchronous contract at the base of Rho. A RhoTask represents one eventual value; a RhoStream represents an asynchronous sequence. Both are open S7 protocols, so transports and worker systems can add methods without changing callers.

Tasks and streams

library(rho.async)

tasks <- list(
  rho_task(20L),
  rho_then(rho_task(21L), function(value) value + 1L)
)
rho_await(rho_all(tasks), timeout = 1000)
#> [[1]]
#> [1] 20
#>
#> [[2]]
#> [1] 22

stream <- rho_list_stream(c("start", "delta", "done"))
rho_stream_collect(stream, timeout = 1000)
#> [[1]]
#> [1] "start"
#>
#> [[2]]
#> [1] "delta"
#>
#> [[3]]
#> [1] "done"

Composition stays asynchronous. rho_await() is deliberately conspicuous and is reserved for a CLI top level, a test, or an interactive request for the finished value. Cancellation and timeout operations use the same task contract; adapters own the mechanics of stopping their underlying handle.

Interactive R

An interactive session may opt into a top-level task callback that drains work already ready in the later loop after each completed R expression.

bridge <- rho_task_callback_bridge()
bridge <- rho_await(rho_register_task_callback(bridge))

# Remove it explicitly when the session no longer needs the bridge.
bridge <- rho_await(rho_remove_task_callback(bridge))

The callback never waits for work and is not a scheduler. nanonext remains the I/O substrate; the bridge only improves progress at interactive R safe points.

Continue with rho.http, or see the rho.async reference.