Fix RabbitMQ queue binding and Gateway GetOriginalStream

src/Gateway/src/Eventuous.Gateway/GatewayMetaHelper.cs

@@ -30,8 +30,8 @@ public static Metadata GetContextMeta(IMessageConsumeContext context) {
 [PublicAPI]
 public static class ProducedMessageExtensions {
     extension(ProducedMessage message) {
-        public Stream? GetOriginalStream()
-            => message.AdditionalHeaders?.Get<Stream>(GatewayContextItems.OriginalStream);
+        public StreamName? GetOriginalStream()
+            => message.AdditionalHeaders?.Get<StreamName>(GatewayContextItems.OriginalStream);
 
         public object? GetOriginalMessage()
             => message.AdditionalHeaders?.Get<object>(GatewayContextItems.OriginalMessage);

src/Gateway/test/Eventuous.Tests.Gateway/GatewayMetaTests.cs

@@ -0,0 +1,64 @@
+using Eventuous.Gateway;
+using Eventuous.Producers;
+
+namespace Eventuous.Tests.Gateway;
+
+public class GatewayMetaTests {
+    [Test]
+    public async Task GetOriginalStream_ReturnsStreamName() {
+        var streamName = new StreamName("Test-123");
+
+        var headers = new Metadata(new Dictionary<string, object?> {
+            [GatewayContextItems.OriginalStream] = streamName
+        });
+
+        var message = new ProducedMessage("test", null, headers);
+
+        var result = message.GetOriginalStream();
+
+        await Assert.That(result).IsNotNull();
+        await Assert.That(result!.Value).IsEqualTo(streamName);
+    }
+
+    [Test]
+    public async Task GetOriginalMessageId_ReturnsMessageId() {
+        var messageId = Guid.NewGuid().ToString();
+
+        var headers = new Metadata(new Dictionary<string, object?> {
+            [GatewayContextItems.OriginalMessageId] = messageId
+        });
+
+        var message = new ProducedMessage("test", null, headers);
+
+        await Assert.That(message.GetOriginalMessageId()).IsEqualTo(messageId);
+    }
+
+    [Test]
+    public async Task GetOriginalMessageType_ReturnsMessageType() {
+        var headers = new Metadata(new Dictionary<string, object?> {
+            [GatewayContextItems.OriginalMessageType] = "test-event"
+        });
+
+        var message = new ProducedMessage("test", null, headers);
+
+        await Assert.That(message.GetOriginalMessageType()).IsEqualTo("test-event");
+    }
+
+    [Test]
+    public async Task GetOriginalStreamPosition_ReturnsPosition() {
+        var headers = new Metadata(new Dictionary<string, object?> {
+            [GatewayContextItems.OriginalStreamPosition] = 42UL
+        });
+
+        var message = new ProducedMessage("test", null, headers);
+
+        await Assert.That(message.GetOriginalStreamPosition()).IsEqualTo(42UL);
+    }
+
+    [Test]
+    public async Task GetOriginalStream_WithNullHeaders_ReturnsNull() {
+        var message = new ProducedMessage("test", null);
+
+        await Assert.That(message.GetOriginalStream()).IsNull();
+    }
+}

src/RabbitMq/src/Eventuous.RabbitMq/Subscriptions/RabbitMqSubscription.cs

@@ -123,10 +123,10 @@ protected override ValueTask Subscribe(CancellationToken cancellationToken) {
             Options.QueueOptions.Arguments
         );
 
-        Log.InfoLog?.Log("Binding exchange {Exchange} to queue {Queue}", exchange, Options.SubscriptionId);
+        Log.InfoLog?.Log("Binding exchange {Exchange} to queue {Queue}", exchange, queue);
 
         _channel.QueueBind(
-            Options.SubscriptionId,
+            queue,
             exchange,
             Options.BindingOptions.RoutingKey,
             Options.BindingOptions.Arguments

src/RabbitMq/test/Eventuous.Tests.RabbitMq/CustomQueueSubscriptionSpec.cs

@@ -0,0 +1,72 @@
+using Eventuous.Producers;
+using Eventuous.RabbitMq.Producers;
+using Eventuous.RabbitMq.Subscriptions;
+using Eventuous.Subscriptions.Filters;
+using Eventuous.TestHelpers.TUnit;
+using Eventuous.TestHelpers.TUnit.Logging;
+using Eventuous.Tests.Subscriptions.Base;
+
+namespace Eventuous.Tests.RabbitMq;
+
+[ClassDataSource<RabbitMqFixture>]
+public class CustomQueueSubscriptionSpec {
+    static CustomQueueSubscriptionSpec() => TypeMap.Instance.RegisterKnownEventTypes(typeof(TestEvent).Assembly);
+
+    RabbitMqProducer                              _producer     = null!;
+    TestEventHandler                              _handler      = null!;
+#pragma warning disable TUnit0023
+    RabbitMqSubscription                          _subscription = null!;
+    TestEventListener                             _es           = null!;
+#pragma warning restore TUnit0023
+    readonly StreamName                           _exchange;
+    readonly ILogger<CustomQueueSubscriptionSpec> _log;
+    readonly ILoggerFactory                       _loggerFactory;
+    readonly RabbitMqFixture                      _fixture;
+
+    public CustomQueueSubscriptionSpec(RabbitMqFixture fixture) {
+        _fixture       = fixture;
+        _exchange      = new(Guid.NewGuid().ToString());
+        _loggerFactory = LoggingExtensions.GetLoggerFactory();
+        _log           = _loggerFactory.CreateLogger<CustomQueueSubscriptionSpec>();
+    }
+
+    [Test]
+    public async Task SubscribeWithCustomQueueName(CancellationToken cancellationToken) {
+        var testEvent = TestEvent.Create();
+        await _producer.Produce(_exchange, testEvent, new(), cancellationToken: cancellationToken);
+        await _handler.AssertThat().Timebox(10.Seconds()).Any().Match(x => x as TestEvent == testEvent).Validate(cancellationToken);
+    }
+
+    [Before(Test)]
+    public async ValueTask InitializeAsync() {
+        _es       = new();
+        _handler  = new();
+        _producer = new(_fixture.ConnectionFactory);
+
+        var subscriptionId = Guid.NewGuid().ToString();
+        var customQueue    = Guid.NewGuid().ToString();
+
+        _subscription = new(
+            _fixture.ConnectionFactory,
+            new RabbitMqSubscriptionOptions {
+                ConcurrencyLimit = 10,
+                SubscriptionId   = subscriptionId,
+                Exchange         = _exchange,
+                ThrowOnError     = true,
+                QueueOptions     = new RabbitMqSubscriptionOptions.RabbitMqQueueOptions { Queue = customQueue }
+            },
+            new ConsumePipe().AddDefaultConsumer(_handler),
+            _loggerFactory
+        );
+        await _subscription.SubscribeWithLog(_log);
+        await _producer.StartAsync();
+    }
+
+    [After(Test)]
+    public async ValueTask DisposeAsync() {
+        await _producer.StopAsync();
+        await _subscription.UnsubscribeWithLog(_log);
+        _es.Dispose();
+        await _subscription.DisposeAsync();
+    }
+}