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.
Types
IoStream = ref object fd*: int loop*: Loop 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 closeAfterWrite(s: IoStream) {....raises: [KeyError, Exception, OSError], tags: [RootEffect], forbids: [].}
- Flush pending writes, then close the fd (delivering EOF to the peer).
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 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 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).