BabelQueue for Java

Polyglot Queues, Simplified. Read and write the canonical BabelQueue message envelope from Java — so your Java/Spring services exchange messages with Laravel, Symfony, Python, Go and Node over one strict JSON format, on the broker you already run.

This is the framework-agnostic Java core: the wire-envelope codec, contracts and dead-letter helpers — zero dependencies (pure JDK, including its own minimal JSON codec, so no Jackson/Gson is forced on you). The full standard is documented at babelqueue.com.

Installation

Maven:

<dependency>
<groupId>com.babelqueue</groupId>
<artifactId>babelqueue-core</artifactId>
<version>1.0.0</version>
</dependency>

Gradle:

implementation("com.babelqueue:babelqueue-core:1.0.0")

Requires Java 17+.

Usage

importcom.babelqueue.*;
importjava.util.Map;
// Produce — build the canonical envelope and publish the JSON to your broker.Envelopeenv = EnvelopeCodec.make(
"urn:babel:orders:created",
Map.of("order_id", 1042L),
"orders",
null);
Stringbody = EnvelopeCodec.encode(env); // compact UTF-8 JSON// jedis.rpush("queues:orders", body);// / channel.basicPublish("", "orders", props, body.getBytes(StandardCharsets.UTF_8));// Consume — decode a message produced by ANY BabelQueue SDK.Envelopein = EnvelopeCodec.decode(body);
if (EnvelopeCodec.accepts(in)) {
switch (EnvelopeCodec.urn(in)) {
case"urn:babel:orders:created" ->
System.out.println(in.data().get("order_id") + " " + in.traceId());
default -> { /* unknown URN */ }
}
}

The envelope is identical to every other SDK's:

{
"job": "urn:babel:orders:created",
"trace_id": "…",
"data": { "order_id": 1042 },
"meta": { "id": "…", "queue": "orders", "lang": "java", "schema_version": 1, "created_at": 1749132727000 },
"attempts": 0
}

JSON numbers decode into data as Long (integers) or Double (decimals); objects as LinkedHashMap (insertion order preserved). encode leaves slashes and non-ASCII unescaped, so the bytes match the PHP/Python/Node cores.

Typed messages (optional)

recordOrderCreated(longorderId) implementsPolyglotMessage, HasTraceId {
publicStringgetBabelUrn() { return"urn:babel:orders:created"; }
publicMap<String, Object> toPayload() { returnMap.of("order_id", orderId); }
publicStringgetBabelTraceId() { returnnull; } // or an inbound trace to continue
}
Envelopeenv = EnvelopeCodec.fromMessage(newOrderCreated(1042L), "orders");

Dead-letter

Envelopedlq = DeadLetters.annotate(env, "failed", "orders", 3, "boom", "java.lang.RuntimeException");
// publish EnvelopeCodec.encode(dlq) to the "orders.dlq" queue

DeadLetters.annotate returns a copy — the original envelope is preserved unchanged inside the dead-lettered message, so any-language consumers can still read it.

Transactional outbox (optional)

The com.babelqueue.outbox helper (ADR-0029) removes the producer dual write — "commit the business row" and "publish to the broker" are two systems that can disagree on a crash. Instead the message is written into your own database, inside your own transaction, so it commits or rolls back atomically with the business data; a separate relay publishes the durable rows afterwards.

importcom.babelqueue.outbox.*;
// Bind the OutboxStore to YOUR database (a JDBC adapter writes the row on `connection`).// The core ships only an in-memory reference — it pulls in no DB driver (GR-7).OutboxStorestore = /* your JDBC-backed store */;
Outboxoutbox = newOutbox(store);
// 1) Write side — the CALLER owns the transaction boundary (this is the whole point):connection.setAutoCommit(false);
try {
insertOrder(connection, order); // the business writeEnvelopeenv = EnvelopeCodec.make("urn:babel:orders:created", data, "orders", null);
outbox.write(env); // same connection, same txconnection.commit(); // both, or neither
} catch (Exceptione) {
connection.rollback();
throwe;
}
// 2) Relay side — run on a short interval AFTER the tx commits. Publish the stored bytes// verbatim through your broker; mark published only after the transport accepts them.OutboxTransporttransport = (queue, body) -> jedis.rpush("queues:" + queue, newString(body, UTF_8));
OutboxRelayrelay = newOutboxRelay(transport, store);
OutboxRelayResultresult = relay.drain(0); // loop until the outbox is empty

Outbox.write stores the EnvelopeCodec-encoded bytes verbatim (GR-1, the envelope never changes) and the relay publishes those exact bytes — so trace_id is preserved end-to-end (GR-4) and the body is byte-identical before store and after relay (GR-5). A publish that throws is recorded (markFailed) with a bounded backoff and left pending for the next pass; one poison row never blocks the batch. This is exactly-once handoff into the broker, then at-least-once on the wire as always — consumers still dedupe on meta.id (the consumer-side com.babelqueue.idempotency helper is the mirror). Relay concurrency (SELECT … FOR UPDATE SKIP LOCKED) is the adapter's job; the in-memory reference store does not implement it.

GDPR field encryption (optional)

The com.babelqueue.gdpr helper (ADR-0030) is the runtime, SDK-enforcement half of the registry's x-gdpr-sensitive declaration: a producer encrypts each marked data field before publish, a consumer decrypts it after decode. The registry only declares which fields are personal data; this enforces it on the wire. It is standalone and opt-in — call it, or don't.

importcom.babelqueue.gdpr.*;
importcom.babelqueue.schema.*;
// The Cipher is YOURS — a seam onto KMS/Vault/HSM/tokenisation. The core ships a JDK-only// reference (AES-256-GCM, random 12-byte IV prepended, Base64) so it pulls no crypto dep (GR-7).Ciphercipher = newAesGcmCipher(key); // key is 16/24/32 bytes; the caller owns itMap<String, Object> schema = provider.schemaFor(env.job()); // the same per-URN schema you validate againstif (schema != null) {
// Producer — validate CLEARTEXT first, then encrypt the marked leaves in place:SchemaValidation.validate(provider, env.job(), env.data());
Gdpr.protect(env.data(), schema, cipher);
}
Stringbody = EnvelopeCodec.encode(env); // ciphertext rides inside data// Consumer — decrypt the marked leaves in place AFTER decode, BEFORE the handler reads data:Envelopein = EnvelopeCodec.decode(body);
Map<String, Object> inSchema = provider.schemaFor(in.job());
if (inSchema != null) {
Gdpr.unprotect(in.data(), inSchema, cipher); // wrong key → DecryptException (retry/DLQ)
}

The wire envelope stays frozen (GR-1): only the value of a sensitive field changes — it becomes a ciphertext string, so data is still pure JSON (GR-3) and any SDK can carry the envelope even without the key. meta.schema_version stays 1 and trace_id is untouched (GR-4). Each leaf is canonically JSON-encoded before encryption and decoded back after, so protect → unprotect restores the value byte-for-byte (a number comes back a number, an object an object). The sensitive paths come from the schema's x-gdpr-sensitive marks (SensitivePaths.of(schema) — nested objects, array items field[], and the root), which are validation-neutral so annotating a schema is never a breaking change. Validate cleartext beforeprotect / afterunprotect: a schema that constrains a sensitive field (minLength, enum, …) would reject the ciphertext string otherwise.

What this core is (and isn't)

It enforces the contract: the envelope shape, URN identity, trace propagation, schema-version gating and the dead-letter block. It is intentionally not a worker/runtime — broker wiring, acks and retry loops stay in your own code (or a future Spring adapter), exactly as with the other SDK cores.

UnknownUrnStrategy (FAIL, DELETE, RELEASE, DEAD_LETTER) is provided for adapters to act on.

Conformance

This core passes the shared cross-SDK conformance suite (vendored under src/test/resources/conformance/) — the same fixtures every BabelQueue SDK must satisfy, so a Java producer and, say, a Laravel consumer agree byte-for-byte.

mvn test

License

MIT © Muhammet Şafak