Client SDKs in six languages: install and connect

QueenMQ speaks JSON over HTTP on port 6632, so every client is a thin layer over the same routes, and curl is a complete client. An SDK adds the parts that are tedious by hand: consume loops that ack for you, lease renewal, push buffering, retries with failover across the nodes of a cluster, and builders for transactions, KV, timers and locks.

Install and connect

npm install queen-mq

import { Queen } from 'queen-mq'

const queen = new Queen({ url: 'http://localhost:6632' })
// ...
await queen.close()

pip install queen-mq

from queen import Queen

async def main():
    queen = Queen(url="http://localhost:6632")
    # ... every call is awaited
    await queen.close()

go get github.com/smartpricing/queen/clients/client-go/v2

import queen "github.com/smartpricing/queen/clients/client-go/v2"

client, err := queen.New("http://localhost:6632")
if err != nil {
	return err
}
defer client.Close(context.Background())

cargo add queen-mq serde_json
cargo add tokio --features rt-multi-thread,macros

use queen_mq::{Config, Queen};

let queen = Queen::connect(Config::new("http://localhost:6632"))?;

#include "queen_client.hpp"

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

g++ -std=c++17 -O2 -pthread \
    -I clients/client-cpp -I clients/server/vendor \
    -I /opt/homebrew/include -I "$(brew --prefix openssl)/include" \
    app.cpp -o app \
    -L "$(brew --prefix openssl)/lib" -lssl -lcrypto -lpthread

composer require queen-mq/php-client

use Queen\Queen;

$queen = new Queen('http://localhost:6632');

QUEUE_CONNECTION=queen
QUEEN_URL=http://127.0.0.1:6632

The six clients are at 2.6.0 and talk to any 2.x broker. Locks need a 2.1 broker, and the binary forms a 2.3 one.

How each SDK names things

These pages name options and methods as the JavaScript client spells them. The others follow the habits of their language:

Where the SDKs differ

The core is the same everywhere: push, pop, consume, ack, transactions, KV, timers and locks. These are the differences the programs behind these pages ran into, on 2.6.0:

Where an SDK lacks a method, the route is still there over HTTP: the C++ code on these pages calls a few routes directly, and says why in a comment.

Parity here is a claim about the surface, not about shared code: each client is written natively in its own idiom. These nineteen rows are that surface, read out of clients/ on this tree. Nine of them are full parity across all six clients; the other ten are where a client stops.

The ten divergent rows, with their conditions:

• Key/value. The eight operations are get, getMany, getPrefix, put, putIfAbsent, delete, incr and check. C++ wraps six of them: getMany and getPrefix have no method, and both remain reachable through POST /api/v1/kv on the shared HttpClient. getPrefix inside a transaction is refused by the broker in every client, which is its rule and not a client’s gap.
• Timers. C++ has schedule and cancel only; peek and list are reads and stay on GET /api/v1/timers/{queue}[/{timerKey}]. There is no reschedule operation anywhere, because schedule is the upsert and status reports which happened; Python, Go, Rust and PHP still spell a reschedule alias over it, JavaScript and C++ do not.
• once. JavaScript (kv.once, transaction().once) and Python (kv.once, transaction().once) fold putIfAbsent plus required into the question people actually ask, “did I win?”. The other four write the putIfAbsent themselves, which is the same wire and one line longer.
• Lock renewal. A lock is a lease its holder renews. The JavaScript, Python, Go and Rust handles renew by themselves, every third of the lifetime, and signal a lock that was lost. PHP and C++ have no background task to renew from: their handle has keepAlive() (keep_alive() in C++), a checkpoint to call inside the work loop, which renews when a third of the lifetime has passed and returns false once the lock is gone. A PHP or C++ holder that never calls it keeps the lock for one lifetime only.
• kv on a streams operator context. No client has it, so it is not a row. Inside a stream the state primitive is state_ops, which commits with the sink and the ack in the cycle’s own transaction; a key/value write from an operator would not, and that atomicity is the thing the stream already gives you for free.
• Admin facade. C++ has no Admin class. Every management and observability route is still reachable, through the shared HttpClient returned by get_http_client().
• affinity. The C++ LoadBalancer implements round-robin and "session" only, and load_balancing_strategy = "affinity" falls through to round-robin silently. The strategy matters because it keeps one consumer’s pops on one backend, which is what keeps two clients from contending on the same partition claim.
• Streams. The Stream builder, the four window kinds, gates and the /streams/v1/* runtime are in the JavaScript, Python, Go and Rust clients. PHP and C++ have none of it.
• Per-message trace. Where it exists it is attached by the consume loop, so a message returned by a manual pop() never carries it. Rust records a trace through Admin::record_trace instead of hanging a method off the message. C++ cannot hang one off a JSON object at all, and its trace hook is an explicit no-op.
• Host override. JavaScript (hostHeader) and Rust (host_header) advertise a request authority, and a TLS SNI, independent of the address the socket dials. That is what addresses a named tenant cluster behind a shared proxy endpoint. The other four clients send the dialled address.
• pop failures. Go, Rust and PHP surface a 4xx, an exhausted 429 budget, a terminal 403 or a network fault to the caller, so an empty result means an empty queue. JavaScript, Python and C++ log the failure and return an empty result, which does not distinguish an empty queue from revoked credentials.

curl and queenctl

Every SDK call is one of the broker’s routes, so anything that can send an HTTP request is a client:

curl -s -X POST http://localhost:6632/api/v1/push -H 'content-type: application/json' \
  -d '{"items":[{"queue":"orders","partition":"customer-123","payload":{"orderId":8891}}]}'

The HTTP reference lists every route. queenctl, the operator CLI (status, tail, push, replay, lag, dead letters), ships inside the Docker image (docker exec queen queenctl status) and as release binaries. See queenctl.

Binary forms, by SDK

The three options of binary push, compression and framed pop are spelled:

Reads by offset, by SDK

fetch reads a partition by position, and changedSince lists the partitions written since a time. Neither takes a lease or moves a cursor.

changedSince answers a safeTime: give it as the next since to see only what changed after this call. It is the broker’s time, so never compare it with your own clock.

Consumer supervision

Every SDK can report what its consumers are doing (workers, handler progress) to the dashboard, through the KV API. It is off by default; turn it on per consume loop with .supervision({ group: 'billing-production' }), spelled as above in each language. Each run of a loop is one instance with fresh counters, published every 10 seconds (PHP: when the loop gets control back). The status contract defines what is published.

The broker inside a Rust program

A Rust program can run the broker in its own process, as the library queen::Broker. Every call goes through the same handler functions the HTTP router dispatches to, so behaviour and defaults are the broker’s, and the data lives in a directory on local disk that you name.

2.0 is not on crates.io yet (the queen-engine published there is 1.3, which ran on PostgreSQL), so depend on the repository, pinned to the 2.0.0 commit:

[dependencies]
queen-engine = { git = "https://github.com/queen-mq/queen", rev = "86562c25d", default-features = false }
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }
serde_json = "1"

The crate imports as queen and needs Rust 1.88. default-features = false leaves out what only the standalone binary needs: the HTTP listener, the dashboard, the embedded proxy and the Kafka facade. Start it on a data directory:

let broker = Broker::start(BrokerConfig::new().raft(&dir))
    .await
    .expect("broker start");

Push, with a transactionId on the first item:

let first = broker
    .push(vec![
        qp::PushItem::new(q.clone(), serde_json::json!({"n": 1}))
            .transaction_id(txn_id.clone()),
        qp::PushItem::new(q.clone(), serde_json::json!({"n": 2})),
        qp::PushItem::new(q.clone(), serde_json::json!({"n": 3})),
    ])
    .await
    .expect("push");

Ack one message and push the next in one transaction:

let txn = broker
    .transaction(
        &qp::TransactionRequest::new(vec![
            qp::TxnOperation::Ack(qp::TxnAckOperation {
                transaction_id: m0.transaction_id.clone(),
                partition_id: m0.partition_id.clone(),
                status: qp::AckStatus::Completed,
                consumer_group: Some(group.to_string()),
                lease_id: Some(m0.lease_id.clone()),
                error: None,
            }),
            qp::TxnOperation::Push {
                items: vec![qp::TxnPushItem::new(
                    q2.clone(),
                    serde_json::json!({"stage": 2}),
                )],
            },
        ])
        .with_required_leases([m0.lease_id.clone()]),
    )
    .await
    .expect("transaction");
assert!(txn.success, "transaction must commit: {txn:?}");

The embedded broker is one node and one tenant. It exposes push, pop, ack, lease renewal, transactions, queue configuration, the dead-letter queue and ephemeral queues. KV and timers are reachable only inside a transaction, so there is no standalone KV read or timer listing, and consumer-group administration, queue listings, traces and streams are not exposed. A panic in one of the broker’s core threads (planner, log writer, apply, checkpoint, raft) aborts the process, and the next start replays from the data directory; any other panic, your own included, unwinds as usual. Never open one data directory from two brokers at once.