README.md

 ███████╗████████╗██████╗ ██╗   ██╗██╗  ██╗███████╗
 ██╔════╝╚══██╔══╝██╔══██╗╚██╗ ██╔╝██║ ██╔╝██╔════╝
 ███████╗   ██║   ██████╔╝ ╚████╔╝ █████╔╝ █████╗
 ╚════██║   ██║   ██╔══██╗  ╚██╔╝  ██╔═██╗ ██╔══╝
 ███████║   ██║   ██║  ██║   ██║   ██║  ██╗███████╗
 ╚══════╝   ╚═╝   ╚═╝  ╚═╝   ╚═╝   ╚═╝  ╚═╝╚══════╝
                    [ z m q ]

[ZEROMQ CLIENT FOR STRYKE // REQ/REP + PUB/SUB + PUSH/PULL + DEALER/ROUTER]

"Brokerless messaging, no daemon to babysit."

ZeroMQ client for stryke — the brokerless messaging library. Five patterns: the four canonical ones (request/reply, publish/subscribe, push/pull pipeline, dealer/router) plus PAIR, over TCP, IPC, or in-process transports. Opt-in package tier, kept out of the stryke core binary so the daily-driver install stays slim.

strykelang · MenkeTechnologiesMeta · stryke-kafka · stryke-redis · stryke-demo

Read the Docs · Engineering Report

---

Table of Contents

• [0x00] Why this is a package, not a builtin
• [0x01] Install
• [0x02] Quick start
• [0x03] Socket handles
• [0x04] API reference
• [0x05] FFI layer
• [0x06] Tests
• [0x07] Dev workflow
• [0x08] Layout
• [0x09] Roadmap
• [0xFF] License

---

[0x00] Why this is a package, not a builtin

ZeroMQ integration requires libzmq, the C library every ZMQ binding wraps. The Rust binding (zmq → zmq-sys → zeromq-src) compiles libzmq and libsodium from source and links them statically. The artifact is big enough that it doesn't belong in stryke core. Opt in once, get all five messaging patterns.

stryke-zmq ships a thin stryke library plus a Rust cdylib (libstryke_zmq.{dylib,so}) dlopened in-process. libzmq is vendored into the cdylib — there is no system libzmq.so/.dylib requirement on the consumer machine.

[0x01] Install

From a release (no rustc + cmake build on the consumer machine):

s pkg install -g github.com/MenkeTechnologies/stryke-zmq

From a local checkout:

cd ~/projects/stryke-zmq
cargo build --release            # first build vendors libzmq via cmake (~1-2 min)
s pkg install -g .               # cdylib lands in ~/.stryke/store/zmq@<version>/

Or:

make install

The cdylib is dlopened in-process on first use Zmq. A shared zmq::Context plus a socket-handle registry are held in OnceCell, so sockets persist across calls — a SUB keeps receiving and a REQ/REP conversation keeps its state between separate Zmq::send/Zmq::recv calls. A build needs cmake + a C/C++ compiler; the first build is the slow one, subsequent builds reuse the cached libzmq archive.

[0x02] Quick start

use Zmq

# REQ/REP over TCP
val $server = Zmq::socket("rep", bind => "tcp://*:5555")
val $client = Zmq::socket("req", connect => "tcp://localhost:5555")

Zmq::send($client, "ping")
val $request = Zmq::recv($server, timeout_ms => 1000)   # "ping"
Zmq::send($server, "pong")
val $reply = Zmq::recv($client, timeout_ms => 1000)     # "pong"

Zmq::close($client)
Zmq::close($server)

PUB/SUB with a topic filter:

val $pub = Zmq::socket("pub", bind => "tcp://*:5556")
val $sub = Zmq::socket("sub", connect => "tcp://localhost:5556", subscribe => "weather")

Zmq::send($pub, "weather sunny")
val $msg = Zmq::recv($sub, timeout_ms => 500)

One-shot request without managing a handle:

val $reply = Zmq::request("tcp://localhost:5555", "hello", timeout_ms => 2000)

[0x03] Socket handles

ZeroMQ sockets are stateful and long-lived, so the API is handle-based: Zmq::socket creates a socket, applies any bind/connect/options, and returns an integer handle. Every later operation takes that handle.

This is deliberate. A SUB socket must stay open to keep receiving; a fork-per-call model would drop messages between calls and re-pay connection setup each time. The cdylib holds the sockets in a registry behind a mutex (ZeroMQ requires one-thread-at-a-time access per socket; the mutex provides exactly that plus the cross-thread memory fence libzmq mandates).

Close sockets you no longer need with Zmq::close($handle). Dropping the process closes everything via libzmq's zmq_close on drop.

[0x04] API reference

Function	Returns	Notes
Zmq::version()	string	package (crate) version
Zmq::lib_version()	hashref	vendored libzmq { major, minor, patch, version }
Zmq::has($capability)	bool | undef	probe curve/gssapi/ipc/pgm/tipc/norm/draft
Zmq::capabilities()	hashref	probe every capability at once → { ipc, pgm, tipc, norm, curve, gssapi, draft } (bool | undef)
Zmq::io_threads($value?)	int	get (no arg) or set the shared context's ZMQ_IO_THREADS pool size (default 1, process-wide)
Zmq::socket($type, %opts)	handle (int)	type: req/rep/pub/sub/push/pull/dealer/router/pair/xpub/xsub/stream. opts: bind, connect, subscribe + any settable option (see set)
Zmq::socket_pair(%opts)	hashref	{ a, b, endpoint } — a connected inproc PAIR in one call (a = bound, b = connected); opts: endpoint (default a unique inproc name)
Zmq::socket_count()	hashref	{ count, handles } — how many sockets the cdylib holds open + their handles (leak detection); reads the local registry, no libzmq call
Zmq::close($handle)	hashref	closes + removes the socket
Zmq::bind($handle, $endpoint)	string	binds; returns the concrete endpoint (resolves tcp://*:*)
Zmq::connect($handle, $endpoint)	hashref	dynamic connect
Zmq::unbind($handle, $endpoint)	hashref	drop a bound endpoint
Zmq::disconnect($handle, $endpoint)	hashref	drop a connected endpoint
Zmq::send($handle, $data, %opts)	hashref	opts: more => 1; encoding => utf8|hex|base64
Zmq::send_multipart($handle, $parts, %opts)	hashref	$parts arrayref; opts: encoding
Zmq::recv($handle, %opts)	string | undef	opts: timeout_ms, encoding. undef on timeout
Zmq::recv_multipart($handle, %opts)	list	opts: timeout_ms, encoding. empty list on timeout
Zmq::recv_more($handle, %opts)	list	drain the trailing frames of the message a prior recv started (while RCVMORE is set); opts: timeout_ms, encoding
Zmq::drain($handle, %opts)	list	non-blocking drain — every whole multipart message already queued (each an arrayref of frames), stopping at the empty queue (ZMQ_DONTWAIT); opts: encoding, max; empty list when nothing queued
Zmq::sendrecv($handle, $data, %opts)	string | undef	send then recv one turn on a kept handle (REQ/DEALER/PAIR) — one round-trip without an ephemeral socket like request; opts: timeout_ms, encoding; undef on reply timeout
Zmq::subscribe($handle, $topic)	hashref	SUB topic filter ("" = all)
Zmq::unsubscribe($handle, $topic)	hashref	remove a subscription
Zmq::set($handle, $opt, $value)	hashref	full socket-option table: timeouts/buffers/hwm, tcp_keepalive*, heartbeat*, ipv6, immediate, conflate, router/req flags, CURVE keys, plain auth, identity…
Zmq::get($handle, $opt)	scalar	read back any option (type, last_endpoint, mechanism, fd, CURVE keys as z85, tcp_keepalive_cnt/idle/intvl, gssapi_principal/service_principal, …)
Zmq::poll($handle, %opts)	hashref	{ readable, writable }; opts: timeout_ms
Zmq::poll_many($handles, %opts)	list	one zmq_poll over many handles → { handle, readable, writable, error }
Zmq::events($handle)	hashref	{ readable, writable, events } — non-blocking readiness via ZMQ_EVENTS (zero-wait peek, not a zmq_poll call)
Zmq::monitor($handle, $endpoint, %opts)	hashref	publish lifecycle events to an inproc endpoint; opts: events
Zmq::monitor_recv($handle, %opts)	hashref | undef	decode one event { event, value, endpoint } from a monitor PAIR
Zmq::proxy($frontend, $backend, %opts)	hashref	backgrounded zmq_proxy device; opts: capture, control (steerable)
Zmq::curve_keypair()	hashref	{ public, secret } z85 keys (needs libsodium-enabled libzmq)
Zmq::curve_public($secret)	hashref	{ secret, public } — derive the public key from a z85 secret (zmq_curve_public)
Zmq::z85_encode($data, %opts) / Zmq::z85_decode($z85, %opts)	string	z85 codec; opts: encoding
Zmq::z85_valid($z85)	{ z85, valid, reason }	structural RFC-32 check: length ÷5 and Z85 alphabet (non-throwing predicate)
Zmq::valid_curve_key($key)	{ key, valid, reason }	a CURVE key is exactly 40 Z85 chars (32 bytes); stricter than z85_valid
Zmq::request($endpoint, $data, %opts)	string | undef	one-shot REQ round-trip; opts: timeout_ms (default 5000), encoding
Zmq::parse_endpoint($endpoint)	hashref	{ transport, address, known_transport, host?, port? } — no socket
Zmq::build_endpoint(%opts)	hashref	{ endpoint, known_transport } — inverse of parse_endpoint; opts: transport, address | host + port
Zmq::endpoint_bind_to_connect($endpoint, %opts)	hashref	{ endpoint, bind_endpoint, changed } — rewrite a wildcard bind host (*/0.0.0.0/::) to a connectable one (opts: host, default localhost)
Zmq::endpoint_connect_to_bind($endpoint, %opts)	hashref	{ endpoint, connect_endpoint, changed } — inverse: rewrite a concrete host to a bind wildcard (opts: host, default *)
Zmq::valid_endpoint($endpoint)	hashref	{ endpoint, valid, reason, transport } — strict transport-syntax check (stricter than parse_endpoint): known transport, host+port (* or 1-65535) for tcp/udp/ws/wss, non-empty path/name/address otherwise
Zmq::topic_match($subscription, $topic)	1 | ""	ZMQ SUB prefix match (empty subscription matches all)
Zmq::topic_overlaps($a, $b)	hashref	{ a, b, overlaps, subsumes } — true when one prefix subscription subsumes the other (prune redundant subs); subsumes is the broader (shorter) one
Zmq::prune_subscriptions(\@subscriptions)	hashref	{ pruned, removed } — reduce a subscription set to its minimal cover, dropping any subscription a strictly-shorter one subsumes (an empty "" subsumes everything); duplicates collapse, order preserved
Zmq::parse_monitor_event(event => $id, value => $v, endpoint => $ep)	hashref	{ event, name, is_error, value_meaning, value?, endpoint? } — decode a zmq_socket_monitor event: map the 16-bit ZMQ_EVENT_* id to its symbolic name, flag failures, and name what the 32-bit value carries (fd, errno, reconnect_interval_ms, protocol_error_code, zap_status_code, or none); value/endpoint pass through when supplied
Zmq::topic_match_any($topic, \@subscriptions)	{ topic, match, matched }	XPUB set routing: which subscriptions prefix-match the topic (in input order)
Zmq::build_subscription($topic, %opts)	{ frame, subscribe, topic }	SUB/XSUB wire frame: leading \x01 (subscribe; subscribe => 0 for \x00) + topic; pure stryke (control byte via chr)
Zmq::parse_subscription($frame)	{ subscribe, topic, frame }	inverse: split the leading 1/0 action byte from the topic
Zmq::valid_socket_type($type)	hashref	{ valid, canonical } — aliases collapse (publish → pub)
Zmq::socket_peers($type)	list	socket types $type can validly connect to (ZMQ messaging-pattern compatibility)
Zmq::socket_types_compatible($a, $b)	1 | ""	whether types $a and $b can be connected as peers (REQ↔REP, PUB↔SUB, PUSH↔PULL, …)
Zmq::socket_caps($type)	{ type, pattern, can_send, can_recv }	messaging pattern + send/recv directionality (PUB send-only, SUB recv-only, …)
Zmq::socket_caps_all()	list	the socket_caps table for every type in one call
Zmq::socket_types()	list	every canonical socket-type name
Zmq::socket_type_id($type)	int	the libzmq ZMQ_* integer for a type (PAIR=0…STREAM=11), for raw interop
Zmq::encoding_names()	list	every payload encoding accepted → { name, aliases, binary_safe }
Zmq::socket_event_names()	list	every monitor event → { name, flag, is_error } (the table behind parse_monitor_event)
Zmq::monitor_event_flag($name)	int	the ZMQ_EVENT_* flag for an event name (inverse of parse_monitor_event; accepts all)
Zmq::subscription_diff(\@current, \@desired)	hashref	{ add, remove, unchanged } — the subscribe/unsubscribe set difference to reconcile a SUB filter
Zmq::socket_option_names()	list	every option set/get accept → { name, settable, gettable, kind } (the table behind set/get)
Zmq::valid_socket_option($opt)	hashref	{ opt, known, settable, gettable, kind } — single-name lookup over the option table to guard a set/get; never errors on an unknown name

The pure helpers (validation, endpoint, topic, and table utilities) create no socket. Endpoints follow ZeroMQ's transport syntax: tcp://host:port, ipc:///tmp/sock, inproc://name, pgm://, epgm://.

Payloads default to UTF-8 framing. For arbitrary bytes (NULs, high bytes) pass encoding => "hex" or encoding => "base64" on both the send and the matching recv so the message round-trips intact.

[0x05] FFI layer

Each zmq__* export in the cdylib is a JSON-string-in / JSON-string-out function. stryke's FFI bridge resolves the symbols listed in stryke.toml's [ffi] table on first use Zmq, passes a JSON-encoded args dict per call, and copies the returned JSON into a stryke string. Allocations returned from the cdylib are freed via stryke_free_cstring.

A handler that errors returns {"error": "..."}, which the stryke wrapper turns into a die "Zmq::<op>: <reason>". A handler that panics is caught at the boundary and returned as an error rather than crossing the FFI boundary and aborting the host shell. A receive timeout returns {"timeout": true}, which the wrapper maps to undef/empty-list.

[0x06] Tests

cargo test                 # Rust unit + FFI-contract tests (run over inproc://)
s test t/                  # stryke assertion suite (needs the cdylib installed)

The Rust tests exercise the full socket lifecycle, push/pull and pair round-trips, the timeout-to-flag mapping, binary-safe hex framing through real sockets, set→get option round-trips, wildcard-bind endpoint discovery, poll_many over multiple handles, the z85 codec, libzmq introspection, an end-to-end socket-event monitor, and every argument-validation path over inproc:// transport — no external broker required. The stryke suite re-checks the same patterns through the wrapper plus a symbol-completeness pin over lib/Zmq.stk.

[0x07] Dev workflow

make release      # cargo build --release
make debug        # cargo build
make test         # cargo test + s test t/
make install      # s pkg install -g .
make clean        # cargo clean

[0x08] Layout

stryke-zmq/
  Cargo.toml             # cdylib crate (zmq -> vendored libzmq)
  src/lib.rs             # zmq__* exports + socket registry + tests
  stryke.toml            # package manifest + [ffi] table
  lib/Zmq.stk            # stryke wrapper (use Zmq)
  examples/              # req_rep, pub_sub, push_pull, multipart
  t/                     # stryke assertion suites
  tests/                 # docs/readme/polish lint gates
  docs/                  # GitHub Pages content
  Makefile

[0x09] Roadmap

Shipped: binary-safe payloads (encoding => hex|base64), multi-socket poll_many plus zero-wait events readiness, the full socket-option table (set/get), dynamic bind/connect/unbind/disconnect, socket-event monitoring (monitor/monitor_recv/socket_event_names/monitor_event_flag), a backgrounded proxy device, the z85 codec + curve_keypair (CURVE key auth on sockets via curve_* options; keypair generation needs a libsodium-enabled libzmq — probe Zmq::has("curve")), trailing-frame recv_more, one-call socket_pair, context io_threads tuning, batch capabilities, and the topology helpers (socket_type_id, socket_caps_all, subscription_diff, encoding_names).

Open:

• Streaming recv callback loop for long-running SUB/PULL consumers.
• A higher-level steerable-proxy control helper (PAUSE/RESUME/TERMINATE).

[0xFF] License

MIT. See LICENSE.