Amazon SQS transport

com.babelqueue:babelqueue-sqs is an Amazon SQS transport on the Java core. It sends the canonical envelope as the MessageBody with the §3 SQS MessageAttributes, and consumes by routing each message to a handler by URN — so a message it produces is consumed by any other BabelQueue SDK, and vice-versa.

Install

Maven:

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

Gradle:

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

Requirements: Java 17+. It pulls babelqueue-core and the AWS SDK for Java v2 (software.amazon.awssdk:sqs) transitively. You supply the AWS SqsClient (an interface), so its credentials/region come from your standard AWS configuration chain.

Produce

import com.babelqueue.sqs.SqsPublisher;
import java.util.Map;
import software.amazon.awssdk.services.sqs.SqsClient;

SqsClient sqs = SqsClient.create(); // your AWS config / credentials chain
String url = "https://sqs.eu-central-1.amazonaws.com/123456789012/orders";

String id = SqsPublisher.create(sqs, url)
    .publish("urn:babel:orders:created", Map.of("order_id", 1042L));

publish(urn, data) returns the message meta.id; publish(urn, data, traceId) continues an existing trace. FIFO queues use the builder: SqsPublisher.builder(sqs, url).fifo(true).build() (the queue URL must end in .fifo).

Consume

Register handlers by URN; poll() receives one batch, routes each message, and deletes the ones handled. A throwing handler leaves the message for SQS to redeliver after the visibility timeout (at-least-once):

import com.babelqueue.sqs.SqsConsumer;

SqsConsumer consumer = SqsConsumer.builder(sqs, url)
    .handler("urn:babel:orders:created", (envelope, message) -> {
        var orderId = envelope.data().get("order_id");
        // ... use envelope.traceId(), envelope.data()
    })
    .onUnknownUrn((envelope, message) -> { /* no handler for this URN */ })
    .onError((error, envelope, message) -> { /* loop never stops */ })
    .build();

while (running) {
    consumer.poll();
}

attempts is reconciled to ApproximateReceiveCount − 1 for the handler (never lowering a runtime-incremented count). Point the SqsClient’s endpoint at LocalStack/ElasticMQ for local testing.

Because the wire format is the canonical envelope, the message a handler receives may have been produced by any BabelQueue SDK.