tur-nng

Scalability protocols for Turmeric via nng (nanomsg-next-generation) -- request/reply, publish/subscribe, pipeline, pair, bus, and survey, over inproc://, ipc://, tcp://, and ws://.

Overview

tur-nng is a cmake-dep spice that builds nng from source, statically, with TLS and the test/tool targets off. The surface is intentionally thin: nng runs its own worker threads and poller, so this spice is a set of blocking calls over one linear Socket handle, plus a length-prefixed Payload for binary messages. On top of that sit three opt-in layers: poll fds and non-blocking try-* calls for event loops, completion-based async (nng/aio), and per-conversation contexts (nng/ctx).

What you get over raw sockets is a messaging pattern rather than a byte stream: nng frames messages, reconnects dropped peers in the background, load-balances a pipeline across workers, and filters a subscription by topic -- none of which you write yourself.

The inproc:// transport connects two sockets inside one process with no network and no port, which is what keeps this spice's test suites network-free and CI-safe (the ws:// suite is the one exception, and it stays on loopback). It is also a real answer for in-process fan-out: the same code moves to tcp:// or ws:// by changing a string.

Install

:spices {
  "nng" {:url    "https://github.com/turmeric-lang/turmeric-spices"
         :ref    "nng-v0.1.0"
         :subdir "spices/nng"}
}

Nothing to install on the host: tur fetch clones and builds nng itself.

Quick start

A request/reply round trip over inproc://:

(import nng/socket :refer [req-open rep-open dial listen close
                           set-recv-timeout-ms])
(import nng/msg    :refer [send-str recv-str])

(let [rep (ok-val (rep-open))
      req (ok-val (req-open))]
  (listen rep "inproc://demo")
  (dial   req "inproc://demo")
  (set-recv-timeout-ms rep 2000)
  (set-recv-timeout-ms req 2000)

  (send-str req "ping")
  (let [asked (recv-str rep)]
    (when (ok? asked)
      (println (ok-val asked))       ;; "ping"
      (send-str rep "pong")))
  (let [heard (recv-str req)]
    (when (ok? heard)
      (println (ok-val heard))))     ;; "pong"

  (close req)
  (close rep))

#lang sweet-exp
import nng/socket :refer [push-open pull-open dial listen close set-recv-timeout-ms]
import nng/msg    :refer [send-str recv-str]

let [push ok-val(push-open())
     pull ok-val(pull-open())]
  listen(push "inproc://jobs")
  dial(pull "inproc://jobs")
  set-recv-timeout-ms(pull 2000)
  send-str(push "job:42")
  let [r recv-str(pull)]
    when ok?(r)
      println $ ok-val r
  close(pull)
  close(push)

Picking a protocol

Each constructor opens a socket speaking one protocol. A socket only talks to a peer speaking the matching half.

The Socket type is the same for all ten. Sending on a receive-only socket, or subscribing on a socket that is not a SUB socket, is a runtime err carrying NNG_ENOTSUP -- not a compile error. Per-protocol types are a deliberate non-goal for v0; see "Not in v0" below.

Addresses

listen binds and fails immediately if the address is taken; dial connects and, once connected, reconnects in the background if the peer goes away.

Port 0 (tcp://127.0.0.1:0, ws://127.0.0.1:0/jobs) asks the OS for a free port. listen binds it but cannot say which port that was; listen-bound-port does the same bind and returns the port. On inproc:// and ipc://, which have no port, it is (err NNG_ENOTSUP) and leaves nothing bound.

Timeouts

Set a receive timeout on anything that might not get an answer. Without one, a receive on a quiet socket parks the calling thread forever:

(set-recv-timeout-ms s 100)
(let [r (recv-str s)]
  (if (ok? r)
    (handle (ok-val r))
    (if (timed-out? (err-val r))
      (nothing-yet)                    ;; not a failure -- just nothing to read
      (println (err-str (err-val r))))))

timed-out? is the one error code worth branching on; err-str renders any of them via nng_strerror.

Payloads: cstr or Payload

send-str / recv-str are the convenience path and carry text. The message must not contain NUL bytes -- recv-str terminates the string it hands back, so an interior NUL truncates it.

send-payload / recv-payload are the real path. Payload is a heap block laid out as { int64 len; uint8 data[] }, so binary messages cross intact:

(import nng/payload :refer [payload-of-cstr payload-free payload->hex])
(import nng/msg :refer [send-payload recv-payload])

(let [b (payload-of-cstr "job:42")]
  (send-payload s b)
  (payload-free b))                        ;; send BORROWS; the Payload is still yours

(let [r (recv-payload s)]
  (when (ok? r)
    (println (payload->hex (ok-val r)))
    (payload-free (ok-val r))))            ;; recv gives you a fresh one to free

Ownership never involves nng's allocator. A receive asks nng for the message, copies it, and calls nng_free before returning -- so everything handed back is an ordinary heap block you release the ordinary way.

Why Payload and not Buf

The layout is deliberately identical to tur-msgpack's Buf (and to stdlib serial.tur's bytes value) -- the same { int64 len; uint8 data[] } block from the same malloc. The name differs because an opaque resolves globally: two spices that each define a Buf cannot both be loaded by one program, and msgpack-over-nng is the pairing both spices' plans were written for. tur-json 0.4.0 renamed its Encode / Decode classes for exactly this reason, so the msgpack cross-check program could exist at all.

With the names distinct, crossing between them is a copy against a documented layout rather than a reinterpret -- see tests/nng/msgpack_test.tur, which derives a codec for a Job struct, pushes it over inproc://, decodes it on the other side, and asserts the wire carries exactly the encoder's bytes:

(defstruct Job [id : int  kind : cstr  urgent : bool])
(derive-msgpack Job (id int) (kind cstr) (urgent bool))

(let [encoded (encode-mp job)          ;; msgpack Buf
      wire    (mp->nng encoded)]       ;; -> nng Payload (one copy)
  (send-payload push wire)
  (payload-free wire)
  (buf-free encoded))

Both mp->nng and its inverse live in that test file, not in either spice: they are ~6 lines of C against a layout both modules document, and they are what a stdlib Bytes would eventually replace.

Pub/sub and the slow joiner

A publisher does not queue for subscribers that are not connected yet, so the first few messages after a dial are routinely dropped. This is nng behaving correctly, not a bug in the spice or in your code. Publish on a loop, or re-publish and re-try until something lands:

(sub-subscribe sub "")                 ;; "" = everything; a SUB socket with
(set-recv-timeout-ms sub 50)           ;;      NO subscription receives nothing
;; then re-send on the pub side and re-try the receive until one arrives

Topic matching is a raw byte prefix, not a namespace: subscribing to "temp" also matches "temperature".

WebSocket (ws://)

ws:// is a transport like tcp://. Every protocol runs over it, and listen / dial take the address the same way they take any other. Reach for it when a peer is a browser, or when the only way through is an HTTP port.

(import nng/socket :refer [pub-open listen-bound-port close])
(import nng/msg    :refer [send-str])

(let [pub  (ok-val (pub-open))
      port (ok-val (listen-bound-port pub "ws://0.0.0.0:0/feed"))]
  (println port)                     ;; the port the OS picked
  (send-str pub "tick")
  (close pub))

#lang sweet-exp
import nng/socket :refer [pub-open listen-bound-port close]
import nng/msg    :refer [send-str]

let [pub  ok-val(pub-open())
     port ok-val(listen-bound-port(pub "ws://0.0.0.0:0/feed"))]
  println(port)
  send-str(pub "tick")
  close(pub)

• The path is part of the address. Sockets in one process can listen on the same host and port at different paths (ws://0.0.0.0:8080/jobs and ws://0.0.0.0:8080/feed). nng serves them all from one HTTP listener. A second listener on the same path is NNG_EADDRINUSE.
• No wss://. It needs nng's TLS support, which is off in this build, so a wss:// address is (err NNG_ENOTSUP). Put a TLS-terminating proxy in front if you need it.
• Nothing to install. nng carries its own HTTP server, SHA-1 and base64, so the transport adds no dependency. build.tur pins it on.

Plain WebSocket peers (browsers)

A browser has no nng, but it can still be a peer. nng's WebSocket transport is ordinary RFC 6455 with three rules on top:

1. Offer the listener's protocol as the subprotocol: <protocol>.sp.nanomsg.org, naming the nng socket you connect to. A PUB listener wants pub.sp.nanomsg.org. A client that offers nothing, or the wrong name, is refused with HTTP 400.
2. Send binary frames only. A text frame closes the connection and delivers nothing. In JavaScript, send a Uint8Array or ArrayBuffer, never a string.
3. Some protocols put a header in front of each message:

const ws = new WebSocket("ws://localhost:8080/feed", "pub.sp.nanomsg.org");
ws.binaryType = "arraybuffer";
ws.onmessage = (e) => console.log(new TextDecoder().decode(e.data));

tests/nng/ws_test.tur checks every row of that table byte for byte, with a small hand-written RFC 6455 client standing in for the browser.

Concurrency

nng's blocking calls park only the CALLING thread -- nng's internal workers keep running -- so a concurrent server is ordinary stdlib/thread.tur threads running ordinary receive loops, as in the tur-valkey spice.

One worker per socket needs nothing more. Several workers on ONE REP socket need a context each (see "Contexts" below): the socket itself has one receive state, so a second worker's receive fails with NNG_ESTATE. For many sockets on one thread, see the next section.

Event loops: poll fds and try-*

One thread can multiplex many sockets through stdlib/reactor.tur.

• The fd. recv-poll-fd / send-poll-fd (nng/socket) hand back a PollFd: the read end of a pipe nng raises while the socket is ready and clears once it drains. It is poll-only: never read, write or close it. poll-fd->int is the step to the plain fd the reactor takes.
• Moving messages. try-recv-str / try-recv-payload / try-send-str / try-send-payload (nng/msg) never block. "Nothing ready" is an ok result, (ok (none)) or (ok false). Every real failure is still an err.

;; drain -- every waiting message; never a blocking receive in a callback.
(defn drain [^borrow s : Socket n : int] : int
  (let [r (try-recv-str s)]
    (if (ok? r)
      (match (ok-val r)
        (Some m) (do (handle m) (drain s (+ n 1)))
        (None)   n)
      n)))

(let [fd (ok-val (recv-poll-fd sub))]
  (reactor-add-fd r (poll-fd->int fd) READ
    (fn [id events user] : nil (drain sub 0))
    (:: 0 :ptr<void>))
  (reactor-run r))

;; drain -- every waiting message; never a blocking receive in a callback.
defn drain [^borrow s : Socket n : int] : int
  let [r try-recv-str(s)]
    if ok?(r)
      (match (ok-val r)
        (Some m) (do (handle m) (drain s {n + 1}))
        (None)   n)
      n

let [fd ok-val(recv-poll-fd(sub))]
  reactor-add-fd r poll-fd->int(fd) READ
    (fn [id events user] : nil (drain sub 0))
    (:: 0 :ptr<void>)
  reactor-run r

Three rules, each learned the hard way:

• Drain to empty. The reactor's macOS backend (kqueue, EV_CLEAR) reports the fd once per empty-to-ready transition. A callback that takes one message per wakeup stalls with the rest still queued. On Linux (epoll) it happens to work either way.
• Never block in the callback. Use try-*. If another consumer drained the socket first, a blocking receive would park the whole loop.
• The send fd is READABLE when a send would not block. Register both fds for READ.

Some protocols have no fd in a direction: PUB and PUSH have no receive fd, and SUB and PULL have no send fd. Asking for one is an err carrying NNG_ENOTSUP. Poll fds and nng contexts must not be mixed on one socket.

Async: nng/aio

nng/aio is completion-based: submit an operation now and collect it later. An Aio is one operation slot, :linear and consumed by aio-free. Submit on it with send-payload-aio, send-str-aio or recv-aio. Each returns at once, (ok nil) when the operation was submitted. Then collect the outcome:

After a receive, aio-take-payload / aio-take-str hand back the message. N slots in flight on one thread are the fan-out.

(let [a (ok-val (aio-alloc))
      r (recv-aio pull a)
      w (aio-wait a)]
  (when (ok? w)
    (let [t (aio-take-str a)]
      (when (ok? t) (println (ok-val t)))))
  (aio-free a))

let [a ok-val(aio-alloc())
     r recv-aio(pull a)
     w aio-wait(a)]
  when ok?(w)
    let [t aio-take-str(a)]
      when ok?(t) println(ok-val(t))
  aio-free(a)

What the slot guarantees, which nng's own nng_aio does not:

• No nng message reaches your code or leaks. A send copies the payload in, so the caller still owns the Payload. A failed send's message is freed for you. An untaken receive is freed by the next submit or by aio-free.
• One operation at a time. A submit while one is in flight is (err NNG_EBUSY) and nothing is submitted. nng itself asserts here.
• Free is safe in flight. aio-free cancels and waits first.
• The completion fd is readable until you collect. Any collector that sees the result clears it, and so does the next submit. A reactor callback that collects therefore neither spins nor stalls. Remove the reactor source before aio-free closes the fd.

nng's completion callback runs on one of its worker threads. The spice's C callback handles the message and the fd there and never calls a Turmeric closure, because closures must not run off their own thread. By default an operation takes the socket's timeout as it stands when the operation is submitted; aio-set-timeout overrides that for the slot.

Contexts: nng/ctx

On a REQ or REP socket, the socket's own calls share one protocol state machine. On REQ a second request cancels the first; on REP a second pending receive fails with NNG_ESTATE. An NngCtx is an independent copy of that state on the socket, so N contexts are N conversations. Each serves its own receive-then-reply cycle:

(let [c (ok-val (ctx-open rep))
      r (ctx-recv-str c)]
  (when (ok? r)
    (println (ok-val r))
    (let [s (ctx-send-str c "pong")]
      (when (err? s) (println "reply failed"))))
  (ctx-close c))

let [c ok-val(ctx-open(rep))
     r ctx-recv-str(c)]
  when ok?(r)
    println(ok-val(r))
    let [s ctx-send-str(c "pong")]
      when err?(s) println("reply failed")
  ctx-close(c)

Contexts come in two flavours:

• Blocking (ctx-send-str, ctx-recv-str, ctx-send-payload, ctx-recv-payload): one thread per context. N worker threads, each with its own context, share one REP socket. A context is :linear and belongs to one thread at a time. The threaded server in tests/nng/ctx_test.tur opens each context on the main thread and moves it into its worker, which serves on it and closes it.
• Async (ctx-send-str-aio, ctx-send-payload-aio, ctx-recv-aio): the nng/aio contract on a context. One thread keeps many conversations in flight. Here two requests are out at once on ONE REQ socket, and each context gets the reply to its own request, whatever order REP answers in:

(let [s1 (ctx-send-str-aio q1 a1 "q1")
      s2 (ctx-send-str-aio q2 a2 "q2")
      w1 (aio-wait a1)
      w2 (aio-wait a2)
      r1 (ctx-recv-aio q1 a1)
      r2 (ctx-recv-aio q2 a2)
      v1 (aio-wait a1)
      v2 (aio-wait a2)]
  (aio-take-str a1))

The rules:

• Contexts exist on REQ, REP, SUB, SURVEYOR and RESPONDENT. ctx-open on PUB, PUSH, PULL, PAIR or BUS is (err NNG_ENOTSUP).
• Each SUB context filters on its own, with ctx-subscribe / ctx-unsubscribe. A context with no subscription receives nothing.
• A context copies its socket's timeouts at ctx-open. The ctx-set-recv-timeout-ms / ctx-set-send-timeout-ms calls change the context's own. An Aio on a context takes the context's timeout.
• A socket uses poll fds or contexts, never both.
• Close each context before its socket. The other order is harmless, since closing the socket reaps its contexts. Every call on a context in between is (err NNG_ECLOSED).

Linear Socket

Socket is a :linear opaque. A socket extracted with ok-val must be consumed exactly once, by close; every other operation takes it by ^borrow, observing it without discharging that obligation. Under -Xsubstructural that makes three mistakes compile-time errors rather than runtime faults:

Aio (nng/aio) follows the same discipline with aio-free as its consumer, and NngCtx (nng/ctx) with ctx-close.

The discipline is inert in ordinary builds, so call sites compile unchanged. errors/ holds one rejected fixture per row, and errors/run.sh asserts each one fails for its own reason.

Operations with no result value

dial, listen, sub-subscribe, and the timeout setters return (Result nil int). nil is the ok payload that says "it worked and carries nothing", so inspect these with ok? / err? alone -- there is no value to read out, and the type says so.

This is what the spice's plan specified from the start. It briefly shipped a (defopaque Ack :int) stand-in instead, because a nil ok payload emitted a void union member and the monomorph would not compile (report, fixed upstream). The alternative it avoided -- (Result int int) with an "ok carries 0" convention -- is the :int stand-in the house rules exist to prevent.

Not in v0

Deliberate omissions, each with a known seam:

• TLS, wss://, and ZeroTier transports. NNG_ENABLE_TLS=OFF keeps mbedTLS out of the build entirely, which rules out tls+tcp:// and wss://. Plain ws:// is in (see "WebSocket" above). tur-tls already vendors mbedTLS 3.6.2, so turning TLS on here wants coordinating with it rather than a second copy.
• Per-protocol socket types. Ten opaques would turn "receive on a PUB socket" into a compile error, which is the right end state -- but it multiplies every shared operation by ten or needs a typeclass over socket kinds, and that design deserves its own pass.
• Zero-copy nng_msg. One ownership model across the spice beats a faster path nothing needs yet.
• Explicit dialer/listener handles and their option tuning. dial and listen use nng's convenience forms, which tie the endpoint's lifetime to the socket's; listen-bound-port reads the one listener property tests need without handing the listener out. WebSocket tuning (extra request/response headers, frame sizes) would land here too, since nng deprecates setting transport options on the socket.

Testing

tur fetch --update      # clone + build nng (once)
tur test tests/nng      # 102 assertions in 9 suites; all over inproc:// except
                        # ws_test.tur, which uses ws:// and tcp:// on 127.0.0.1
errors/run.sh           # the nine compile-fail linear fixtures (Socket, Aio, NngCtx)

See also

• Guide
• API reference
• Source: https://github.com/turmeric-lang/turmeric-spices/tree/main/spices/nng
• tur-msgpack -- the binary codec whose buffer layout this spice shares
• tur-valkey -- the structural model for this spice