powpow/stream.nim — Raw-fd streaming for arbitrary file descriptors.

IoStream wraps a non-blocking fd and drives it from the powpow event loop:

let s = openStream(loop, pipeFd, onData = proc(s: IoStream, data: openArraybyte) = ..., onClose = proc(s: IoStream) = ...) discard s.write("hello") # direct write, buffers + flushes on Write s.pause() # read backpressure (blocks the pipe writer) s.resume() s.closeAfterWrite() # flush pending writes then EOF to the peer

This enables pipes, subprocess stdout, socketpair/UDS IPC and (on kqueue) regular-file fds. The write path mirrors Connection: direct writes are drained until EAGAIN and the remainder is buffered with an edge-triggered Write watch; reads drain fully per edge-triggered Read event.

onClose fires exactly once — from close, which is the single terminal transition, whether reached by peer EOF, a hard error, or an explicit call.

Platform notes:

POSIX only. The Windows IOCP backend can only register sockets, so this module compiles to nothing there (a dedicated pipe path is future work).
Regular files: kqueue (macOS/BSD) monitors them fine; epoll rejects regular-file fds (EPERM), so Linux file streaming belongs to the planned thread-pool work.
Writes to a closed peer return EPIPE and are treated as a fatal error (powpow ignores SIGPIPE globally), closing the stream.

Imports

loop, types, net/common, net/tcp

Types

IoStream = ref object fd*: int loop*: Loop ownsLoop*: bool state*: IoStreamState onClose*: proc (s: IoStream) {.closure.} data*: pointer ## user state, like Connection
IoStreamState = enum StreamOpen, StreamClosed

Consts

MaxIoStreamWriteBuffer = 33554432: Per-stream cap on the queued write buffer. A peer that stops reading must not make us accumulate an unbounded payload (slow-read DoS), the same guard Connection applies.

Procs

proc close(s: IoStream) {....raises: [Exception, KeyError], tags: [RootEffect], forbids: [].}
proc closeAfterWrite(s: IoStream) {....raises: [Exception, KeyError, OSError], tags: [RootEffect], forbids: [].}: Flush pending writes, then close the fd (delivering EOF to the peer).
proc getLoop(s: IoStream): Loop {.inline, ...raises: [], tags: [], forbids: [].}: The event loop driving this stream (owned or borrowed).
proc newStreamPair(loop: Loop; onData: proc (s: IoStream; data: openArray[byte]) {. closure.}; onClose: proc (s: IoStream) {.closure.} = nil; onData2: proc ( s: IoStream; data: openArray[byte]) {.closure.} = nil; onClose2: proc (s: IoStream) {.closure.} = nil): (IoStream, IoStream) {....raises: [NetError, KeyError, OSError], tags: [], forbids: [].}: Create a full-duplex socketpair; both ends are returned as IoStreams on loop. Usable as the plumbing for IPC and subprocess stdio.
proc newStreamPair(onData: proc (s: IoStream; data: openArray[byte]) {.closure.}; onClose: proc (s: IoStream) {.closure.} = nil; onData2: proc ( s: IoStream; data: openArray[byte]) {.closure.} = nil; onClose2: proc (s: IoStream) {.closure.} = nil): (IoStream, IoStream) {....raises: [OSError, Exception, CatchableError], tags: [TimeEffect, RootEffect], forbids: [].}: Create a full-duplex socketpair with its own internal event loop. Both ends share one owned loop; run/stop via getLoop. Closing either end tears the loop down (the surviving peer's fd is reclaimed by that teardown), so close both ends together and treat the first close as terminal for the pair.
proc openStream(fd: int; onData: proc (s: IoStream; data: openArray[byte]) {.closure.}; onClose: proc (s: IoStream) {.closure.} = nil): IoStream {. ...raises: [OSError, NetError, KeyError], tags: [TimeEffect], forbids: [].}: Open a stream with its own internal event loop. run drives it; close frees the fd and the owned loop.
proc openStream(loop: Loop; fd: int; onData: proc (s: IoStream; data: openArray[byte]) {.closure.}; onClose: proc (s: IoStream) {.closure.} = nil): IoStream {. ...raises: [NetError, KeyError, OSError], tags: [], forbids: [].}: Take ownership of fd and drive it from loop. The fd is put into non-blocking mode; onData fires per read chunk (edge-triggered drain), onClose fires once when the peer closes (EOF), the fd errors, or close is called.
proc pause(s: IoStream) {....raises: [KeyError, OSError], tags: [], forbids: [].}: Pause reading: no further onData deliveries until resume. A paused reader lets the kernel buffer fill, applying backpressure to the writer (the point of pipes). Pending writes still flush.
proc resume(s: IoStream) {....raises: [KeyError, OSError, Exception], tags: [RootEffect], forbids: [].}: Resume reading after pause; any data buffered meanwhile is delivered immediately.
proc run(s: IoStream) {....raises: [Exception, KeyError], tags: [TimeEffect, RootEffect], forbids: [].}: Block driving the stream's event loop until stop/close. When the stream owns its loop, the loop is freed on exit.
proc stop(s: IoStream) {.inline, ...raises: [Exception, KeyError], tags: [RootEffect], forbids: [].}: Stop the stream (use from a callback). Frees an owned loop via close semantics once run returns.
proc write(s: IoStream; data: openArray[byte]): int {. ...raises: [Exception, KeyError, OSError], tags: [RootEffect], forbids: [].}: Write data to the stream. Tries a direct write, draining until EAGAIN, then buffers the remainder and watches for writability (edge-triggered Write events only fire on a readiness transition, so the remainder must never sit buffered while the fd is writable). Returns data.len on acceptance (buffered or fully sent), -1 on a hard error (the stream is closed; onClose fired).
proc write(s: IoStream; data: string): int {. ...raises: [Exception, KeyError, OSError], tags: [RootEffect], forbids: [].}