C++ client

One header, queen_client.hpp, namespace queen. Everything is inline, so there is nothing to link. Version 1.0.0, aligned with the broker.

#include "queen_client.hpp"

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

Build requirements

The relative include means the header expects to sit next to the repository’s server/ directory, or to be compiled with that directory on the include path (-I../server/include -I../server/vendor, as the shipped Makefile does).

Configuration

struct Retry429Options {
    int max_attempts = 0;   // 0 = kind-based default
    int base_millis  = 0;   // 0 = 500
    int cap_millis   = 0;   // 0 = 30000
};

struct ClientConfig {
    std::vector<std::string> urls;
    int timeout_millis = 30000;
    int retry_attempts = 3;
    int retry_delay_millis = 1000;
    std::string load_balancing_strategy = "round-robin";
    bool enable_failover = true;
    std::string bearer_token;
    Retry429Options retry_429;
};

Retry429Options covers HTTP 429 only, separately from retry_attempts, which covers 5xx and network faults. A zero field means “use the kind-based default”: max_attempts is 10 for ordinary requests and unbounded for a long-poll pop; base_millis is 500; cap_millis is 30000. A Retry-After header (seconds) wins over the computed delay, and every delay is jittered ±20%.

QueenClient

QueenClient(const std::string& url);
QueenClient(const std::vector<std::string>& urls, const ClientConfig& config = ClientConfig());

ack

message is a message object, a transaction-id string, or a JSON array of either. status is bool only: true → completed, false → failed. context accepts a group key and an error key.

partitionId is required on every message object; without it the call throws std::runtime_error("Message must have partitionId property"). A leaseId is forwarded when present. One message posts to /api/v1/ack; an array posts to /api/v1/ack/batch.

Return shape:

renew

Accepts a lease-id string, a message object, or an array of either. Lease ids are deduplicated in insertion order: with multi-partition pop one extend call renews every claimed partition, so passing the whole message array issues one HTTP call. Returns an array of {leaseId, success, newExpiresAt} for an array input, or the single element for a scalar input.

QueueBuilder

client.queue(name). Configuration methods return QueueBuilder&; terminal methods are push(), pop(), consume(), create(), del(), dlq() and flush_buffer().

Addressing

Queue lifecycle

QueueConfig and its to_json() output:

create() always sends the complete set, so calling it without config() applies the struct’s defaults, including leaseTime: 300, whereas a queue created implicitly by a first push gets a 60-second lease. to_json() emits a fixed key set, so options outside that struct cannot be set through config(); post to /api/v1/configure yourself when you need one. And remember /configure is a full replace: keys absent from the body reset to the broker’s defaults. The option list is in Reference.

Producing

Each item may carry data, payload, or be the payload itself. The client mints a UUIDv7 transactionId when the item has none, and forwards traceId only when it is a valid UUID. A direct push returns the broker’s per-item array; a buffered push returns {"buffered": true, "count": n} without sending.

client.queue("orders")
      .partition("customer-42")
      .push({{{"data", {{"orderId", 8891}}}}});

Consuming

pop()

json pop() returns the messages array, or an empty array. Route selection: queue plus non-Default partition uses /api/v1/pop/queue/:queue/partition/:partition; queue alone uses /api/v1/pop/queue/:queue; namespace or task alone uses /api/v1/pop. None of the three throws.

consume()

void consume(std::function<void(const json&)> handler,
             std::atomic<bool>* stop_signal = nullptr);

Blocks until every worker stops. The handler receives one message in each() mode and the whole messages array otherwise. Workers stop when *stop_signal becomes true, when is_shutdown_requested() is set by close(), on the limit, or on the idle timeout.

std::atomic<bool> stop{false};
client.queue("orders").group("workers").batch(10).each()
      .consume([](const queen::json& msg) { /* … */ }, &stop);

Worker loop behaviour:

• A handler exception with auto_ack on nacks the message and the worker continues.
• A 429 that escapes the HTTP layer’s own retry sleeps for Retry-After (or 1 s) and continues.
• A timeout while waiting, or a connection failure, retries after a short sleep.
• A 403 stops that worker and is re-thrown to the caller of consume() once every worker has finished. Workers run inside packaged_tasks that are only waited on, so the exception is captured and re-thrown rather than discarded.
• Any other HttpError or exception is re-thrown from the worker.

Buffering

flush_buffer() flushes this builder’s queue/partition buffer; throws with no queue name. get_buffer_stats() on the client returns the aggregate. Buffers are keyed on "<queue>/<partition>", so builders addressing the same pair share one, and a buffer flushes at message_count messages or time_millis after its first message.

Dead-letter queue

dlq(const std::string& consumer_group = "") returns a DLQBuilder; throws with no queue name.

Read-only, and this client has no admin surface, so replay means reading the row and pushing the payload again, or calling POST /api/v1/messages/{partitionId}/{transactionId}/retry yourself.

TransactionBuilder

client.transaction(). Pushes and acks in one PostgreSQL transaction, all-or-nothing.

HttpClient

There is no Admin class in the C++ client. Management and observability routes are reached through the shared HttpClient from get_http_client():

auto http = client.get_http_client();
queen::json overview = http->get("/api/v1/resources/overview");

All four return json, with nullptr for an empty body or a 204 (the broker sends no body at all on 204), and throw HttpError on any status >= 400. request_timeout_millis of 0 uses the client default. Pass RetryKind::Pop for a long-poll request so a 429 backs off indefinitely instead of using the bounded budget.

The complete route list is in Reference.

HttpError

Derives from std::runtime_error, with what() set to the broker’s error string, so existing catch (const std::exception&) sites keep working.

A transport failure with no response at all surfaces as a plain std::runtime_error, not an HttpError.

LoadBalancer

Constructed automatically when the two-argument QueenClient gets more than one URL; throws std::invalid_argument on an empty vector. get_next_url(session_key = "") returns a sticky URL for "session" and a round-robin URL otherwise. get_all_urls(), get_strategy() and reset() are also public.

util helpers

Not in this client

There is no streaming SDK for C++ and no Admin facade. The Stream builder, windows, gates and the /streams/v1/* runtime exist in the JavaScript, Python and Go clients only.

Same surface, other languages