← Socket Patterns | PUB/SUB →

PAIR Socket

1. Overview

The PAIR socket forms an exclusive 1:1 bidirectional connection with exactly one peer. If a second peer connects, that later connection is rejected — the first peer keeps the pipe.

Key characteristics: - Only a single pipe is allowed (1:1 exclusive) - Bidirectional free messaging (send/recv order does not matter) - The simplest socket type

Valid socket combinations: PAIR ↔ PAIR

%%{init: {'flowchart': {'nodeSpacing': 32, 'rankSpacing': 40, 'padding': 8, 'wrappingWidth': 180}, 'themeVariables': {'fontSize': '18px'}}}%%
flowchart LR
    A[PAIR A] <-->|Bidirectional| B[PAIR B]

2. Basic Usage

Creation and Connection

void *ctx = zlink_ctx_new();

/* Server side */
void *server = zlink_socket(ctx, ZLINK_SOCKET_PAIR);
zlink_bind(server, "tcp://*:5555");

/* Client side */
void *client = zlink_socket(ctx, ZLINK_SOCKET_PAIR);
zlink_connect(client, "tcp://127.0.0.1:5555");

Message Exchange

PAIR's public receive API is recv/poller-only: receive one complete record with zlink_recv(), typically inside a poller loop. Both peers can send and receive freely. PAIR has exactly one peer, so there is no routing id to select — pass NULL for source_rid_out_.

/* Client → Server */
zlink_msg_t msg;
zlink_msg_init_size(&msg, 5);
memcpy(zlink_msg_data(&msg), "Hello", 5);
zlink_send(client, &msg, 1, ZLINK_SEND_FLAGS_NONE, NULL, NULL);

/* Server receives with zlink_recv() (typically inside a poller loop) */
zlink_msg_t parts[8];
size_t part_count = 0;
if (zlink_recv(server, NULL, parts, 8, &part_count, ZLINK_RECV_FLAGS_NONE) == ZLINK_RECV_OK) {
    printf("Received: %.*s\n",
           (int)zlink_msg_size(&parts[0]),
           (char *)zlink_msg_data(&parts[0]));
    zlink_multipart_close(parts, part_count);
}

/* Server → Client (bidirectional; client uses the same recv+poller pattern) */
zlink_msg_t reply;
zlink_msg_init_size(&reply, 5);
memcpy(zlink_msg_data(&reply), "World", 5);
zlink_send(server, &reply, 1, ZLINK_SEND_FLAGS_NONE, NULL, NULL);

Sending Multipart Data

Place multipart data in array order and make one zlink_send() call. Core submits the complete array atomically as one record.

zlink_msg_t parts[2];
zlink_msg_init_size(&parts[0], 3);
memcpy(zlink_msg_data(&parts[0]), "foo", 3);
zlink_msg_init_size(&parts[1], 6);
memcpy(zlink_msg_data(&parts[1]), "foobar", 6);

zlink_submit_result_t rc = zlink_send(
    server, parts, 2, ZLINK_SEND_FLAGS_NONE, NULL, NULL);
/* One zlink_recv() call returns both parts in the same order. */

Reference: core/tests/integration/test_pair_inproc.cpp -- test_zlink_send_multipart() test

Receive Modes

PAIR is recv/poller-only in the public API. Use zlink_recv() to receive one complete record synchronously.

void *pair = zlink_socket(ctx, ZLINK_SOCKET_PAIR);
zlink_bind(pair, "tcp://*:5556");

zlink_msg_t parts[8];
size_t part_count = 0;
zlink_recv_result_t rc = zlink_recv(
    pair, 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);
}

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

3. Message Format

PAIR socket message frames contain application data only.

Single frame:     [data]
Multipart frame:  [frame1][frame2]...[frameN]

For source_rid and the common receive interface, see Socket Patterns Overview.

Multipart send:

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(
    server, parts, 2, ZLINK_SEND_FLAGS_NONE, NULL, NULL);

4. Socket Options

Option	Type	Default	Description
ZLINK_OPT_SNDHWM	uint64_t bytes	automatic	Auto-HWM sized for PAIR's peer-queue role. Manual settings take precedence; 0 is unlimited
ZLINK_OPT_RCVHWM	uint64_t bytes	automatic	Auto-HWM sized for PAIR's peer-queue role. Manual settings take precedence; 0 is unlimited
ZLINK_OPT_LINGER	int	-1	Wait time for unsent messages on close (ms), -1=infinite
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

uint64_t hwm_bytes = 5 * 1024 * 1024;  /* HWM is bytes, passed as exactly 8 bytes */
zlink_set_option(socket, ZLINK_OPT_SNDHWM, &hwm_bytes, sizeof(hwm_bytes));

int linger = 0;  /* return immediately on close */
zlink_set_option(socket, ZLINK_OPT_LINGER, &linger, sizeof(linger));

5. Usage Patterns

Pattern 1: Inter-thread Signaling (inproc)

The most common PAIR use case. Zero-copy communication between threads via the inproc transport.

/* Main thread */
void *signal = zlink_socket(ctx, ZLINK_SOCKET_PAIR);
zlink_bind(signal, "inproc://signal");

/* Worker thread */
void *worker_signal = zlink_socket(ctx, ZLINK_SOCKET_PAIR);
zlink_connect(worker_signal, "inproc://signal");

/* Worker → Main: task completion signal */
zlink_msg_t msg;
zlink_msg_init_size(&msg, 4);
memcpy(zlink_msg_data(&msg), "DONE", 4);
zlink_send(worker_signal, &msg, 1, ZLINK_SEND_FLAGS_NONE, NULL, NULL);

/* Main: receives "DONE" via its poller loop (zlink_recv) */

Reference: core/tests/integration/test_pair_inproc.cpp -- bind → connect → bounce pattern

Pattern 2: TCP Communication

1:1 communication over the network. Wildcard bind enables automatic port assignment.

/* Server: wildcard port */
void *server = zlink_socket(ctx, ZLINK_SOCKET_PAIR);
zlink_bind(server, "tcp://127.0.0.1:*");

/* Query the assigned endpoint */
char endpoint[256];
size_t len = sizeof(endpoint);
zlink_get_option(server, ZLINK_OPT_LAST_ENDPOINT, endpoint, &len);

/* Client: connect using the queried endpoint */
void *client = zlink_socket(ctx, ZLINK_SOCKET_PAIR);
zlink_connect(client, endpoint);

Reference: core/tests/integration/test_pair_tcp.cpp -- bind_loopback_ipv4() + wildcard bind

Pattern 3: Connection by DNS Name

You can also connect using a hostname.

void *client = zlink_socket(ctx, ZLINK_SOCKET_PAIR);
zlink_connect(client, "tcp://localhost:5555");

Reference: core/tests/integration/test_pair_tcp.cpp -- test_pair_tcp_connect_by_name()

Pattern 4: IPC Communication

Inter-process communication on the same machine (Linux/macOS).

void *server = zlink_socket(ctx, ZLINK_SOCKET_PAIR);
zlink_bind(server, "ipc:///tmp/myapp.ipc");

void *client = zlink_socket(ctx, ZLINK_SOCKET_PAIR);
zlink_connect(client, "ipc:///tmp/myapp.ipc");

Reference: core/tests/integration/test_pair_ipc.cpp -- includes IPC path length validation

6. Caveats

Only a Single Peer Allowed

A PAIR socket maintains only one connection. If a second peer connects, that later connection is rejected; the first peer keeps the pipe.

 Allowed:  PAIR A ↔ PAIR B      (1:1)
 Invalid:  PAIR A ← PAIR B      (N:1 attempt: later peers rejected)
               ← PAIR C

Use DEALER/ROUTER if N:1 communication is needed.

inproc bind Order

With the inproc transport, bind must be called before connect.

/* Correct order */
zlink_bind(a, "inproc://signal");     /* 1. bind first */
zlink_connect(b, "inproc://signal");  /* 2. connect */

/* Wrong order -- fails */
zlink_connect(b, "inproc://signal");  /* fails because bind has not been called yet */
zlink_bind(a, "inproc://signal");

IPC Path Length

The file path of an IPC endpoint cannot exceed the system limit (typically 108 characters).

/* Path too long → ENAMETOOLONG error */
zlink_bind(socket, "ipc:///very/long/path/.../endpoint.ipc");

Reference: core/tests/integration/test_pair_ipc.cpp -- test_endpoint_too_long()

HWM Behavior

When there is no peer or the peer is slow, outgoing messages are queued up to the HWM. When the HWM is exceeded, zlink_send() blocks (default) or returns ZLINK_SUBMIT_BACKPRESSURED (ZLINK_DONTWAIT).

LINGER Setting

When zlink_close() is called and there are unsent messages remaining, it waits for the LINGER duration. For tests or when a fast shutdown is needed:

int linger = 0;
zlink_set_option(socket, ZLINK_OPT_LINGER, &linger, sizeof(linger));

---

← Socket Patterns | PUB/SUB →

Full language examples

C++C#/.NETJavaKotlinPythonNode/TypeScriptJavaScriptGoRust

zlink::context_t ctx;
zlink::pair_socket_t server (ctx);
zlink::pair_socket_t client (ctx);
zlink::socket_monitor_t server_monitor = server.monitor_open ();
zlink::socket_monitor_t client_monitor = client.monitor_open ();

server.bind ("tcp://127.0.0.1:0");
const std::string endpoint = server.options ().last_endpoint ();
assert (!endpoint.empty ());
client.connect (endpoint);
assert (detail::wait_connected (server_monitor, client_monitor));

const std::string sent = detail::k_pair_payload;
zlink::message_t outbound = detail::make_message (sent);
client.send ().message (outbound).submit ();

zlink::received_t inbound;
assert (server.recv (inbound) == 0);
assert (inbound.parts ().size () == 1);
const std::string received = inbound.parts ()[0].to_string ();
assert (received == detail::k_pair_payload);
inbound.close ();
std::printf ("[pair/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 sender = ctx.CreatePairSocket();
using var receiver = ctx.CreatePairSocket();
string endpoint = SampleSupport.NewEndpoint("tcp", "sample");
using var senderMonitor = sender.MonitorOpen(SocketEvent.ConnectionReady);
using var receiverMonitor = receiver.MonitorOpen(SocketEvent.ConnectionReady);
sender.Bind(endpoint);
receiver.Connect(endpoint);
SampleSupport.WaitConnected(senderMonitor, receiverMonitor);

const string payload = "hello-pair";
using (Message message = Message.From(payload))
    sender.Send().Message(message).Submit();
string receivedPayload = SampleSupport.ReceiveUtf8(receiver, 2000);
Console.WriteLine(
    $"[pair/recv] send: \"hello-pair\" -> recv: \"{receivedPayload}\"");

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

try (Context ctx = Zlink.createContext();
     PairSocket server = ctx.createPairSocket();
     PairSocket client = ctx.createPairSocket();
     var serverMonitor = server.monitorOpen(
         systems.zlink.contracts.eventing.MonitorEventType.CONNECTION_READY);
     var clientMonitor = client.monitorOpen(
         systems.zlink.contracts.eventing.MonitorEventType.CONNECTION_READY)) {
    server.bind(endpoint);
    client.connect(endpoint);
    SampleSupport.waitConnected(serverMonitor, clientMonitor);

    try (Message outbound = Message.from(SampleSupport.PAIR_PAYLOAD)) {
        client.send().message(outbound).submit().admitted()
            .toCompletableFuture().join();
    }

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

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

Zlink.createContext().use { ctx ->
    ctx.createPairSocket().use { server ->
        ctx.createPairSocket().use { client ->
            server.monitorOpen(MonitorEventType.CONNECTION_READY).use { serverMonitor ->
                client.monitorOpen(MonitorEventType.CONNECTION_READY).use { clientMonitor ->
                    server.bind(endpoint)
                    client.connect(endpoint)
                    SampleSupport.waitConnected(serverMonitor, clientMonitor)

                    Message.from(SampleSupport.PAIR_PAYLOAD).use { outbound ->
                        client.send().message(outbound).submit()
                    }

                    Received().use { received ->
                        server.recv(received, RecvFlags.NONE)
                        val value = SampleSupport.singleUtf8(received)
                        check(SampleSupport.PAIR_PAYLOAD == value) { "unexpected payload: $value" }
                        println(
                            "[pair/recv] send: \"${SampleSupport.PAIR_PAYLOAD}\"" +
                                " → recv: \"$value\""
                        )
                    }
                }
            }
        }
    }
}

_, endpoint = tcp_endpoint()

with zlink.create_context() as ctx:
    with zlink.create_pair_socket(ctx) as server:
        with zlink.create_pair_socket(ctx) as client:
            with server.monitor_open(zlink.MonitorEventMask.CONNECTION_READY) as server_monitor:
                with client.monitor_open(zlink.MonitorEventMask.CONNECTION_READY) as client_monitor:
                    server.bind(endpoint)
                    client.connect(endpoint)
                    wait_connected(server_monitor, client_monitor)

            async def send():
                submission = client.send().message(b"hello-pair").submit()
                await submission.admitted

            asyncio.run(send())
            received = zlink.create_received()
            if not server.recv_into(received):
                raise AssertionError("expected pair payload")
            with received:
                payload = received.to_bytes_list()
                if payload != [b"hello-pair"]:
                    raise AssertionError(f"unexpected pair payload: {payload!r}")
            print('[pair/recv] send: "hello-pair" → recv: "hello-pair"')

const endpoint = await tcpEndpoint();
const ctx = zlink.createContext();
const server = zlink.createPairSocket(ctx);
const client = zlink.createPairSocket(ctx);

try {
  const serverMonitor = server.monitorOpen([zlink.MonitorEventType.ConnectionReady]);
  const clientMonitor = client.monitorOpen([zlink.MonitorEventType.ConnectionReady]);
  try {
    server.bind(endpoint);
    client.connect(endpoint);
    serverMonitor.recv();
    clientMonitor.recv();
  } finally {
    serverMonitor.close();
    clientMonitor.close();
  }

  const sent = 'hello-pair';
  await client.send().message(Buffer.from(sent)).submit().admitted;

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

const port = await reservePort();
const endpoint = `tcp://127.0.0.1:${port}`;
const ctx = zlink.createContext();
const server = zlink.createPairSocket(ctx);
const client = zlink.createPairSocket(ctx);

try {
  const serverMonitor = server.monitorOpen([zlink.MonitorEventType.ConnectionReady]);
  const clientMonitor = client.monitorOpen([zlink.MonitorEventType.ConnectionReady]);
  try {
    server.bind(endpoint);
    client.connect(endpoint);
    serverMonitor.recv();
    clientMonitor.recv();
  } finally {
    serverMonitor.close();
    clientMonitor.close();
  }

  const sent = 'hello-pair';
  client.send().message(Buffer.from(sent)).submit();

  const received = new zlink.Received();
  server.recv(received);
  try {
    const recv = received.parts[0].data().toString();
    assert.equal(recv, sent);
    console.log(`[pair/recv] send: "${sent}" → recv: "${recv}"`);
  } finally {
    received.close();
  }
} finally {
  client.close();
  server.close();
  ctx.close();
}

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

server, err := ctx.PairSocket()
samplecommon.Must(err)
defer server.Close()
client, err := ctx.PairSocket()
samplecommon.Must(err)
defer client.Close()

serverMon := samplecommon.OpenMonitor(server)
defer serverMon.Close()
clientMon := samplecommon.OpenMonitor(client)
defer clientMon.Close()

endpoint := samplecommon.UniqueTCP("pair-recv")
samplecommon.Must(server.Bind(endpoint))
samplecommon.Must(client.Connect(endpoint))
samplecommon.WaitConnected(serverMon, clientMon)

sent := "hello-pair"
submission, err := client.Send().Message(samplecommon.Message(sent)).Submit(context.Background())
if err == nil {
    err = submission.Admitted(context.Background())
}
samplecommon.Must(err)

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

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

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

let server = ctx.pair_socket().expect("server socket failed");
let client = ctx.pair_socket().expect("client socket failed");

let server_mon = SocketMonitor::open(&server).expect("server monitor open failed");
let client_mon = SocketMonitor::open(&client).expect("client monitor open failed");

server.bind(&endpoint).expect("bind failed");
client.connect(&endpoint).expect("connect failed");

sample_support::wait_connected(&[&server_mon, &client_mon]);
drop(server_mon);
drop(client_mon);

let msg = Message::try_from(b"hello-pair").expect("message creation failed");
let submission = client.send().message(msg).submit().expect("send failed");
sample_support::block_on(submission.admitted).expect("send admission failed");

let mut received = zlink::Received::empty();
server
    .recv(&mut received, zlink::RecvFlags::NONE)
    .expect("recv failed");
let payload = received.parts()[0].as_str().expect("utf8 error");
assert_eq!(payload, "hello-pair");
println!("[pair/recv] send: \"hello-pair\" → recv: \"{}\"", payload);

---

← Socket Patterns | PUB/SUB →