Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,12 @@ 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.

## [1.3.0] - 2026-11-03

### Added
Expand Down
19 changes: 19 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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<string, string>();
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
Expand Down
26 changes: 26 additions & 0 deletions src/BabelQueue.Artemis/ArtemisPublisher.cs
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,32 @@ public async Task<string> PublishWithHeadersAsync(
CancellationToken cancellationToken = default)
{
var envelope = EnvelopeCodec.Make(urn, data, _sender.Address, traceId);
return await PublishEnvelopeAsync(envelope, headers, delay, cancellationToken).ConfigureAwait(false);
}

/// <summary>
/// Sends an <b>already-built</b> <paramref name="envelope"/> exactly as given (the same §7 projection, header
/// overlay and scheduled delivery as <see cref="PublishWithHeadersAsync"/>) without rebuilding it:
/// <c>meta.id</c>, <c>trace_id</c> and every extension field reach the wire unchanged. Use it as the <c>send</c>
/// callback of <c>Telemetry.PublishAsync(..., headers, env =&gt; publisher.PublishEnvelopeAsync(env, headers), queue)</c>,
/// so the producer span's <c>messaging.message.id</c> is the <c>meta.id</c> on the wire. Returns <c>meta.id</c>.
/// </summary>
/// <param name="envelope">The envelope to send; its <c>meta.queue</c> is the caller's responsibility.</param>
/// <remarks>
/// <c>meta.queue</c> is whatever the envelope carries; it is not rewritten to this publisher's target. An envelope
/// built by <c>Telemetry.PublishAsync</c> carries that call's <c>queue</c> argument (default <c>"default"</c>), so pass
/// the queue/topic name there, or the wire <c>meta.queue</c> and the span's destination will not name it.
/// </remarks>
/// <param name="headers">The out-of-band transport headers to ride beside the envelope.</param>
/// <param name="delay">A positive delay schedules native delayed delivery.</param>
/// <param name="cancellationToken">Cancellation token.</param>
public async Task<string> PublishEnvelopeAsync(
Envelope envelope,
IReadOnlyDictionary<string, string>? 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;
Expand Down
2 changes: 1 addition & 1 deletion src/BabelQueue.Artemis/BabelQueue.Artemis.csproj
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
<AnalysisLevel>latest-recommended</AnalysisLevel>

<PackageId>BabelQueue.Artemis</PackageId>
<Version>1.3.0</Version>
<Version>1.4.0</Version>
<Authors>Muhammet Şafak</Authors>
<Company>BabelQueue</Company>
<Description>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.</Description>
Expand Down
105 changes: 105 additions & 0 deletions tests/BabelQueue.Artemis.Tests/ArtemisPublisherTests.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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<Activity>();
using var listener = new ActivityListener
{
ShouldListenTo = source => source.Name == Telemetry.ActivitySourceName,
Sample = (ref ActivityCreationOptions<ActivityContext> _) => 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<string, string>();
var id = await Telemetry.PublishAsync(
envelopeUrn,
new Dictionary<string, object?> { ["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<string, object?> { ["order_id"] = 7 }, "orders", "trace-fixed");
var envelope = made with
{
Extras = new Dictionary<string, JsonElement> { ["x_tenant"] = JsonSerializer.SerializeToElement("t1") },
Meta = made.Meta! with
{
Extras = new Dictionary<string, JsonElement> { ["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);
}
}
Loading