conduit 1.0.0

Conduit

22 Conduit provides the low-level building blocks for client-server applications that communicate over Unix domain sockets using a compact binary protocol. It includes frame encoding/decoding, EINTR-safe socket transport, POSIX shared memory management, and JSON-RPC support.

Table of Contents

• Setup
• Tutorial: Building a Client-Server App
• Step 1: Define Your Message Types
• Step 2: Write Encode/Decode Functions
• Step 3: Build the Server
• Step 4: Build the Client
• API Reference
• conduit.error
• conduit.codec
• conduit.transport
• conduit.connection
• conduit.shm
• conduit.jsonrpc
• Examples
• Frame Format
• Platform Support
• Building
• License

Setup

Add conduit as a dependency in your dub.json:

{
    "dependencies": {
        "conduit": "~>0.1.0"
    }
}

Or for a local path dependency during development:

{
    "dependencies": {
        "conduit": { "path": "../conduit" }
    }
}

Then import what you need:

import conduit;              // everything
import conduit.codec;        // just the codec primitives
import conduit.connection;   // just socket management

Tutorial: Building a Client-Server App

This walkthrough builds a simple request-response service where a client sends a Ping with a sequence number and the server replies with a Pong.

Step 1: Define Your Message Types

Conduit gives you the framing and transport -- you define the message vocabulary. Assign each message a ushort type ID and create a struct for its fields.

module myapp.messages;

// Type IDs -- pick any ushort values, just keep client and server in sync.
enum MsgType : ushort {
    ping = 0x01,
    pong = 0x81,
}

struct Ping {
    uint seq;
}

struct Pong {
    uint seq;
}

Step 2: Write Encode/Decode Functions

Use conduit.codec primitives to serialize your structs into frame payloads.

module myapp.codec;

import conduit.codec;
import myapp.messages;

// --- Encode ---

ubyte[] encode(Ping msg) @safe pure {
    ubyte[] payload;
    appendUint(payload, msg.seq);
    return encodeFrame(MsgType.ping, payload);
}

ubyte[] encode(Pong msg) @safe pure {
    ubyte[] payload;
    appendUint(payload, msg.seq);
    return encodeFrame(MsgType.pong, payload);
}

// --- Decode ---

Ping decodePing(const(ubyte)[] payload) @safe pure {
    size_t off = 0;
    return Ping(readUint(payload, off));
}

Pong decodePong(const(ubyte)[] payload) @safe pure {
    size_t off = 0;
    return Pong(readUint(payload, off));
}

For messages with strings, use encodeLString / decodeLString:

struct Greeting {
    LString name;
    uint age;
}

ubyte[] encode(Greeting msg) @safe pure {
    ubyte[] payload;
    payload ~= encodeLString(msg.name);
    appendUint(payload, msg.age);
    return encodeFrame(0x02, payload);
}

Greeting decodeGreeting(const(ubyte)[] payload) @safe pure {
    size_t off = 0;
    auto name = decodeLString(payload, off);
    auto age = readUint(payload, off);
    return Greeting(name, age);
}

Step 3: Build the Server

The server creates a listening Unix socket, accepts connections, reads frames, and dispatches by type ID.

import conduit;
import myapp.messages;
import myapp.codec;

void runServer() {
    // Bind and listen on a Unix domain socket
    auto listenResult = createListenSocket("/tmp/myapp.sock");
    if (!listenResult.isOk) return;
    auto listenFd = listenResult.value;

    // Accept one client
    auto acceptResult = acceptConnection(listenFd);
    if (!acceptResult.isOk) return;
    auto conn = acceptResult.value;
    scope(exit) closeConnection(conn);

    // Read-dispatch loop
    while (true) {
        auto frameResult = receiveFrame(conn);
        if (!frameResult.isOk) break;  // client disconnected or error

        auto frame = frameResult.value;
        auto typeId = peekType(frame);
        auto payload = frame[6 .. $];

        switch (typeId) {
            case MsgType.ping:
                auto ping = decodePing(payload);

                // Reply with Pong carrying the same sequence number
                auto reply = encode(Pong(ping.seq));
                sendAll(conn, reply);
                break;

            default:
                break;  // unknown type, ignore
        }
    }
}

Step 4: Build the Client

The client connects, sends a Ping, and reads back the Pong.

import conduit;
import myapp.messages;
import myapp.codec;

void runClient() {
    auto connResult = connectTo("/tmp/myapp.sock");
    if (!connResult.isOk) return;
    auto conn = connResult.value;
    scope(exit) closeConnection(conn);

    // Send a Ping
    auto frame = encode(Ping(1));
    sendAll(conn, frame);

    // Read the reply
    auto replyResult = receiveFrame(conn);
    if (!replyResult.isOk) return;

    auto reply = replyResult.value;
    auto typeId = peekType(reply);
    if (typeId == MsgType.pong) {
        auto pong = decodePong(reply[6 .. $]);
        assert(pong.seq == 1);
    }
}

API Reference

conduit.error

Error types used across all conduit modules. Every fallible operation returns either an ErrorCode directly or a Result!T that pairs a value with an error.

ErrorCode

enum ErrorCode {
    none = 0,           // success

    // connection errors
    socketCreateFailed,
    pathTooLong,
    bindFailed,
    listenFailed,
    connectFailed,
    acceptFailed,

    // transport errors
    writeFailed,
    readFailed,
    connectionClosed,
    payloadTooLarge,

    // shared memory errors
    invalidArgument,
    shmOpenFailed,
    truncateFailed,
    mmapFailed,
}

Result

struct Result(T) {
    T value;
    ErrorCode error;

    bool isOk() const;  // true when error == none
}

Usage pattern:

auto result = connectTo("/tmp/myapp.sock");
if (!result.isOk) {
    writefln("connect failed: %s", result.error);
    return;
}
auto conn = result.value;

For functions that return no value on success, the return type is ErrorCode directly:

auto err = sendAll(conn, data);
if (err != ErrorCode.none) {
    writefln("send failed: %s", err);
}

conduit.codec

Binary frame encoding and decoding primitives. All encoding/decoding functions are @safe pure -- no I/O, no side effects.

Types

`LString` -- Length-prefixed string for protocol serialization.

struct LString {
    const(char)[] data;
}

Frame Functions

`encodeFrame(ushort typeId, const(ubyte)[] payload) -> ubyte[]`

Builds a binary frame: 4-byte LE payload length + 2-byte LE type ID + payload bytes.

auto frame = encodeFrame(0x01, payload);
// frame = [len0, len1, len2, len3, type0, type1, payload...]

`peekType(const(ubyte)[] header) -> ushort`

Extracts the 2-byte type ID from a frame header (bytes 4-5). The header must be at least 6 bytes. Does not consume the header.

auto typeId = peekType(frame);
auto payload = frame[6 .. $];

LString Functions

`encodeLString(LString s) -> ubyte[]`

Encodes a string as a 4-byte LE length prefix followed by the raw UTF-8 bytes.

`decodeLString(const(ubyte)[] data, ref size_t offset) -> LString`

Reads a length-prefixed string from data at the given offset. Advances offset past the consumed bytes.

size_t off = 0;
auto name = decodeLString(payload, off);
auto role = decodeLString(payload, off);  // reads next string after name

Integer Functions

All integer functions use little-endian byte order.

Append	Read	Type
appendUint(ref ubyte[] buf, uint val)	readUint(const(ubyte)[] data, ref size_t offset) -> uint	4-byte unsigned
appendUshort(ref ubyte[] buf, ushort val)	readUshort(const(ubyte)[] data, ref size_t offset) -> ushort	2-byte unsigned
appendInt(ref ubyte[] buf, int val)	readInt(const(ubyte)[] data, ref size_t offset) -> int	4-byte signed

Append functions grow the buffer by appending bytes. Read functions advance offset by the number of bytes consumed.

conduit.transport

EINTR-safe socket I/O for sending and receiving data over connections.

`sendAll(Connection conn, const(ubyte)[] data) -> ErrorCode`

Writes all bytes to the connection, retrying on partial writes and EINTR. Returns ErrorCode.none on success, writeFailed or connectionClosed on error.

`receiveExact(Connection conn, size_t count) -> Result!(ubyte[])`

Reads exactly count bytes from the connection, retrying on partial reads and EINTR. Returns the byte buffer on success, or readFailed/connectionClosed on error.

`receiveFrame(Connection conn, uint maxPayload = DEFAULT_MAX_PAYLOAD) -> Result!(ubyte[])`

Reads a complete binary frame (6-byte header + payload) in a single allocation. Returns the full frame on success, or an error on disconnect/oversize payload. Use peekType to inspect the type ID and slice [6 .. $] for the payload.

auto result = receiveFrame(conn);
if (!result.isOk) return;  // disconnected or error

auto typeId = peekType(result.value);
auto payload = result.value[6 .. $];

`sendLine(int fd, string data) -> ErrorCode`

Sends a string followed by \n over a raw file descriptor. Useful for text-based handshake protocols.

`readLine(int fd, uint maxLength = DEFAULT_MAX_LINE) -> Result!string`

Reads bytes from a raw file descriptor until \n is encountered. Returns the string without the trailing newline. Returns payloadTooLarge if the line exceeds maxLength (default 64 KB).

conduit.connection

Unix domain socket lifecycle management.

`Connection` -- Wrapper around a socket file descriptor.

struct Connection {
    int fd = -1;
    bool isValid() const @safe pure;
}

`createListenSocket(string path, int backlog = 5, bool unlinkFirst = true) -> Result!int`

Creates a listening Unix domain socket bound to path. Removes any existing socket file by default. Returns the listening fd on success, or socketCreateFailed/pathTooLong/bindFailed/listenFailed on error.

`connectTo(string path) -> Result!Connection`

Connects to an existing Unix domain socket at path. Returns connectFailed/pathTooLong on error.

`connectToFd(string path) -> Result!int`

Like connectTo but returns the raw file descriptor instead of a Connection.

`acceptConnection(int listenFd) -> Result!Connection`

Accepts an incoming connection on a listening socket. Returns acceptFailed on error.

`closeConnection(Connection conn)`

Closes the connection's file descriptor. Safe to call on invalid connections.

`initUnixAddr(ref sockaddr_un addr, string path) -> ErrorCode`

Low-level helper: initializes a sockaddr_un for the given path. Returns pathTooLong if the path exceeds sun_path capacity.

conduit.shm

POSIX shared memory buffer with ownership tracking, mmap-based I/O, and resize support. The entire interface is @nogc @trusted nothrow.

`ShmBuffer`

struct ShmBuffer {
    @property inout(ubyte)* ptr() inout;  // pointer to mapped memory
    @property size_t size() const;         // current buffer size
    @property bool isValid() const;        // true if mapped

    // Create a new shared memory region (caller becomes owner)
    static Result!ShmBuffer create(const(char)* name, size_t size);

    // Open an existing shared memory region (non-owner)
    static Result!ShmBuffer open(const(char)* name, size_t size);

    // Resize the mapped region (Linux only for owner)
    ErrorCode resize(size_t newSize);

    // Unmap memory, close fd, and unlink if owner
    void close();
}

Usage:

// Server (owner): create shared memory
auto createResult = ShmBuffer.create("/myapp-shm\0".ptr, 1_048_576);
if (!createResult.isOk) return;
auto buf = createResult.value;
scope(exit) buf.close();

// Write data to shared memory
import core.stdc.string : memcpy;
memcpy(buf.ptr, data.ptr, data.length);

// Client (non-owner): open the same region
auto openResult = ShmBuffer.open("/myapp-shm\0".ptr, 1_048_576);
if (!openResult.isOk) return;
auto clientBuf = openResult.value;
scope(exit) clientBuf.close();

// Read data written by server
auto firstByte = clientBuf.ptr[0];

Ownership rules:

• The process that calls create is the owner. On close(), it unlinks the shared memory object.
• Processes that call open are non-owners. On close(), they unmap but do not unlink.
• resize only works for the owner on Linux. macOS does not support re-truncating shared memory.
• Double close() is safe (no-op on already-closed buffers).
• The caller must ensure that open uses the same size as create. Passing a larger size causes undefined behavior (POSIX limitation).

conduit.jsonrpc

Lightweight JSON-RPC 2.0 message parsing and formatting with Content-Length framing. No external JSON library dependency -- uses hand-rolled extraction.

`JsonRpcMessage` -- Parsed request/notification.

struct JsonRpcMessage {
    string method;       // e.g. "initialize", "textDocument/hover"
    int id;              // request ID (only valid if hasId is true)
    bool hasId;          // false for notifications
    string rawParams;    // raw JSON string of the params object
}

`parseMessage(string content) -> JsonRpcMessage`

Parses a JSON-RPC message body. Extracts method, id, and params fields.

auto msg = parseMessage(`{"jsonrpc":"2.0","method":"initialize","id":1,"params":{}}`);
// msg.method == "initialize"
// msg.id == 1
// msg.hasId == true
// msg.rawParams == "{}"

`formatResponse(int id, string resultJson) -> string`

Formats a JSON-RPC success response with Content-Length header.

auto resp = formatResponse(1, `{"capabilities":{}}`);
// "Content-Length: 55\r\n\r\n{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"capabilities\":{}}}"

`formatNotification(string method, string paramsJson) -> string`

Formats a JSON-RPC notification (no id field) with Content-Length header.

`formatErrorResponse(int id, int code, string message) -> string`

Formats a JSON-RPC error response with Content-Length header.

auto err = formatErrorResponse(1, -32601, "Method not found");

`readStdinMessage() -> string`

Reads a Content-Length framed message from stdin. Useful for implementing LSP servers. Returns null on EOF.

`writeStdout(string data)`

Writes a response string to stdout and flushes.

Helpers (also public):

• extractStringValue(string json, size_t startIdx) -> string
• extractIntValue(string json, size_t startIdx) -> int
• extractObjectValue(string json, size_t startIdx) -> string
• intToStr(int n) -> string

Examples

Two runnable examples live in examples/, each a standalone dub subpackage:

`examples/ping-pong` -- Client-server binary frame round-trip over Unix sockets. Demonstrates createListenSocket, acceptConnection, connectTo, sendAll, receiveFrame, encodeFrame, and peekType.

# Terminal 1: start server
cd examples/ping-pong && dub run -- --server --path /tmp/demo.sock

# Terminal 2: run client
cd examples/ping-pong && dub run -- --client --path /tmp/demo.sock

`examples/shm-demo` -- Cross-process shared memory write and read. Demonstrates ShmBuffer.create, ShmBuffer.open, and direct memory access.

# Write a message into shared memory
cd examples/shm-demo && dub run -- --write --name /demo_shm --message "hello"

# Read it back from another process
cd examples/shm-demo && dub run -- --read --name /demo_shm --message "hello"

Both examples are also used as integration tests via tests/integration/run.sh.

Frame Format

Conduit uses a 6-byte header followed by a variable-length payload:

Offset  Size  Field
------  ----  ------------------------------
0       4     Payload length (uint32, LE)
4       2     Type ID (uint16, LE)
6       N     Payload bytes

A frame with no payload is exactly 6 bytes (length = 0).

Strings inside payloads use LString encoding:

Offset  Size  Field
------  ----  ---------------
0       4     String length (uint32, LE)
4       N     UTF-8 bytes

All multi-byte integers are little-endian.

Platform Support

Feature	Linux	macOS
Frame codec	Yes	Yes
Unix socket transport	Yes	Yes
Shared memory (create/open)	Yes	Yes
Shared memory (resize)	Yes	No
JSON-RPC	Yes	Yes

Building

# Build the library
dub build

# Run unit tests
dub test

# Build examples
for ex in ping-pong shm-demo; do (cd examples/$ex && dub build); done

# Run integration tests
bash tests/integration/run.sh

Requires LDC or DMD compiler with the D standard library.

License

BSD 3-Clause License. See LICENSE for details.