From cea99c108080a073b50562940be0e52f37e1e2af Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Muhammet=20=C5=9Eafak?= Date: Wed, 11 Nov 2026 09:28:38 +0300 Subject: [PATCH 1/2] feat(artemis): add PublishEnvelopeAsync so a Telemetry-built envelope is sent as is Telemetry.PublishAsync records the built envelope's meta.id on the producer span, but the send callback had to call PublishWithHeadersAsync(urn, data), which rebuilt a different envelope, so messaging.message.id never matched the wire meta.id. PublishEnvelopeAsync sends the given envelope unchanged; PublishWithHeadersAsync now delegates to it. --- CHANGELOG.md | 4 + README.md | 19 ++++ src/BabelQueue.Artemis/ArtemisPublisher.cs | 26 +++++ .../ArtemisPublisherTests.cs | 105 ++++++++++++++++++ 4 files changed, 154 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7a6fe42..ea3d6f9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ The envelope wire format is versioned separately by `meta.schema_version` ## [Unreleased] +### Added + +- **`ArtemisPublisher.PublishEnvelopeAsync(envelope, headers, ...)`.** Sends an already-built `Envelope` exactly as given, with `delay` and the §7 projection and the optional out-of-band `headers`, and returns `meta.id`. Use it as the `send` callback of `Telemetry.PublishAsync(..., headers, env => publisher.PublishEnvelopeAsync(env, headers), queue)`: the producer span's `messaging.message.id` was the `meta.id` of the envelope `Telemetry` built, while `PublishWithHeadersAsync(urn, data, ...)` rebuilt a different one, so the span id never matched the wire. Now it does. `PublishWithHeadersAsync` and `PublishAsync` delegate to it; their behaviour is unchanged. Envelope unchanged (`schema_version` stays `1`); no new `BabelQueue.Core` requirement. + ## [1.3.0] - 2026-11-03 ### Added diff --git a/README.md b/README.md index 0a9c530..396c6ce 100644 --- a/README.md +++ b/README.md @@ -120,6 +120,25 @@ The `bq-` application-property values are strings (integers as decimal, e.g. `"1 is `"babelqueue"`. The envelope is unchanged (`schema_version` stays `1`); Artemis is purely additive. +## Tracing a publish (ADR-0028, ADR-0038) + +`PublishEnvelopeAsync(envelope, headers, ...)` sends an **already-built** envelope exactly as given (the same +§7 projection, header overlay and scheduled delivery as `PublishWithHeadersAsync`, no rebuild). Use it as the `send` callback of +`BabelQueue.Core`'s `Telemetry.PublishAsync`, so the producer span's `messaging.message.id` is the `meta.id` that +reaches the wire and the span's `traceparent` rides beside the envelope: + +```csharp +using BabelQueue.Tracing; + +var headers = new Dictionary(); +await Telemetry.PublishAsync("urn:babel:orders:created", data, headers, + env => publisher.PublishEnvelopeAsync(env, headers), "orders", Telemetry.Systems.ActiveMq); +``` + +The `queue` argument you give `Telemetry.PublishAsync` becomes `meta.queue`; `PublishEnvelopeAsync` does not change +the envelope. The envelope is unchanged (`schema_version` stays `1`). Needs no `BabelQueue.Core` newer than the +one this package already requires. + ## Build & test ```bash diff --git a/src/BabelQueue.Artemis/ArtemisPublisher.cs b/src/BabelQueue.Artemis/ArtemisPublisher.cs index afafed4..33daa87 100644 --- a/src/BabelQueue.Artemis/ArtemisPublisher.cs +++ b/src/BabelQueue.Artemis/ArtemisPublisher.cs @@ -58,6 +58,32 @@ public async Task PublishWithHeadersAsync( CancellationToken cancellationToken = default) { var envelope = EnvelopeCodec.Make(urn, data, _sender.Address, traceId); + return await PublishEnvelopeAsync(envelope, headers, delay, cancellationToken).ConfigureAwait(false); + } + + /// + /// Sends an already-built exactly as given (the same §7 projection, header + /// overlay and scheduled delivery as ) without rebuilding it: + /// meta.id, trace_id and every extension field reach the wire unchanged. Use it as the send + /// callback of Telemetry.PublishAsync(..., headers, env => publisher.PublishEnvelopeAsync(env, headers), queue), + /// so the producer span's messaging.message.id is the meta.id on the wire. Returns meta.id. + /// + /// The envelope to send; its meta.queue is the caller's responsibility. + /// + /// meta.queue is whatever the envelope carries; it is not rewritten to this publisher's target. An envelope + /// built by Telemetry.PublishAsync carries that call's queue argument (default "default"), so pass + /// the queue/topic name there, or the wire meta.queue and the span's destination will not name it. + /// + /// The out-of-band transport headers to ride beside the envelope. + /// A positive delay schedules native delayed delivery. + /// Cancellation token. + public async Task PublishEnvelopeAsync( + Envelope envelope, + IReadOnlyDictionary? headers = null, + TimeSpan? delay = null, + CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(envelope); var message = AmqpProperties.ToMessage(envelope, delay, headers); await _sender.SendAsync(message, cancellationToken).ConfigureAwait(false); return envelope.Meta?.Id ?? string.Empty; diff --git a/tests/BabelQueue.Artemis.Tests/ArtemisPublisherTests.cs b/tests/BabelQueue.Artemis.Tests/ArtemisPublisherTests.cs index cd69eb6..507c79a 100644 --- a/tests/BabelQueue.Artemis.Tests/ArtemisPublisherTests.cs +++ b/tests/BabelQueue.Artemis.Tests/ArtemisPublisherTests.cs @@ -260,4 +260,109 @@ public void ExtractLeavesOutTheContractProperties() Assert.Equal(Traceparent, Assert.Single(map).Value); } + + // O6: the envelope Telemetry.PublishAsync builds is the one that is sent, so the producer span's + // messaging.message.id is the meta.id on the wire (and the traceparent rides beside the envelope). + [Fact] + public async Task TelemetryPublishEnvelopeRecordsTheWireMessageId() + { + var (sender, sent) = SenderOn("orders"); + var publisher = new ArtemisPublisher(sender.Object); + const string envelopeUrn = "urn:babel:orders:envelope-id"; + var activities = new List(); + using var listener = new ActivityListener + { + ShouldListenTo = source => source.Name == Telemetry.ActivitySourceName, + Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllData, + ActivityStopped = activity => + { + if (activity.Kind == ActivityKind.Producer && Equals(activity.GetTagItem("babelqueue.urn"), envelopeUrn)) + { + lock (activities) + { + activities.Add(activity); + } + } + }, + }; + ActivitySource.AddActivityListener(listener); + + var headers = new Dictionary(); + var id = await Telemetry.PublishAsync( + envelopeUrn, + new Dictionary { ["order_id"] = 7 }, + headers, + env => publisher.PublishEnvelopeAsync(env, headers), + "orders"); + + var producerSpan = Assert.Single(activities); + var message = sent()!; + var body = AmqpProperties.Text(message); + var traceparent = message.ApplicationProperties?.Map.ContainsKey("traceparent") == true + ? (string)message.ApplicationProperties.Map["traceparent"] + : null; + var wire = EnvelopeCodec.Decode(body); + + Assert.Equal(id, wire.Meta!.Id); + Assert.Equal(wire.Meta.Id, producerSpan.GetTagItem("messaging.message.id")); + Assert.Equal(Telemetry.UuidOf(producerSpan.TraceId), wire.TraceId); + Assert.Equal(wire.TraceId, producerSpan.GetTagItem("messaging.message.conversation_id")); + Assert.Equal(EnvelopeCodec.SchemaVersion, wire.Meta.SchemaVersion); + Assert.Equal(headers["traceparent"], traceparent); + Assert.Contains(producerSpan.TraceId.ToHexString(), traceparent, StringComparison.Ordinal); + Assert.DoesNotContain("traceparent", body, StringComparison.Ordinal); + } + + [Fact] + public async Task PublishEnvelopeSendsTheGivenEnvelopeUnchanged() + { + var (sender, sent) = SenderOn("orders"); + var publisher = new ArtemisPublisher(sender.Object); + var made = EnvelopeCodec.Make( + "urn:babel:orders:created", new Dictionary { ["order_id"] = 7 }, "orders", "trace-fixed"); + var envelope = made with + { + Extras = new Dictionary { ["x_tenant"] = JsonSerializer.SerializeToElement("t1") }, + Meta = made.Meta! with + { + Extras = new Dictionary { ["x_region"] = JsonSerializer.SerializeToElement("eu") }, + }, + }; + + var id = await publisher.PublishEnvelopeAsync(envelope); + + var message = sent()!; + var body = AmqpProperties.Text(message); + var traceparent = message.ApplicationProperties?.Map.ContainsKey("traceparent") == true + ? (string)message.ApplicationProperties.Map["traceparent"] + : null; + Assert.Equal(envelope.Meta!.Id, id); + Assert.Equal(EnvelopeCodec.Encode(envelope), body); + var wire = EnvelopeCodec.Decode(body); + Assert.Equal("trace-fixed", wire.TraceId); + Assert.Equal(EnvelopeCodec.SchemaVersion, wire.Meta!.SchemaVersion); + Assert.Equal(envelope.Meta.Id, wire.Meta.Id); + Assert.Equal("t1", wire.Extras!["x_tenant"].GetString()); + Assert.Equal("eu", wire.Meta.Extras!["x_region"].GetString()); + Assert.Null(traceparent); + } + + // Documents the PublishEnvelopeAsync contract: meta.queue is whatever the envelope carries. Telemetry stamps + // its queue argument (default "default"), not this publisher's target, so callers must pass the queue. + [Fact] + public async Task TelemetryPublishWithoutAQueueArgumentStampsDefaultMetaQueue() + { + var (sender, sent) = SenderOn("orders"); + var publisher = new ArtemisPublisher(sender.Object); + await Telemetry.PublishAsync( + "urn:babel:orders:created", null, env => publisher.PublishEnvelopeAsync(env)); + + var message = sent()!; + var body = AmqpProperties.Text(message); + var traceparent = message.ApplicationProperties?.Map.ContainsKey("traceparent") == true + ? (string)message.ApplicationProperties.Map["traceparent"] + : null; + Assert.Equal("default", EnvelopeCodec.Decode(body).Meta!.Queue); + Assert.Null(traceparent); + } } From a328a0775126d4c660272195a4499fb2a88f7a23 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Muhammet=20=C5=9Eafak?= Date: Wed, 11 Nov 2026 14:16:19 +0300 Subject: [PATCH 2/2] Release v1.4.0 --- CHANGELOG.md | 2 ++ src/BabelQueue.Artemis/BabelQueue.Artemis.csproj | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ea3d6f9..0a1b2a1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ The envelope wire format is versioned separately by `meta.schema_version` ## [Unreleased] +## [1.4.0] - 2026-11-11 + ### Added - **`ArtemisPublisher.PublishEnvelopeAsync(envelope, headers, ...)`.** Sends an already-built `Envelope` exactly as given, with `delay` and the §7 projection and the optional out-of-band `headers`, and returns `meta.id`. Use it as the `send` callback of `Telemetry.PublishAsync(..., headers, env => publisher.PublishEnvelopeAsync(env, headers), queue)`: the producer span's `messaging.message.id` was the `meta.id` of the envelope `Telemetry` built, while `PublishWithHeadersAsync(urn, data, ...)` rebuilt a different one, so the span id never matched the wire. Now it does. `PublishWithHeadersAsync` and `PublishAsync` delegate to it; their behaviour is unchanged. Envelope unchanged (`schema_version` stays `1`); no new `BabelQueue.Core` requirement. diff --git a/src/BabelQueue.Artemis/BabelQueue.Artemis.csproj b/src/BabelQueue.Artemis/BabelQueue.Artemis.csproj index bac61d4..b3d9fed 100644 --- a/src/BabelQueue.Artemis/BabelQueue.Artemis.csproj +++ b/src/BabelQueue.Artemis/BabelQueue.Artemis.csproj @@ -13,7 +13,7 @@ latest-recommended BabelQueue.Artemis - 1.3.0 + 1.4.0 Muhammet Şafak BabelQueue Apache ActiveMQ Artemis transport for BabelQueue — a canonical-envelope publisher and a URN-routed AMQP 1.0 consumer (accept/release settlement, native scheduled delivery, cross-language DLQ) over AMQPNetLite, on the framework-agnostic core.