ZmqSharp

· · License: Apache-2.0

A fully asynchronous, AOT-compatible .NET implementation of the ZMTP 3.0 protocol (ZeroMQ wire protocol) with libzmq-compatible socket semantics.

• Value-type messages (ZMessage / ZFrame / ZSegment) with explicit ownership and move semantics — no per-message allocations on the hot path.
• All 11 libzmq socket types: PAIR, PUSH, PULL, PUB, SUB, REQ, REP, DEALER, ROUTER, XPUB, XSUB, each verified against the NetMQ (libzmq-compatible) implementation over TCP in both directions.
• TCP and ipc (Unix domain socket) transports with a single string-endpoint facade (tcp://host:port, ipc://path). libzmq supports ipc on Unix only, and NetMQ's ipc:// is a TCP hash rather than a Unix domain socket; ZmqSharp offers true ipc:// Unix domain sockets on every platform, including Windows — a concrete differentiator (0015 section 5.3, 0020 section 7/8).
• Per-peer bounded queues with configurable full modes (Wait, DropWrite, DropNewest, DropOldest) and mandatory reclamation of dropped messages.
• Copy-on-write peer snapshots: zero-allocation send and receive hot paths in optimized builds.
• Full Native AOT: no runtime reflection or dynamic code generation.

Usage

using ZmqSharp;

// PAIR over TCP.
await using var server = new ZPairSocket();
await using var client = new ZPairSocket();
await server.BindAsync("tcp://127.0.0.1:5555");
await client.ConnectAsync("tcp://127.0.0.1:5555");

await client.SendAsync("hello"u8.ToArray());
using var message = await server.Messages.ReadAsync();

// PAIR over ipc (Unix domain socket): an absolute path, a relative one
// resolved against the system temp directory, or - on Linux - an abstract
// namespace name (libzmq's ipc://@name convention) that creates no
// filesystem entry.
await using var ipcServer = new ZPairSocket();
await using var ipcClient = new ZPairSocket();
await ipcServer.BindAsync("ipc:///tmp/zmqsharp.sock");
await ipcClient.ConnectAsync("ipc:///tmp/zmqsharp.sock");

REQ/REP (operation-oriented):

await using var rep = new ZRepSocket();
await rep.BindAsync("tcp://127.0.0.1:5556");
rep.BindRequestHandler((context, token) =>
    rep.SendReplyAsync(context, ZMessage.FromOwned("pong"u8.ToArray()), token));

await using var req = new ZReqSocket();
await req.ConnectAsync("tcp://127.0.0.1:5556");
using var reply = await req.RequestAsync(ZMessage.FromOwned("ping"u8.ToArray()));

Endpoint shutdown can be awaited with the same registered string:

await client.DisconnectAsync("tcp://127.0.0.1:5555", cancellationToken);
await server.UnbindAsync("tcp://127.0.0.1:5555", cancellationToken);

These operations complete after cleanup. Unbind preserves accepted peers; cancellation after shutdown begins cancels only the caller's wait. Generic Disconnect/Unbind initiate shutdown without waiting. Request cancellation retires the selected peer and finishes after the send stops using borrowed data. Subscription prefixes are copied; changing the original array does not change filtering or reconnect subscriptions. Receive rejections can be inspected as ZReceiveRejectedException in PeerEnded.

See the stage-one migration guide for public API changes and custom CURVE backend requirements. The replacement-model migration guide covers byte-only transports, fixed security roles, frame codecs and ZPeer signatures.

Design

Design documents live in docs/impl/. An internal socket runtime owns a single peer registry. Byte transports provide I/O, ZMTP sessions serialize messages and own optional frame codecs, and pattern coordinators receive explicit routing/send capabilities. Surfaces are thin composition roots constructed directly (new ZPairSocket()), with the queue surface as the default receive path (Messages) and BindAsync/ConnectAsync as the repeatable endpoint surface (0022, 0023).

License

Apache-2.0 — see LICENSE. Third-party notices in NOTICE.