← PUB/SUB | ROUTER →

DEALER Socket

This chapter's contract-owning document — the DEALER socket spec owns the contract. This chapter shows that contract through language examples.

1. Overview

The DEALER socket is an asynchronous request socket. It sends to multiple peers using round-robin distribution and receives using fair-queue. There is no enforced send/recv ordering, enabling free asynchronous messaging.

Key characteristics: - Send: Round-robin -- cyclic distribution across connected peers - Receive: Fair-queue -- fair reception from all peers - No enforced send/recv ordering (asynchronous)

Valid socket combinations: DEALER ↔ ROUTER, DEALER ↔ DEALER

%%{init: {'flowchart': {'nodeSpacing': 32, 'rankSpacing': 40, 'padding': 8, 'wrappingWidth': 180}, 'themeVariables': {'fontSize': '18px'}}}%%
flowchart LR
    D1[DEALER 1] -->|round-robin| R[ROUTER]
    D2[DEALER 2] -->|round-robin| R

Concrete Scenario: 3 DEALERs Sending to 1 ROUTER

Three DEALER clients connect to a single ROUTER server. Each DEALER sends requests independently; the ROUTER receives them via fair-queue and distinguishes each sender by source_rid.

Sender	routing_id	Message	ROUTER sees
DEALER 1	D1	"buy AAPL 100"	source_rid=D1, data="buy AAPL 100"
DEALER 2	D2	"sell TSLA 50"	source_rid=D2, data="sell TSLA 50"
DEALER 3	D3	"buy MSFT 200"	source_rid=D3, data="buy MSFT 200"

The ROUTER replies to each DEALER using zlink_send_rid() with the corresponding source_rid. Because DEALER uses round-robin for outgoing connections, if a single DEALER connects to multiple ROUTERs, its messages cycle across them (msg1 -> ROUTER-A, msg2 -> ROUTER-B, ...).

2. Basic Usage

Creation and Connection

void *dealer = zlink_socket(ctx, ZLINK_SOCKET_DEALER);

/* Set routing_id (optional, used for identification by ROUTER) */
zlink_set_routing_id(dealer, "client-1", 8);

/* Connect to server */
zlink_connect(dealer, "tcp://127.0.0.1:5558");

Sending and Receiving Messages

/* Each call submits a complete single-part record, so requests may be
   sent consecutively without multipart ordering constraints. */
zlink_msg_t msg1, msg2, msg3;
zlink_msg_init_size(&msg1, 9);
memcpy(zlink_msg_data(&msg1), "request-1", 9);
zlink_send(dealer, &msg1, 1, ZLINK_SEND_FLAGS_NONE, NULL, NULL);

zlink_msg_init_size(&msg2, 9);
memcpy(zlink_msg_data(&msg2), "request-2", 9);
zlink_send(dealer, &msg2, 1, ZLINK_SEND_FLAGS_NONE, NULL, NULL);

zlink_msg_init_size(&msg3, 9);
memcpy(zlink_msg_data(&msg3), "request-3", 9);
zlink_send(dealer, &msg3, 1, ZLINK_SEND_FLAGS_NONE, NULL, NULL);

/* Ordinary DATA is drained with zlink_recv(). Results for
   zlink_request() are drained with zlink_completion_recv(). */

Receive Modes

Use zlink_recv() to receive one complete record synchronously. source_rid_out_ is optional for DEALER; pass NULL when the caller does not need it (DEALER returns NULL there in any case).

zlink_msg_t parts[8];
size_t part_count = 0;
zlink_recv_result_t rc = zlink_recv(
    dealer, NULL, parts, 8, &part_count, ZLINK_RECV_FLAGS_NONE);
if (rc == ZLINK_RECV_OK) {
    /* Process parts[0..part_count), then close the complete array. */
    zlink_multipart_close(parts, part_count);
}
/* other rc values: ZLINK_RECV_NO_DATA (EAGAIN), TERMINATED, INVALID_HANDLE */

When HWM is reached, zlink_send() blocks (default) or returns ZLINK_SUBMIT_BACKPRESSURED with ZLINK_SEND_FLAGS_DONTWAIT. For advanced backpressure patterns, see Performance Guide.

3. Usage Example

/* DEALER → ROUTER send: one two-part record in one array. */
zlink_msg_t parts[2];
zlink_msg_init_size(&parts[0], 6);
memcpy(zlink_msg_data(&parts[0]), "header", 6);
zlink_msg_init_size(&parts[1], 4);
memcpy(zlink_msg_data(&parts[1]), "body", 4);

zlink_submit_result_t rc = zlink_send(
    dealer, parts, 2, ZLINK_SEND_FLAGS_NONE, NULL, NULL);

4. Socket Options

Option	Type	Default	Description
zlink_set_routing_id()	binary	Auto (UUID)	ID for identification by ROUTER (dedicated function)
ZLINK_DEALER_OPT_PROBE	int	0	Send empty message on connect (connection notification)
ZLINK_DEALER_OPT_REQUEST_TIMEOUT_MS	int (ms)	5000	Default timeout used when a request's timeout_ms_ == 0
ZLINK_DEALER_OPT_WEIGHT	int	100	Per-peer load-balancing weight for outgoing round-robin
ZLINK_OPT_SNDHWM	uint64_t bytes	automatic	Auto-HWM sized for DEALER's peer-queue role. Manual settings take precedence; 0 is unlimited
ZLINK_OPT_RCVHWM	uint64_t bytes	automatic	Auto-HWM sized for DEALER's peer-queue role. Manual settings take precedence; 0 is unlimited
ZLINK_OPT_LINGER	int	-1	Wait time on close (ms)
ZLINK_OPT_SNDTIMEO	int	1000	Send timeout (ms); set -1 explicitly for infinite wait
ZLINK_OPT_RCVTIMEO	int	1000	Receive timeout (ms); set -1 explicitly for infinite wait

Setting routing_id

To allow ROUTER to identify a DEALER, explicitly set the routing_id.

/* Set before bind/connect */
zlink_set_routing_id(dealer, "D1", 2);
zlink_connect(dealer, "tcp://127.0.0.1:5558");

Reference: core/tests/integration/test_router_multiple_dealers.cpp -- zlink_set_routing_id(dealer1, "D1", 2)

4.1 Request-Reply

When DEALER needs a correlated reply, use zlink_request() instead of ordinary DATA send/recv. It attaches a ZMP request-reply envelope and places the reply or terminal result in the socket completion queue.

For the ZMP request-reply envelope wire format, see ZMP Protocol.

int timeout_ms = 1000;
zlink_set_dealer_option(
  dealer,
  ZLINK_DEALER_OPT_REQUEST_TIMEOUT_MS,
  &timeout_ms,
  sizeof(timeout_ms));

zlink_msg_t req;
zlink_msg_init_size(&req, 4);
memcpy(zlink_msg_data(&req), "ping", 4);
zlink_completion_id_t id = 0;
zlink_submit_result_t rc = zlink_request(
    dealer, NULL, &req, 1, ZLINK_SEND_FLAGS_NONE,
    0 /* uses ZLINK_DEALER_OPT_REQUEST_TIMEOUT_MS */, NULL, &id);
if (rc == ZLINK_SUBMIT_OK) {
    zlink_completion_t completion = {0};
    completion.struct_size = sizeof(completion);
    if (zlink_completion_recv(dealer, &completion, ZLINK_RECV_FLAGS_NONE)
        == ZLINK_RECV_OK) {
        /* completion.request_result is OK, TIMED_OUT, NOT_FOUND, ... */
        zlink_completion_close(&completion);
    }
}

When timeout_ms_ == 0 is passed to zlink_request(), DEALER uses the ZLINK_DEALER_OPT_REQUEST_TIMEOUT_MS socket default (5000ms unless set otherwise).

5. Usage Patterns

Pattern 1: DEALER → ROUTER Request-Reply

The most basic pattern. DEALER sends an ordinary raw message, ROUTER distinguishes the sender by source_rid and replies to that same id.

void *router = zlink_socket(ctx, ZLINK_SOCKET_ROUTER);
zlink_bind(router, "tcp://*:5558");

void *dealer = zlink_socket(ctx, ZLINK_SOCKET_DEALER);
zlink_set_routing_id(dealer, "D1", 2);
zlink_connect(dealer, "tcp://127.0.0.1:5558");

/* Client typed request */
zlink_msg_t req;
zlink_msg_init_size(&req, 5);
memcpy(zlink_msg_data(&req), "Hello", 5);
zlink_completion_id_t request_id = 0;
zlink_request(dealer, NULL, &req, 1, ZLINK_SEND_FLAGS_NONE,
              0, NULL, &request_id);

/* Server: receive REQUEST with source_rid + opaque reply_token. */
const zlink_routing_id_t *source_rid = NULL;
zlink_reply_token_t reply_token = 0;
zlink_msg_t parts[8];
size_t part_count = 0;
zlink_router_recv(router, &source_rid, &reply_token,
                  parts, 8, &part_count, ZLINK_RECV_FLAGS_NONE);
printf("Received from [%.*s]: %.*s\n", (int)source_rid->size, source_rid->data,
       (int)zlink_msg_size(&parts[0]), (char *)zlink_msg_data(&parts[0]));

/* A nonzero token marks REQUEST and must be returned unchanged. */
zlink_msg_t reply;
zlink_msg_init_size(&reply, 5);
memcpy(zlink_msg_data(&reply), "World", 5);
zlink_reply(router, source_rid, reply_token, &reply, 1);
zlink_multipart_close(parts, part_count);

/* Client: receive the reply as REQUEST completion, not DATA. */
zlink_completion_t completion = {0};
completion.struct_size = sizeof(completion);
zlink_completion_recv(dealer, &completion, ZLINK_RECV_FLAGS_NONE);
zlink_completion_close(&completion);

Reference: core/tests/integration/test_router_multiple_dealers.cpp -- TCP/IPC/inproc examples

Pattern 2: Multiple DEALER Load Balancing

Multiple DEALERs connect to a single ROUTER. ROUTER distinguishes each DEALER by routing_id.

void *router = zlink_socket(ctx, ZLINK_SOCKET_ROUTER);
zlink_bind(router, "tcp://127.0.0.1:*");

char endpoint[256];
size_t len = sizeof(endpoint);
zlink_get_option(router, ZLINK_OPT_LAST_ENDPOINT, endpoint, &len);

void *dealer1 = zlink_socket(ctx, ZLINK_SOCKET_DEALER);
zlink_set_routing_id(dealer1, "D1", 2);
zlink_connect(dealer1, endpoint);

void *dealer2 = zlink_socket(ctx, ZLINK_SOCKET_DEALER);
zlink_set_routing_id(dealer2, "D2", 2);
zlink_connect(dealer2, endpoint);

/* Each DEALER sends a message */
zlink_msg_t m1;
zlink_msg_init_size(&m1, 12);
memcpy(zlink_msg_data(&m1), "from_dealer1", 12);
zlink_send(dealer1, &m1, 1, ZLINK_SEND_FLAGS_NONE, NULL, NULL);

zlink_msg_t m2;
zlink_msg_init_size(&m2, 12);
memcpy(zlink_msg_data(&m2), "from_dealer2", 12);
zlink_send(dealer2, &m2, 1, ZLINK_SEND_FLAGS_NONE, NULL, NULL);

/* zlink_router_recv() distinguishes each DEALER's message by
   its source_rid */

Reference: core/tests/integration/test_router_multiple_dealers.cpp -- test_router_multiple_dealers_tcp()

Pattern 3: Proxy Pattern (ROUTER-DEALER)

Build a multi-threaded server using ROUTER (frontend) + DEALER (backend). zlink_proxy() forwards multipart frames bidirectionally between the two raw sockets, including the client's routing-id envelope, so worker code never calls a routing-id-addressed send -- it just relays the frames it receives.

/* Frontend: clients connect here */
void *frontend = zlink_socket(ctx, ZLINK_SOCKET_ROUTER);
zlink_bind(frontend, "tcp://*:5558");

/* Backend: worker threads connect here */
void *backend = zlink_socket(ctx, ZLINK_SOCKET_DEALER);
zlink_bind(backend, "inproc://backend");

/* Start worker threads then run the proxy loop (blocks the caller) */
zlink_proxy(frontend, backend, NULL);

/* Worker thread: a plain DEALER connected to the backend. The proxy
   presents the client's envelope and payload as one record, so the worker
   receives the complete array and echoes the same envelope ahead of its
   reply. No target routing id is passed explicitly. */
void worker_thread(void *arg) {
    void *worker = zlink_socket(ctx, ZLINK_SOCKET_DEALER);
    zlink_connect(worker, "inproc://backend");

    zlink_msg_t parts[2];
    size_t part_count = 0;
    zlink_recv(worker, NULL, parts, 2, &part_count, ZLINK_RECV_FLAGS_NONE);
    /* parts[0] is the envelope and parts[1] is the request payload. */

    /* Process request, then reply: re-send the envelope, then the
       reply payload */
    zlink_msg_close(&parts[1]);
    zlink_msg_init_size(&parts[1], 5);
    memcpy(zlink_msg_data(&parts[1]), "World", 5);
    zlink_send(worker, parts, 2, ZLINK_SEND_FLAGS_NONE, NULL, NULL);

    /* Worker stays alive until socket is closed */
}

Reference: core/tests/integration/test_proxy.cpp -- ROUTER(frontend) + DEALER(backend) + worker pool

Pattern 4: DEALER ↔ DEALER Asynchronous Communication

Both sides use DEALER for fully asynchronous P2P communication.

void *a = zlink_socket(ctx, ZLINK_SOCKET_DEALER);
zlink_bind(a, "tcp://*:5558");

void *b = zlink_socket(ctx, ZLINK_SOCKET_DEALER);
zlink_connect(b, "tcp://127.0.0.1:5558");

/* Bidirectional free send */
zlink_msg_t ping;
zlink_msg_init_size(&ping, 4);
memcpy(zlink_msg_data(&ping), "ping", 4);
zlink_send(a, &ping, 1, ZLINK_SEND_FLAGS_NONE, NULL, NULL);

zlink_msg_t pong;
zlink_msg_init_size(&pong, 4);
memcpy(zlink_msg_data(&pong), "pong", 4);
zlink_send(b, &pong, 1, ZLINK_SEND_FLAGS_NONE, NULL, NULL);

/* b receives "ping" and a receives "pong" via zlink_recv() */

6. Caveats

No Peer Connected vs. HWM Backpressure

These are two distinct results. When no peer is connected (no positive-weight pipe), a send returns ZLINK_SUBMIT_NOT_ADMITTED — the message is not queued. When a peer is connected but its send queue has reached the HWM, the call blocks (default) or returns ZLINK_SUBMIT_BACKPRESSURED with ZLINK_SEND_FLAGS_DONTWAIT.

/* Send with no peer connected */
zlink_msg_t msg;
zlink_msg_init_size(&msg, 4);
memcpy(zlink_msg_data(&msg), "data", 4);
zlink_completion_id_t wait_token = 0;   /* identifies the WRITABLE record when BACKPRESSURED */
zlink_submit_result_t rc = zlink_send(
    dealer, &msg, 1, ZLINK_SEND_FLAGS_DONTWAIT, NULL, &wait_token);
if (rc == ZLINK_SUBMIT_NOT_ADMITTED) {
    /* No connected peer to admit the message */
} else if (rc == ZLINK_SUBMIT_BACKPRESSURED) {
    /* A peer is connected but its queue is at the HWM — the part was consumed, so
       keep a copy, wait for ZLINK_POLLCOMPLETION, read wait_token's WRITABLE record
       with zlink_completion_recv(), then resubmit */
}

Round-Robin Distribution

When multiple peers are connected, messages are distributed in a round-robin fashion. To send to a specific peer, use ROUTER instead.

Weight-Aware Outbound Selection

Remote peers advertise a weight (0..10000). DEALER automatically drops weight-0 peers from its candidate set. Positive peers remain eligible, and unequal positive weights change the send ratio. The underlying connections stay alive, so a peer that flips back to a positive weight rejoins the rotation without reconnect.

If every known peer is 0, zlink_send() and zlink_request() return ZLINK_SUBMIT_NOT_ADMITTED. The caller should wait for at least one peer to return to a positive weight before retrying; treating NOT_ADMITTED as a hard failure would discard messages that are expected to succeed once maintenance ends.

For the full contract, see Weight-aware outbound selection in the DEALER spec.

Set routing_id Before connect

zlink_set_routing_id() must be called before zlink_connect(). Changes after connection are not applied.

/* Correct order */
zlink_set_routing_id(dealer, "D1", 2);
zlink_connect(dealer, endpoint);  /* identified as D1 */

---

← PUB/SUB | ROUTER →

Full language examples

C++C#/.NETJavaKotlinPythonNode/TypeScriptJavaScriptGoRust

zlink::context_t ctx;
zlink::router_socket_t router (ctx);
zlink::dealer_socket_t dealer (ctx);
zlink::socket_monitor_t router_monitor = router.monitor_open ();
zlink::socket_monitor_t dealer_monitor = dealer.monitor_open ();

router.bind ("tcp://127.0.0.1:0");
const std::string endpoint = router.options ().last_endpoint ();
assert (!endpoint.empty ());
dealer.connect (endpoint);
assert (detail::wait_connected (router_monitor, dealer_monitor, 2000, &router));

const std::string sent = detail::k_dealer_router_request;
zlink::message_t outbound = detail::make_message (sent);
dealer.send ().message (std::move (outbound)).submit ();

zlink::received_t inbound;
assert (router.recv (inbound) == 0);
assert (inbound.routing_id ().has_value ());
assert (inbound.parts ().size () == 1);
assert (inbound.parts ()[0].to_string () == detail::k_dealer_router_request);

const std::string reply_payload = detail::k_dealer_router_reply;
zlink::message_t reply = detail::make_message (reply_payload);
// Reply on the ROUTER socket, addressed by the received envelope's routing id.
router.send (*inbound.routing_id ()).message (std::move (reply)).submit ();
inbound.close ();

zlink::received_t echoed;
assert (dealer.recv (echoed) == 0);
assert (echoed.parts ().size () == 1);
const std::string received = echoed.parts ()[0].to_string ();
assert (received == detail::k_dealer_router_reply);
echoed.close ();
std::printf ("[dealer-router/recv] send: \"%s\" → recv: \"%s\"\n", sent.c_str (),
             received.c_str ());
return 0;

if (!SampleSupport.IsNativeAvailable())
    return;

using var ctx = Zlink.CreateContext();
using var dealer = ctx.CreateDealerSocket();
using var router = ctx.CreateRouterSocket();
string endpoint = SampleSupport.NewEndpoint("tcp", "sample");
using var dealerMonitor = dealer.MonitorOpen(SocketEvent.ConnectionReady);
using var routerMonitor = router.MonitorOpen(SocketEvent.ConnectionReady);
router.Bind(endpoint);
dealer.Connect(endpoint);
SampleSupport.WaitConnected(routerMonitor, dealerMonitor);

using (Message request = Message.From("ping"))
    await dealer.Send().Message(request).Async().Admitted;
using var received = Received.Create();
if (!router.Recv(received))
    throw new InvalidOperationException("recv failed");
string requestPayload = received.Parts[0].GetString();
SampleSupport.EnsureEqual("ping", requestPayload, "request");

using var reply = Message.From("pong");
received.Send().Message(reply).Submit();
string payload = SampleSupport.ReceiveUtf8(dealer, 2000);
Console.WriteLine($"[dealer-router/recv] send: \"ping\" -> recv: \"{payload}\"");

SampleSupport.ensureNative();
String endpoint = SampleSupport.tcpEndpoint();

try (Context ctx = Zlink.createContext();
     RouterSocket router = ctx.createRouterSocket();
     DealerSocket dealer = ctx.createDealerSocket();
     var routerMonitor = router.monitorOpen(
         systems.zlink.contracts.eventing.MonitorEventType.CONNECTION_READY);
     var dealerMonitor = dealer.monitorOpen(
         systems.zlink.contracts.eventing.MonitorEventType.CONNECTION_READY)) {
    router.bind(endpoint);
    dealer.connect(endpoint);
    SampleSupport.waitConnected(routerMonitor, dealerMonitor);

    try (Message request = Message.from(SampleSupport.DEALER_REQUEST)) {
        dealer.send().message(request).submit().admitted().toCompletableFuture().join();
    }

    try (systems.zlink.contracts.messaging.Received received = new systems.zlink.contracts.messaging.Received()) {
        router.recv(received, systems.zlink.contracts.sockets.RecvFlags.NONE);
        String value = SampleSupport.singleUtf8(received);
        if (!SampleSupport.DEALER_REQUEST.equals(value)) {
            throw new IllegalStateException("unexpected request: " + value);
        }
        try (Message reply = Message.from(SampleSupport.DEALER_REPLY)) {
            received.send().message(reply).submit();
        }
    }

    try (systems.zlink.contracts.messaging.Received received = new systems.zlink.contracts.messaging.Received()) {
        dealer.recv(received, systems.zlink.contracts.sockets.RecvFlags.NONE);
        String value = SampleSupport.singleUtf8(received);
        if (!SampleSupport.DEALER_REPLY.equals(value)) {
            throw new IllegalStateException("unexpected reply: " + value);
        }
        System.out.println("[dealer-router/recv] send: \""
            + SampleSupport.DEALER_REQUEST + "\" \u2192 recv: \"" + value + "\"");
    }
}

SampleSupport.ensureNative()
val endpoint = SampleSupport.tcpEndpoint()

Zlink.createContext().use { ctx ->
    ctx.createRouterSocket().use { router ->
        ctx.createDealerSocket().use { dealer ->
            router.monitorOpen(MonitorEventType.CONNECTION_READY).use { routerMonitor ->
                dealer.monitorOpen(MonitorEventType.CONNECTION_READY).use { dealerMonitor ->
                    router.bind(endpoint)
                    dealer.connect(endpoint)
                    SampleSupport.waitConnected(routerMonitor, dealerMonitor)

                    Message.from(SampleSupport.DEALER_REQUEST).use { request ->
                        dealer.send().message(request).submit().await()
                    }

                    Received().use { received ->
                        router.recv(received, RecvFlags.NONE)
                        val value = SampleSupport.singleUtf8(received)
                        check(SampleSupport.DEALER_REQUEST == value) { "unexpected request: $value" }
                        Message.from(SampleSupport.DEALER_REPLY).use { reply ->
                            received.send().message(reply).submit()
                        }
                    }

                    Received().use { received ->
                        dealer.recv(received, RecvFlags.NONE)
                        val value = SampleSupport.singleUtf8(received)
                        check(SampleSupport.DEALER_REPLY == value) { "unexpected reply: $value" }
                        println(
                            "[dealer-router/recv] send: \"${SampleSupport.DEALER_REQUEST}\"" +
                                " → recv: \"$value\""
                        )
                    }
                }
            }
        }
    }
}

_, endpoint = tcp_endpoint()

with zlink.create_context() as ctx:
    with zlink.create_router_socket(ctx) as router:
        with zlink.create_dealer_socket(ctx) as dealer:
            with router.monitor_open(zlink.MonitorEventMask.CONNECTION_READY) as router_monitor:
                with dealer.monitor_open(zlink.MonitorEventMask.CONNECTION_READY) as dealer_monitor:
                    dealer.set_routing_id(b"CLIENT")
                    router.bind(endpoint)
                    dealer.connect(endpoint)
                    wait_connected(router_monitor, dealer_monitor)

            send = dealer.send().message(b"ping").submit()
            await send.admitted
            request = zlink.create_received()
            if not router.recv_into(request):
                raise AssertionError("expected dealer-router request")
            with request:
                if request.routing_id != zlink.RoutingId(b"CLIENT"):
                    raise AssertionError(f"unexpected routing id: {request.routing_id!r}")
                if request.to_bytes_list() != [b"ping"]:
                    raise AssertionError("unexpected dealer-router request payload")
                send = request.send().message(b"pong").submit()
                await send.admitted

            reply = zlink.create_received()
            if not dealer.recv_into(reply):
                raise AssertionError("expected dealer-router reply")
            with reply:
                if reply.to_bytes_list() != [b"pong"]:
                    raise AssertionError("unexpected dealer-router reply payload")
            print('[dealer-router/recv] send: "ping" → recv: "pong"')

const endpoint = await tcpEndpoint();
const ctx = zlink.createContext();
const router = zlink.createRouterSocket(ctx);
const dealer = zlink.createDealerSocket(ctx);

try {
  const routerMonitor = router.monitorOpen([zlink.MonitorEventType.ConnectionReady]);
  const dealerMonitor = dealer.monitorOpen([zlink.MonitorEventType.ConnectionReady]);
  try {
    router.bind(endpoint);
    dealer.connect(endpoint);
    await waitForConnectionReady(routerMonitor, zlink);
    await waitForConnectionReady(dealerMonitor, zlink);
  } finally {
    routerMonitor.close();
    dealerMonitor.close();
  }

  const sent = 'ping';
  await dealer.send().message(Buffer.from(sent)).submit().admitted;

  const reply = 'pong';
  const request = new zlink.Received();
  router.recv(request);
  try {
    const recvReq = request.parts[0].data().toString();
    assert.equal(recvReq, sent);
    assert.ok(request.routingId instanceof zlink.RoutingId);
    request.send().message(Buffer.from(reply)).submit();
  } finally {
    request.close();
  }

  const response = new zlink.Received();
  dealer.recv(response);
  try {
    const recv = response.parts[0].data().toString();
    assert.equal(recv, reply);
    console.log(`[dealer-router/recv] send: "${sent}" \u2192 recv: "${recv}"`);
  } finally {
    response.close();
  }
} finally {
  dealer.close();
  router.close();
  ctx.close();
}

const port = await reservePort();
const endpoint = `tcp://127.0.0.1:${port}`;
const ctx = zlink.createContext();
const router = zlink.createRouterSocket(ctx);
const dealer = zlink.createDealerSocket(ctx);

try {
  const routerMonitor = router.monitorOpen([zlink.MonitorEventType.ConnectionReady]);
  const dealerMonitor = dealer.monitorOpen([zlink.MonitorEventType.ConnectionReady]);
  try {
    router.bind(endpoint);
    dealer.connect(endpoint);
    await waitForConnectionReady(routerMonitor);
    await waitForConnectionReady(dealerMonitor);
  } finally {
    routerMonitor.close();
    dealerMonitor.close();
  }

  const sent = 'ping';
  dealer.send().message(Buffer.from(sent)).submit();

  const reply = 'pong';
  const request = new zlink.Received();
  router.recv(request);
  try {
    const recvReq = request.parts[0].data().toString();
    assert.equal(recvReq, sent);
    assert.ok(request.routingId instanceof zlink.RoutingId);
    request.send().message(Buffer.from(reply)).submit();
  } finally {
    request.close();
  }

  const response = new zlink.Received();
  dealer.recv(response);
  try {
    const recv = response.parts[0].data().toString();
    assert.equal(recv, reply);
    console.log(`[dealer-router/recv] send: "${sent}" → recv: "${recv}"`);
  } finally {
    response.close();
  }
} finally {
  dealer.close();
  router.close();
  ctx.close();
}

ctx, err := zlink.NewContext()
samplecommon.Must(err)
defer ctx.Close()

router, err := ctx.RouterSocket()
samplecommon.Must(err)
defer router.Close()
dealer, err := ctx.DealerSocket()
samplecommon.Must(err)
defer dealer.Close()

routerMon := samplecommon.OpenMonitor(router)
defer routerMon.Close()
dealerMon := samplecommon.OpenMonitor(dealer)
defer dealerMon.Close()

endpoint := samplecommon.UniqueTCP("dealer-router-recv")
rid := zlink.NewRoutingID([]byte("dealer-sample"))
samplecommon.Must(router.Bind(endpoint))
samplecommon.Must(dealer.SetRoutingID(rid))
samplecommon.Must(dealer.Connect(endpoint))
samplecommon.WaitConnected(routerMon, dealerMon)

submission, err := dealer.Send().Message(
    samplecommon.Message("ping")).Submit(context.Background())
samplecommon.Must(err)
samplecommon.Must(submission.Admitted(context.Background()))

var request zlink.Received
_, err = router.Recv(&request, zlink.RecvFlagsNone)
samplecommon.Must(err)
defer request.Close()
submission, err = request.Send().Message(samplecommon.Message("pong")).Submit(context.Background())
if err == nil {
    err = submission.Admitted(context.Background())
}
samplecommon.Must(err)

var reply zlink.Received
_, err = dealer.Recv(&reply, zlink.RecvFlagsNone)
samplecommon.Must(err)
defer reply.Close()
part, err := reply.SinglePartOrError()
samplecommon.Must(err)
if !bytes.Equal(part.Data(), []byte("pong")) {
    samplecommon.Must(fmt.Errorf("unexpected reply %q", string(part.Data())))
}

fmt.Printf("[dealer-router/recv] send: %q -> recv: %q\n", "ping", string(part.Data()))

let ctx = Context::new().expect("context creation failed");
let endpoint = sample_support::tcp_endpoint();

let router = ctx.router_socket().expect("router socket failed");
let dealer = ctx.dealer_socket().expect("dealer socket failed");
let rid = RoutingId::from(b"dealer-node-7");
dealer.set_routing_id(&rid).expect("set routing id failed");

let router_mon = SocketMonitor::open(&router).expect("router monitor open failed");
let dealer_mon = SocketMonitor::open(&dealer).expect("dealer monitor open failed");

router.bind(&endpoint).expect("bind failed");
dealer.connect(&endpoint).expect("connect failed");

sample_support::wait_connected(&[&router_mon, &dealer_mon]);
drop(router_mon);
drop(dealer_mon);

let req = Message::try_from(b"ping").expect("message failed");
let submission = dealer.send().message(req).submit().expect("send failed");
sample_support::block_on(submission.admitted).expect("send admission failed");

let mut received = zlink::Received::empty();
router
    .recv(&mut received, zlink::RecvFlags::NONE)
    .expect("router recv failed");
assert!(received.routing_id().is_some());
assert_eq!(received.parts()[0].as_str().unwrap(), "ping");

let resp = Message::try_from(b"pong").expect("message failed");
let submission = received
    .send()
    .message(resp)
    .submit()
    .expect("received send failed");
sample_support::block_on(submission.admitted).expect("received send admission failed");

let mut response = zlink::Received::empty();
dealer
    .recv(&mut response, zlink::RecvFlags::NONE)
    .expect("dealer recv failed");
assert_eq!(response.parts()[0].as_str().unwrap(), "pong");
println!(
    "[dealer-router/recv] send: \"ping\" → recv: \"{}\"",
    response.parts()[0].as_str().unwrap()
);

---

← PUB/SUB | ROUTER →