C++

clients/client-cpp/queen_client.hpp is a single header, about 2,150 lines, that implements the message plane, transactions, the DLQ query, client-side buffering, round-robin and session load balancing, and a thread-pool consumer. It needs C++17 and three third-party headers. It is the least complete of the SDKs: there is no admin surface at all, and lease renewal inside the consume loop is unimplemented.

Building it

The header opens with #include <json.hpp>, #include <httplib.h> and #include "../server/include/threadpool.hpp", and the Makefile adds -I../server/vendor -I../server/include. Relative to clients/client-cpp/, those resolve to clients/server/vendor/ and clients/server/include/, directories that do not exist in the repository. Two of the three headers are third-party and gitignored; the third, threadpool.hpp (the MIT-licensed astp::ThreadPool), was recovered from git history and is vendored at test/vendor/cpp/threadpool.hpp.

The test runner’s Dockerfile is the authoritative recipe. From the repository root:

Step 3 only needs a directory already on the compiler’s include path. The Makefile probes /opt/homebrew/include, /usr/local/include and /usr/include for httplib.h and adds whichever it finds first, and on macOS brew install cpp-httplib puts it in the Homebrew prefix. OpenSSL is on by default: USE_SSL=1 defines CPPHTTPLIB_OPENSSL_SUPPORT and links -lssl -lcrypto. Setting USE_SSL=0 drops only the flags: queen_client.hpp defines CPPHTTPLIB_OPENSSL_SUPPORT itself if it is not already defined, so cpp-httplib still compiles its TLS paths and expects the OpenSSL symbols. Keep USE_SSL=1.

make test builds bin/test_client, which takes the broker URL as argv[1] (there is no environment override, and the default is http://localhost:6632). make retry429 builds bin/test_retry429, the proxy-contract suite (bearer auth, 429 backoff, terminal 403), which serves its own responses from an in-process httplib::Server and therefore needs neither a broker nor PostgreSQL.

In your own project, put queen_client.hpp wherever you like and satisfy the three includes however your build system prefers; the ../server/... path in the header is the only thing that forces the directory layout above.

Constructing a client

#include "queen_client.hpp"

using namespace queen;

QueenClient client("http://localhost:6632");

For a token, a cell endpoint or several brokers, use the vector constructor with a ClientConfig:

ClientConfig config;
config.bearer_token = std::getenv("QUEEN_TOKEN");
config.timeout_millis = 30000;
config.retry_429.max_attempts = 20;

QueenClient client({"https://cell.example"}, config);

The single-URL constructor ignores ClientConfig entirely: it hardcodes the defaults and passes no bearer token. Authenticated use must go through the vector constructor.

Push, pop, consume

Everything is synchronous and returns nlohmann::json.

auto res = client.queue("orders")
                 .partition("acct-42")
                 .push({{{"transactionId", "order-1"}, {"data", {{"id", 1}}}}});
// res[0]["status"] == "queued"

push() takes a std::vector<json>. The body is data, or payload, or the item itself; a missing transactionId is minted as a UUIDv7 and a traceId is forwarded only if it is a valid UUID. The response is the broker’s per-item array with status set to queued, duplicate, buffered or failed.

json messages = client.queue("orders").group("billing").batch(10).pop();

pop() returns the messages array, or an empty array.

std::atomic<bool> stop{false};

client.queue("jobs")
      .group("workers")
      .concurrency(4)
      .batch(1)
      .each()
      .auto_ack(true)
      .consume([](const json& msg) { process(msg); }, &stop);

consume() blocks until every worker stops. Workers stop on the stop_signal you pass, on client.close() having been called (the worker checks is_shutdown_requested()), on limit, on idle_millis, or on a terminal 403, which is recorded and rethrown to the caller once all workers have finished, because the workers run inside packaged_tasks whose futures are only waited on.

The builder methods are group, concurrency, batch, partitions, limit, idle_millis, auto_ack, wait, renew_lease, subscription_mode, subscription_from, each and namespace_name / task. auto_ack is client-side: the worker calls POST /api/v1/ack itself after your handler returns (completed) or throws (failed).

Acknowledge, renew, transact

json result = client.ack(messages, true, {{"group", "billing"}});

ack() takes a single message object or an array, a boolean status (true → completed, false → failed) and a context object carrying group and error. There is no way to send retry or dlq from this SDK: the boolean maps only to completed or failed. A missing partitionId throws for an array and returns {"success": false, …} for a single message.

The return value is {"success": true, "result": <broker array>}, where success reflects whether the request was accepted, not whether each ack landed. A rejected ack arrives as HTTP 200 with success: false on its item, so read result["result"][i]["success"].

client.transaction()
      .queue("stage-2").push({{{"data", {{"value", 2}}}}})
      .ack(message)
      .commit();

commit() sends the collected pushes and acks as one PostgreSQL transaction, together with the acked messages’ lease IDs as requiredLeases. It throws if the broker reports failure or if there are no operations. Note that the C++ transaction push sub-builder has no partition() method, so transactional pushes always land in the Default partition.

client.renew(messages);

Buffering and the DLQ

BufferOptions buf;
buf.message_count = 500;
buf.time_millis = 200;

client.queue("events").buffer(buf).push(items);   // {"buffered": true, "count": N}

Buffered pushes flush on a count or time trigger, keyed per queue/partition. client.queue("events").flush_buffer() flushes one address, client.flush_all_buffers() flushes all, client.get_buffer_stats() reports the counters. The buffer is in process memory: an unflushed buffer dies with the process.

json dlq = client.queue("orders").dlq("billing").limit(50).get();

get() returns {"messages": […], "total": 0} on any error.

What is missing

• No admin or observability surface. queen_client.hpp defines no Admin class. Resource listings, queue and partition stats, /api/v1/status, analytics, traces-by-name, consumer group management, seek, health, /metrics and the maintenance-mode switches are all unreachable from this SDK. Use curl or another client for those, or call them yourself through client.get_http_client().
• No lease renewal during consume (see above).
• No affinity load balancing. Round-robin and session only, so pops for one queue:partition:group lane do not pin to one broker the way they do in the other SDKs.
• No per-builder long-poll timeout.
• No retry or dlq ack status.
• No streaming SDK.
• No partition() on a transactional push.

429 and 403

The 429/403 contract is fully implemented and covered by test_retry429.cpp, which runs without a broker. A 429 is retried in place against the same backend, honouring Retry-After (seconds) when present and otherwise backing off base_millis * 2^attempt capped at cap_millis, jittered. Ordinary requests get ten attempts; a long-poll pop (RetryKind::Pop) retries unboundedly unless max_attempts is set, in which case it applies to both.

Non-2xx responses throw queen::HttpError, which derives from std::runtime_error and keeps what() equal to the broker’s error message, so existing catch (const std::exception&) sites keep working:

try {
    client.queue("orders").push(items);
} catch (const HttpError& e) {
    if (e.is_cluster_suspended()) {
        // terminal: operator action required
    }
    log(e.status_code(), e.code(), e.body());
}

status_code(), code(), body() and retry_after_seconds() are all available. code() is empty when the response body has no code field. A transport failure with no response at all still surfaces as a plain std::runtime_error, not an HttpError.

Shutdown

client.close();

close() sets the shutdown flag that consume workers poll, flushes every buffer, and cleans up the buffer manager.