diff --git a/CHANGELOG.md b/CHANGELOG.md index 6ed6edc..76ba398 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,14 @@ The envelope wire format is versioned separately by `meta.schema_version` ## [Unreleased] +## [1.2.0] - 2026-11-03 + +### Added + +- **Dead-letter observer (ADR-0039 section 5.1).** `AsbConsumer.Builder#deadLetterObserver(DeadLetterObserver)` takes the core's optional `com.babelqueue.DeadLetterObserver` (needs a `babelqueue-core` that ships it). It is called once per dead-letter this consumer carried out, after it is confirmed: a record sent to the BabelQueue DLQ sender (poison body with `reason` `poison`, raw version-gated body with an empty `reason`), and the SDK-initiated native `deadLetter` call made for a version-gated delivery when no DLQ sender is set (`deadLetterQueue` and `reason` empty, as the SDK writes no `dead_letter` block; no extra attribute). `DeadLetterEvent.queue` is the receiver's entity. It is not called when the send or the native call fails, for a broker-initiated dead-letter (`MaxDeliveryCount`), or when the message is only abandoned; an observer failure is logged and never changes the settle outcome. Without an observer nothing changes. +- **W3C `traceparent` out-of-band propagation (ADR-0028).** `AsbPublisher`: `publishWithHeaders(Envelope, Map)` (the produce-side seam for the core's `com.babelqueue.otel.HeaderSender`) and a `publish(..., Map headers)` overload carry out-of-band headers (e.g. `traceparent`, `tracestate`) as string `ApplicationProperties` beside the contract `bq-` entries. A contract entry always wins a collision; blank keys/values are skipped; with no/empty headers the message is identical to a plain `publish`. The envelope body is untouched (GR-1, `schema_version` stays 1). +- `BabelHandler.headersOf(ServiceBusReceivedMessage)` surfaces the delivered string `ApplicationProperties` as a `Map` — the `Supplier` source for `Tracing.wrapHandler(tracer, handler, headers)`, so the consumer span becomes a true child of the producer span. No OpenTelemetry dependency is added (the core's OTel API stays optional). + ## [1.1.1] - 2026-10-26 ### Changed diff --git a/README.md b/README.md index b75ba97..8e4464d 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,7 @@ propagation) as the .NET, Python, Go and Node SDKs. Implements com.babelqueue babelqueue-azure-servicebus - 1.1.1 + 1.2.0 ``` diff --git a/pom.xml b/pom.xml index 61aa7cf..ddba56b 100644 --- a/pom.xml +++ b/pom.xml @@ -6,7 +6,7 @@ com.babelqueue babelqueue-azure-servicebus - 1.1.1 + 1.2.0 jar BabelQueue Azure Service Bus @@ -49,7 +49,7 @@ 17 UTF-8 UTF-8 - 1.14.0 + 1.15.0 7.17.5 5.10.3 5.12.0 diff --git a/src/main/java/com/babelqueue/azureservicebus/AsbConsumer.java b/src/main/java/com/babelqueue/azureservicebus/AsbConsumer.java index c84fd99..bcddfdc 100644 --- a/src/main/java/com/babelqueue/azureservicebus/AsbConsumer.java +++ b/src/main/java/com/babelqueue/azureservicebus/AsbConsumer.java @@ -7,6 +7,8 @@ import com.azure.messaging.servicebus.ServiceBusSenderClient; import com.azure.messaging.servicebus.models.DeadLetterOptions; import com.babelqueue.BabelQueueException; +import com.babelqueue.DeadLetterEvent; +import com.babelqueue.DeadLetterObserver; import com.babelqueue.Envelope; import com.babelqueue.EnvelopeCodec; import com.babelqueue.PoisonRecord; @@ -79,6 +81,7 @@ public interface UnknownUrnHandler { private final ErrorHandler onError; private final UnknownUrnHandler onUnknownUrn; private final ServiceBusSenderClient deadLetterSender; + private final DeadLetterObserver deadLetterObserver; private AsbConsumer(Builder builder) { this.receiver = builder.receiver; @@ -88,6 +91,7 @@ private AsbConsumer(Builder builder) { this.onError = builder.onError; this.onUnknownUrn = builder.onUnknownUrn; this.deadLetterSender = builder.deadLetterSender; + this.deadLetterObserver = builder.deadLetterObserver; } public static Builder builder(ServiceBusReceiverClient receiver) { @@ -131,6 +135,8 @@ private void handle(ServiceBusReceivedMessage message) { receiver.deadLetter(message, new DeadLetterOptions() .setDeadLetterReason(SCHEMA_VERSION_DEAD_LETTER_REASON) .setDeadLetterErrorDescription(description)); + // The SDK itself dead-lettered it natively (ADR-0039 section 5.1 (2)): counted, with no BabelQueue DLQ. + notifyDeadLetter("", ""); report(new BabelQueueException(description), null, message); return; } @@ -144,6 +150,7 @@ private void handle(ServiceBusReceivedMessage message) { report(sendFailure, null, message); return; } + notifyDeadLetter(deadLetterSender.getEntityPath(), ""); receiver.complete(message); report(new BabelQueueException(description), null, message); return; @@ -216,6 +223,7 @@ private void settlePoison(ServiceBusReceivedMessage message, Envelope envelope, report(sendFailure, envelope, message); return; } + notifyDeadLetter(deadLetterSender.getEntityPath(), "poison"); receiver.complete(message); } @@ -266,6 +274,19 @@ private static ServiceBusMessage rawMessage(ServiceBusReceivedMessage message) { return forwarded; } + /** + * Reports a dead-letter the SDK just carried out (ADR-0039 section 5.1): a confirmed send to the BabelQueue DLQ, or + * a native {@code deadLetter} call that returned. {@code queue} is the receiver's entity (a version-gated message is + * never decoded, so there is no {@code meta.queue}); {@code deadLetterQueue} is empty for the native dead-letter. + * An observer failure never escapes. + */ + private void notifyDeadLetter(String deadLetterQueue, String reason) { + if (deadLetterObserver == null) { + return; + } + DeadLetterObserver.notifySafely(deadLetterObserver, new DeadLetterEvent(entityName(), deadLetterQueue, reason)); + } + private void send(ServiceBusMessage message) { deadLetterSender.sendMessage(message); } @@ -356,6 +377,7 @@ public static final class Builder { private ErrorHandler onError; private UnknownUrnHandler onUnknownUrn; private ServiceBusSenderClient deadLetterSender; + private DeadLetterObserver deadLetterObserver; private Builder(ServiceBusReceiverClient receiver) { this.receiver = Objects.requireNonNull(receiver, "receiver"); @@ -406,6 +428,19 @@ public Builder deadLetterSender(ServiceBusSenderClient sender) { return this; } + /** + * Optional observer called once per dead-letter this consumer carried out (ADR-0039 section 5.1), after it is + * confirmed: a record sent to the BabelQueue DLQ (poison body, raw version-gated body), or the native + * {@code deadLetter} call made for a version-gated delivery when no {@link #deadLetterSender sender} is set + * ({@code deadLetterQueue} and {@code reason} are then empty: the SDK writes no {@code dead_letter} block). Not called when + * the send or the native call fails, or for a broker-initiated dead-letter ({@code MaxDeliveryCount}). A failure + * of the observer is logged and never changes the settle outcome. + */ + public Builder deadLetterObserver(DeadLetterObserver deadLetterObserver) { + this.deadLetterObserver = deadLetterObserver; + return this; + } + public AsbConsumer build() { return new AsbConsumer(this); } diff --git a/src/main/java/com/babelqueue/azureservicebus/AsbMessages.java b/src/main/java/com/babelqueue/azureservicebus/AsbMessages.java index 2ab027d..93a50e2 100644 --- a/src/main/java/com/babelqueue/azureservicebus/AsbMessages.java +++ b/src/main/java/com/babelqueue/azureservicebus/AsbMessages.java @@ -2,8 +2,13 @@ import com.azure.core.util.BinaryData; import com.azure.messaging.servicebus.ServiceBusMessage; +import com.azure.messaging.servicebus.ServiceBusReceivedMessage; import com.babelqueue.Envelope; import com.babelqueue.EnvelopeCodec; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; /** * Projects the envelope's contract fields onto native Service Bus message fields — @@ -34,4 +39,47 @@ static ServiceBusMessage project(Envelope envelope) { } return message; } + + /** + * Adds the out-of-band {@code extra} headers (e.g. {@code traceparent}, ADR-0028) to {@code message} as string + * {@code ApplicationProperties} beside the contract {@code bq-} properties. Blank keys/values are skipped, a + * property already present always wins (the contract is never clobbered), and keys are added in sorted order + * so the result is deterministic. A {@code null}/empty map leaves the message untouched. + */ + static void addOutOfBand(ServiceBusMessage message, Map extra) { + if (extra == null || extra.isEmpty()) { + return; + } + Map properties = message.getApplicationProperties(); + List keys = new ArrayList<>(extra.keySet()); + keys.removeIf(k -> k == null || k.isEmpty()); + keys.sort(String::compareTo); + for (String key : keys) { + String value = extra.get(key); + if (value != null && !value.isEmpty() && !properties.containsKey(key)) { + properties.put(key, value); + } + } + } + + /** + * A delivered message's {@code ApplicationProperties} as a flat {@code Map} (string-typed, + * non-blank values only) — the consume-side counterpart of {@link #addOutOfBand}. Both the contract + * {@code bq-} properties that are strings and out-of-band ones (e.g. {@code traceparent}) surface; numeric + * contract properties ({@code bq-schema-version}, {@code bq-created-at}) are not strings and are left out. + * Empty when the message is {@code null} or has none. + */ + static Map toMap(ServiceBusReceivedMessage message) { + Map out = new LinkedHashMap<>(); + Map properties = message == null ? null : message.getApplicationProperties(); + if (properties == null) { + return out; + } + for (Map.Entry e : properties.entrySet()) { + if (e.getKey() != null && e.getValue() instanceof CharSequence value && value.length() > 0) { + out.put(e.getKey(), value.toString()); + } + } + return out; + } } diff --git a/src/main/java/com/babelqueue/azureservicebus/AsbPublisher.java b/src/main/java/com/babelqueue/azureservicebus/AsbPublisher.java index 384759e..20ba9c2 100644 --- a/src/main/java/com/babelqueue/azureservicebus/AsbPublisher.java +++ b/src/main/java/com/babelqueue/azureservicebus/AsbPublisher.java @@ -22,6 +22,10 @@ * .connectionString(cs).sender().queueOrTopicName("orders").buildClient(); * String id = AsbPublisher.create(sender).publish("urn:babel:orders:created", Map.of("order_id", 1042)); * } + * + *

Out-of-band transport headers (e.g. the W3C {@code traceparent}, ADR-0028) ride as extra string + * {@code ApplicationProperties} beside the contract {@code bq-} properties — never inside the envelope + * (GR-1); see {@link #publishWithHeaders(Envelope, Map)}, the seam for {@code com.babelqueue.otel.HeaderSender}. */ public final class AsbPublisher { @@ -51,13 +55,40 @@ public String publish(String urn, Map data, String traceId) { * {@code ScheduledEnqueueTime} (the broker withholds the message until {@code now + delay}). */ public String publish(String urn, Map data, String traceId, Duration delay) { + return publish(urn, data, traceId, delay, null); + } + + /** + * Publish as {@link #publish(String, Map, String, Duration)}, additionally carrying the out-of-band + * {@code headers} (e.g. {@code traceparent}) as string {@code ApplicationProperties} beside the contract + * {@code bq-} properties. A contract property already present always wins a collision; blank keys or values + * are skipped; {@code null}/empty {@code headers} is byte-identical to the plain overload. + */ + public String publish(String urn, Map data, String traceId, Duration delay, + Map headers) { Envelope envelope = EnvelopeCodec.make(urn, data, sender.getEntityPath(), traceId); + return send(envelope, delay, headers); + } + + /** + * Publish an already-built {@code envelope} together with out-of-band transport {@code headers} + * (e.g. a W3C {@code traceparent}, ADR-0028). This is the produce-side seam the optional + * {@code com.babelqueue.otel.HeaderSender} wires to. The body stays the byte-identical envelope JSON. + * Returns {@code meta.id}. + */ + public String publishWithHeaders(Envelope envelope, Map headers) { + return send(envelope, null, headers); + } + + private String send(Envelope envelope, Duration delay, Map headers) { ServiceBusMessage message = AsbMessages.project(envelope); if (delay != null && !delay.isZero() && !delay.isNegative()) { message.getApplicationProperties().put("bq-delay", delay.toMillis()); + AsbMessages.addOutOfBand(message, headers); sender.scheduleMessage(message, OffsetDateTime.now(ZoneOffset.UTC).plus(delay)); } else { + AsbMessages.addOutOfBand(message, headers); sender.sendMessage(message); } return envelope.meta().id(); diff --git a/src/main/java/com/babelqueue/azureservicebus/BabelHandler.java b/src/main/java/com/babelqueue/azureservicebus/BabelHandler.java index f1d778f..8d9c69f 100644 --- a/src/main/java/com/babelqueue/azureservicebus/BabelHandler.java +++ b/src/main/java/com/babelqueue/azureservicebus/BabelHandler.java @@ -2,8 +2,23 @@ import com.azure.messaging.servicebus.ServiceBusReceivedMessage; import com.babelqueue.Envelope; +import java.util.Map; -/** Processes one decoded, validated envelope and the raw Service Bus message it arrived on. */ +/** + * Processes one decoded, validated envelope and the raw Service Bus message it arrived on. + * + *

The raw message carries the out-of-band transport headers (ADR-0028) as {@code ApplicationProperties}. + * {@link #headersOf} surfaces them as a plain map, ready for the optional OpenTelemetry wrapper so the + * consumer span becomes a true child of the producer span: + * + * 

{@code
+ * AsbConsumer.builder(receiver)
+ *     .handler(urn, (env, message) -> Tracing
+ *         .wrapHandler(tracer, h, () -> BabelHandler.headersOf(message))
+ *         .handle(env))
+ *     .build();
+ * }
+ */ @FunctionalInterface public interface BabelHandler { @@ -13,4 +28,14 @@ public interface BabelHandler { * incrementing {@code DeliveryCount}). */ void handle(Envelope envelope, ServiceBusReceivedMessage message) throws Exception; + + /** + * The message's string-typed {@code ApplicationProperties} as a {@code Map} (non-blank values + * only) — the delivered-headers {@code Supplier} source for {@code com.babelqueue.otel.Tracing#wrapHandler}. + * Contains any out-of-band header (e.g. {@code traceparent}, {@code tracestate}). Empty when {@code message} + * is {@code null}. Needs no OpenTelemetry dependency. + */ + static Map headersOf(ServiceBusReceivedMessage message) { + return AsbMessages.toMap(message); + } } diff --git a/src/test/java/com/babelqueue/azureservicebus/AsbConsumerTest.java b/src/test/java/com/babelqueue/azureservicebus/AsbConsumerTest.java index 33afe5d..14b060f 100644 --- a/src/test/java/com/babelqueue/azureservicebus/AsbConsumerTest.java +++ b/src/test/java/com/babelqueue/azureservicebus/AsbConsumerTest.java @@ -22,6 +22,8 @@ import com.azure.messaging.servicebus.ServiceBusSenderClient; import com.azure.messaging.servicebus.models.DeadLetterOptions; import com.babelqueue.BabelQueueException; +import com.babelqueue.DeadLetterEvent; +import com.babelqueue.DeadLetterObserver; import com.babelqueue.DeadLetters; import com.babelqueue.Envelope; import com.babelqueue.EnvelopeCodec; @@ -498,4 +500,171 @@ void versionGatedHookOnlyConsumerNeverRunsTheHookAndDeadLettersNatively() { verify(receiver).deadLetter(eq(message), any(DeadLetterOptions.class)); verify(receiver, never()).complete(any(ServiceBusReceivedMessage.class)); } + + @Test + @SuppressWarnings("unchecked") + void publishedTraceparentReachesTheHandlerViaHeadersOfAndNotTheBody() { + String traceparent = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"; + com.azure.messaging.servicebus.ServiceBusSenderClient sender = + mock(com.azure.messaging.servicebus.ServiceBusSenderClient.class); + when(sender.getEntityPath()).thenReturn("orders"); + org.mockito.ArgumentCaptor captor = + org.mockito.ArgumentCaptor.forClass(com.azure.messaging.servicebus.ServiceBusMessage.class); + AsbPublisher.create(sender).publish(URN, Map.of("order_id", 1), "trace-1", null, + Map.of("traceparent", traceparent)); + verify(sender).sendMessage(captor.capture()); + com.azure.messaging.servicebus.ServiceBusMessage sent = captor.getValue(); + + // the broker hands the same application properties and body back + ServiceBusReceivedMessage delivered = received(1, sent.getBody().toString()); + when(delivered.getApplicationProperties()).thenReturn(sent.getApplicationProperties()); + Map[] seen = new Map[]{null}; + + AsbConsumer.builder(receiverWith(delivered)) + .handler(URN, (e, m) -> seen[0] = BabelHandler.headersOf(m)) + .build() + .poll(); + + assertEquals(traceparent, seen[0].get("traceparent")); + org.junit.jupiter.api.Assertions.assertFalse(sent.getBody().toString().contains("traceparent")); + } + + // ---- ADR-0039 section 5.1 dead-letter observer ---------------------------------------------- + + private static ServiceBusSenderClient senderTo(String entity) { + ServiceBusSenderClient sender = mock(ServiceBusSenderClient.class); + when(sender.getEntityPath()).thenReturn(entity); + return sender; + } + + @Test + void observerSeesAPoisonRecordSentToTheBabelQueueDlq() { + ServiceBusReceivedMessage message = received(1, "{}"); + ServiceBusReceiverClient receiver = receiverFor("orders", message); + List events = new ArrayList<>(); + + AsbConsumer.builder(receiver).deadLetterSender(senderTo("orders.dlq")) + .deadLetterObserver(events::add).build().poll(); + + assertEquals(List.of(new DeadLetterEvent("orders", "orders.dlq", "poison")), events); + verify(receiver).complete(message); + } + + @Test + void observerSeesAVersionGatedRawBodySentToTheBabelQueueDlqWithAnEmptyReason() { + ServiceBusReceivedMessage message = withVersion("not json", 2, true); + ServiceBusReceiverClient receiver = receiverFor("orders", message); + List events = new ArrayList<>(); + + AsbConsumer.builder(receiver).deadLetterSender(senderTo("orders.dlq")) + .deadLetterObserver(events::add).build().poll(); + + assertEquals(List.of(new DeadLetterEvent("orders", "orders.dlq", "")), events); + verify(receiver).complete(message); + } + + @Test + void observerSeesTheSdkInitiatedNativeDeadLetterExactlyOnceWithAnEmptyDeadLetterQueue() { + ServiceBusReceivedMessage message = withVersion("not json", 2, true); + ServiceBusReceiverClient receiver = receiverFor("orders", message); + List events = new ArrayList<>(); + + AsbConsumer.builder(receiver).deadLetterObserver(events::add).build().poll(); + + assertEquals(List.of(new DeadLetterEvent("orders", "", "")), events); + assertEquals("", events.get(0).deadLetterQueue()); + verify(receiver, times(1)).deadLetter(eq(message), any(DeadLetterOptions.class)); + } + + @Test + void observerIsNotCalledWhenTheNativeDeadLetterCallFails() { + ServiceBusReceivedMessage message = withVersion("not json", 2, true); + ServiceBusReceiverClient receiver = receiverFor("orders", message); + org.mockito.Mockito.doThrow(new IllegalStateException("lock lost")).when(receiver) + .deadLetter(any(ServiceBusReceivedMessage.class), any(DeadLetterOptions.class)); + List events = new ArrayList<>(); + AsbConsumer consumer = AsbConsumer.builder(receiver).deadLetterObserver(events::add).build(); + + assertThrows(IllegalStateException.class, consumer::poll); + + assertTrue(events.isEmpty()); + } + + @Test + void observerIsNotCalledWhenTheDlqSendFails() { + List events = new ArrayList<>(); + for (ServiceBusReceivedMessage message : List.of(received(1, "{}"), withVersion("not json", 2, true))) { + ServiceBusReceiverClient receiver = receiverFor("orders", message); + ServiceBusSenderClient sender = senderTo("orders.dlq"); + org.mockito.Mockito.doThrow(new IllegalStateException("dlq down")).when(sender) + .sendMessage(any(ServiceBusMessage.class)); + + AsbConsumer.builder(receiver).deadLetterSender(sender).deadLetterObserver(events::add).build().poll(); + + verify(receiver).abandon(message); + } + + assertTrue(events.isEmpty()); + } + + @Test + void observerIsNotCalledForANonDeadLetterSettle() { + List events = new ArrayList<>(); + ServiceBusReceiverClient abandoned = receiverFor("orders", received(1, "{}")); + + // poison without a DLQ sender is abandoned (left to MaxDeliveryCount), and a handled message is completed + AsbConsumer.builder(abandoned).deadLetterObserver(events::add).build().poll(); + AsbConsumer.builder(receiverFor("orders", received(1, envelopeJson()))) + .handler(URN, (env, msg) -> { }) + .deadLetterSender(senderTo("orders.dlq")) + .deadLetterObserver(events::add).build().poll(); + AsbConsumer.builder(receiverFor("orders", received(1, envelopeJson()))) + .handler(URN, (env, msg) -> { throw new IllegalStateException("boom"); }) + .deadLetterSender(senderTo("orders.dlq")) + .deadLetterObserver(events::add).build().poll(); + + assertTrue(events.isEmpty()); + } + + @Test + void aThrowingObserverDoesNotChangeTheSettleOutcome() { + DeadLetterObserver boom = event -> { throw new IllegalStateException("observer down"); }; + DeadLetterObserver fatal = event -> { throw new AssertionError("observer asserts"); }; + + for (DeadLetterObserver observer : List.of(boom, fatal)) { + ServiceBusReceivedMessage poison = received(1, "{}"); + ServiceBusReceiverClient poisonReceiver = receiverFor("orders", poison); + AsbConsumer.builder(poisonReceiver).deadLetterSender(senderTo("orders.dlq")) + .deadLetterObserver(observer).build().poll(); + verify(poisonReceiver).complete(poison); + verify(poisonReceiver, never()).abandon(any(ServiceBusReceivedMessage.class)); + + ServiceBusReceivedMessage gated = withVersion("not json", 2, true); + ServiceBusReceiverClient gatedReceiver = receiverFor("orders", gated); + AsbConsumer.builder(gatedReceiver).deadLetterSender(senderTo("orders.dlq")) + .deadLetterObserver(observer).build().poll(); + verify(gatedReceiver).complete(gated); + + ServiceBusReceivedMessage nativeGated = withVersion("not json", 2, true); + ServiceBusReceiverClient nativeReceiver = receiverFor("orders", nativeGated); + AsbConsumer.builder(nativeReceiver).deadLetterObserver(observer).build().poll(); + verify(nativeReceiver, times(1)).deadLetter(eq(nativeGated), any(DeadLetterOptions.class)); + verify(nativeReceiver, never()).abandon(any(ServiceBusReceivedMessage.class)); + } + } + + @Test + void eventQueueIsAlwaysTheReceiverEntityWhateverTheBodyMetaQueueSays() { + // ASB reports the receiver's entity: a poison or version-gated body is never trusted for a queue name, so no + // blank rule applies. A Unicode-space meta.queue must not leak into the event. + for (String queue : new String[] {"\\u2003", "\\u00a0", " \\t", "billing"}) { + ServiceBusReceivedMessage message = received(1, "{\"meta\":{\"queue\":\"" + queue + "\"}}"); + List events = new ArrayList<>(); + + AsbConsumer.builder(receiverFor("orders", message)).deadLetterSender(senderTo("orders.dlq")) + .deadLetterObserver(events::add).build().poll(); + + assertEquals(List.of(new DeadLetterEvent("orders", "orders.dlq", "poison")), events, queue); + } + } } diff --git a/src/test/java/com/babelqueue/azureservicebus/AsbMessagesTest.java b/src/test/java/com/babelqueue/azureservicebus/AsbMessagesTest.java index aa92e92..22a68dc 100644 --- a/src/test/java/com/babelqueue/azureservicebus/AsbMessagesTest.java +++ b/src/test/java/com/babelqueue/azureservicebus/AsbMessagesTest.java @@ -41,4 +41,21 @@ void bodyIsTheCanonicalEnvelope() { assertTrue(EnvelopeCodec.accepts(decoded)); assertEquals("urn:babel:orders:created", EnvelopeCodec.urn(decoded)); } + + @Test + void toMapSurfacesStringPropertiesAndSkipsNumbersBlanksAndNull() { + com.azure.messaging.servicebus.ServiceBusReceivedMessage received = + org.mockito.Mockito.mock(com.azure.messaging.servicebus.ServiceBusReceivedMessage.class); + Map props = new java.util.LinkedHashMap<>(); + props.put("traceparent", "00-x"); + props.put("blank", ""); + props.put("bq-schema-version", 1); + org.mockito.Mockito.when(received.getApplicationProperties()).thenReturn(props); + + assertEquals(Map.of("traceparent", "00-x"), AsbMessages.toMap(received)); + assertTrue(AsbMessages.toMap(null).isEmpty()); + assertTrue(BabelHandler.headersOf(null).isEmpty()); + assertTrue(AsbMessages.toMap(org.mockito.Mockito.mock( + com.azure.messaging.servicebus.ServiceBusReceivedMessage.class)).isEmpty()); + } } diff --git a/src/test/java/com/babelqueue/azureservicebus/AsbPublisherTest.java b/src/test/java/com/babelqueue/azureservicebus/AsbPublisherTest.java index 220becf..7983cd8 100644 --- a/src/test/java/com/babelqueue/azureservicebus/AsbPublisherTest.java +++ b/src/test/java/com/babelqueue/azureservicebus/AsbPublisherTest.java @@ -1,6 +1,8 @@ package com.babelqueue.azureservicebus; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -9,6 +11,8 @@ import com.azure.messaging.servicebus.ServiceBusMessage; import com.azure.messaging.servicebus.ServiceBusSenderClient; +import com.babelqueue.Envelope; +import com.babelqueue.EnvelopeCodec; import java.time.Duration; import java.time.OffsetDateTime; import java.util.Map; @@ -57,4 +61,86 @@ void publishTwoArgMintsTrace() { verify(sender).sendMessage(captor.capture()); assertEquals("urn:babel:orders:created", captor.getValue().getSubject()); } + + private static final String TRACEPARENT = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"; + private static final String URN = "urn:babel:orders:created"; + + private static ServiceBusMessage sendWithHeaders(Envelope env, Map headers) { + ServiceBusSenderClient sender = mock(ServiceBusSenderClient.class); + when(sender.getEntityPath()).thenReturn("orders"); + ArgumentCaptor captor = ArgumentCaptor.forClass(ServiceBusMessage.class); + + AsbPublisher.create(sender).publishWithHeaders(env, headers); + + verify(sender).sendMessage(captor.capture()); + return captor.getValue(); + } + + @Test + void publishWithHeadersCarriesTraceparentBesideContractPropertiesAndNotInTheBody() { + Envelope env = EnvelopeCodec.make(URN, Map.of("order_id", 7), "orders", "trace-1"); + + ServiceBusMessage sent = sendWithHeaders(env, Map.of("traceparent", TRACEPARENT, "tracestate", "vendor=1")); + + assertEquals(TRACEPARENT, sent.getApplicationProperties().get("traceparent")); + assertEquals("vendor=1", sent.getApplicationProperties().get("tracestate")); + assertEquals(env.meta().schemaVersion(), sent.getApplicationProperties().get("bq-schema-version")); + assertEquals(URN, sent.getSubject()); + String body = sent.getBody().toString(); + assertFalse(body.contains("traceparent")); + assertFalse(body.contains(TRACEPARENT)); + } + + @Test + void bodyAndPropertiesAreIdenticalWithoutHeaders() { + Envelope env = EnvelopeCodec.make(URN, Map.of("order_id", 7), "orders", "trace-1"); + + ServiceBusMessage withHeaders = sendWithHeaders(env, Map.of("traceparent", TRACEPARENT)); + ServiceBusMessage plain = sendWithHeaders(env, null); + ServiceBusMessage empty = sendWithHeaders(env, Map.of("traceparent", "")); + + assertEquals(plain.getBody().toString(), withHeaders.getBody().toString()); + assertEquals(plain.getApplicationProperties(), empty.getApplicationProperties()); + assertFalse(plain.getApplicationProperties().containsKey("traceparent")); + } + + @Test + void contractPropertyWinsACollision() { + Envelope env = EnvelopeCodec.make(URN, Map.of(), "orders", "trace-1"); + + ServiceBusMessage sent = sendWithHeaders(env, Map.of("bq-schema-version", "99", "traceparent", TRACEPARENT)); + + assertEquals(env.meta().schemaVersion(), sent.getApplicationProperties().get("bq-schema-version")); + assertEquals(TRACEPARENT, sent.getApplicationProperties().get("traceparent")); + } + + @Test + void publishOverloadWithHeadersRidesTheScheduledMessageToo() { + ServiceBusSenderClient sender = mock(ServiceBusSenderClient.class); + when(sender.getEntityPath()).thenReturn("orders"); + when(sender.scheduleMessage(any(ServiceBusMessage.class), any(OffsetDateTime.class))).thenReturn(1L); + ArgumentCaptor captor = ArgumentCaptor.forClass(ServiceBusMessage.class); + + AsbPublisher.create(sender) + .publish(URN, Map.of(), null, Duration.ofSeconds(30), Map.of("traceparent", TRACEPARENT)); + + verify(sender).scheduleMessage(captor.capture(), any(OffsetDateTime.class)); + assertEquals(TRACEPARENT, captor.getValue().getApplicationProperties().get("traceparent")); + assertEquals(30000L, captor.getValue().getApplicationProperties().get("bq-delay")); + } + + @Test + void publishWithHeadersMatchesTheOtelHeaderSenderContract() { + ServiceBusSenderClient sender = mock(ServiceBusSenderClient.class); + when(sender.getEntityPath()).thenReturn("orders"); + ArgumentCaptor captor = ArgumentCaptor.forClass(ServiceBusMessage.class); + AsbPublisher publisher = AsbPublisher.create(sender); + com.babelqueue.otel.HeaderSender headerSender = publisher::publishWithHeaders; + + assertDoesNotThrow(() -> headerSender.send(EnvelopeCodec.make(URN, Map.of(), "orders", "t"), + Map.of("traceparent", TRACEPARENT))); + + verify(sender).sendMessage(captor.capture()); + assertEquals(TRACEPARENT, captor.getValue().getApplicationProperties().get("traceparent")); + } }