v7: UseActiveMq with options from configuration, and headers over ActiveMQ

.github/workflows/dotnet.yml

@@ -49,6 +49,9 @@ jobs:
         options: >-
           --health-cmd "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092"
           --health-interval 5s --health-timeout 10s --health-retries 20
+      activemq:
+        image: apache/activemq-classic:6.2.0
+        ports: ['61616:61616']
 
     steps:
     - name: Checkout
@@ -82,11 +85,13 @@ jobs:
         RABBITMQ_URI: amqp://localhost:5672
         NATS_URL: nats://localhost:4222
         KAFKA_BOOTSTRAP_SERVERS: localhost:9092
+        ACTIVEMQ_URI: activemq:tcp://localhost:61616
       run: |
         dotnet test --project Obvs.RabbitMQ.Tests --no-build --configuration Release -- --filter-trait "Category=Explicit"
         dotnet test --project Obvs.NATS.Tests --no-build --configuration Release -- --filter-trait "Category=Explicit"
         dotnet test --project Obvs.Kafka.Tests --no-build --configuration Release -- --filter-trait "Category=Explicit"
         dotnet test --project Obvs.NetMQ.Tests --no-build --configuration Release -- --filter-trait "Category=Explicit"
+        dotnet test --project Obvs.ActiveMQ.Tests --no-build --configuration Release -- --filter-trait "Category=Explicit"
     - name: Test
       # xUnit v3 runs on Microsoft.Testing.Platform (global.json). Exit code 8 means a project ran no tests,
       # as the transport test projects only contain Explicit tests.

Obvs.ActiveMQ.Tests/Hosting/HostingMessages.cs

@@ -0,0 +1,30 @@
+using Obvs.Types;
+
+namespace Obvs.ActiveMQ.Tests.Hosting
+{
+    public interface IOrderMessage : IMessage { }
+
+    public class PlaceOrder : IOrderMessage, ICommand
+    {
+        public string OrderId { get; set; }
+    }
+
+    public class OrderPlaced : IOrderMessage, IEvent
+    {
+        public string OrderId { get; set; }
+    }
+
+    public class GetQuote : IOrderMessage, IRequest
+    {
+        public string Sku { get; set; }
+        public string RequestId { get; set; }
+        public string RequesterId { get; set; }
+    }
+
+    public class Quote : IOrderMessage, IResponse
+    {
+        public decimal Price { get; set; }
+        public string RequestId { get; set; }
+        public string RequesterId { get; set; }
+    }
+}

Obvs.ActiveMQ.Tests/Hosting/TestActiveMqHosting.cs

@@ -0,0 +1,188 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
+using System.Reactive.Linq;
+using System.Reactive.Threading.Tasks;
+using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Obvs.Hosting;
+using Obvs.Serialization.Json;
+using Xunit;
+
+namespace Obvs.ActiveMQ.Tests.Hosting
+{
+    // End to end through AddObvs and UseActiveMq against a real broker (ACTIVEMQ_URI, default activemq:tcp://localhost:61616).
+    // Each test uses its own service name, so its topics don't meet the other tests'.
+    public class TestActiveMqHosting
+    {
+        private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(10);
+
+        [Fact, Trait("Category", "Explicit")]
+        public async Task ShouldSendCommandsAndPublishEventsOverActiveMq()
+        {
+            using var host = await StartHostAsync("obvs-activemq-commands-events");
+            var bus = host.Services.GetRequiredService<IServiceBus>();
+
+            var command = bus.Commands.OfType<PlaceOrder>().FirstAsync().ToTask();
+            await bus.SendAsync(new PlaceOrder { OrderId = "o-1" });
+            Assert.Equal("o-1", (await command.WaitAsync(Timeout, TestContext.Current.CancellationToken)).OrderId);
+
+            var placed = bus.Events.OfType<OrderPlaced>().FirstAsync().ToTask();
+            await bus.PublishAsync(new OrderPlaced { OrderId = "o-1" });
+            Assert.Equal("o-1", (await placed.WaitAsync(Timeout, TestContext.Current.CancellationToken)).OrderId);
+        }
+
+        [Fact, Trait("Category", "Explicit")]
+        public async Task ShouldReturnResponsesOverActiveMq()
+        {
+            using var host = await StartHostAsync("obvs-activemq-requests", service => service
+                .HandleRequest<GetQuote, Quote>((request, context, ct) => Task.FromResult(new Quote { Price = 9.99m })));
+
+            var quote = await host.Services.GetRequiredService<IServiceBusClient>()
+                .RequestAsync<GetQuote, Quote>(new GetQuote { Sku = "sku-1" }, Timeout, TestContext.Current.CancellationToken);
+
+            Assert.Equal(9.99m, quote.Price);
+        }
+
+        [Fact, Trait("Category", "Explicit")]
+        public async Task ShouldDeliverToEveryBusInTheProcess()
+        {
+            const string serviceName = "obvs-activemq-every-bus";
+            using var hostA = await StartHostAsync(serviceName);
+            using var hostB = await StartHostAsync(serviceName);
+            var busA = hostA.Services.GetRequiredService<IServiceBus>();
+            var busB = hostB.Services.GetRequiredService<IServiceBus>();
+
+            var receivedByA = busA.Commands.OfType<PlaceOrder>().FirstAsync().ToTask();
+            var receivedByB = busB.Commands.OfType<PlaceOrder>().FirstAsync().ToTask();
+            await busA.SendAsync(new PlaceOrder { OrderId = "o-1" });
+
+            Assert.Equal("o-1", (await receivedByA.WaitAsync(Timeout, TestContext.Current.CancellationToken)).OrderId);
+            Assert.Equal("o-1", (await receivedByB.WaitAsync(Timeout, TestContext.Current.CancellationToken)).OrderId);
+        }
+
+        [Fact, Trait("Category", "Explicit")]
+        public async Task ShouldKeepWorkingWhenAnotherHostInTheProcessStops()
+        {
+            // Two hosts in one process, as tests have: stopping one mustn't close the other's connection
+            using var hostA = await StartHostAsync("obvs-activemq-two-hosts");
+            var hostB = await StartHostAsync("obvs-activemq-two-hosts");
+            await hostB.StopAsync(TestContext.Current.CancellationToken);
+            hostB.Dispose();
+            var busA = hostA.Services.GetRequiredService<IServiceBus>();
+
+            var command = busA.Commands.OfType<PlaceOrder>().FirstAsync().ToTask();
+            await busA.SendAsync(new PlaceOrder { OrderId = "o-1" });
+
+            Assert.Equal("o-1", (await command.WaitAsync(Timeout, TestContext.Current.CancellationToken)).OrderId);
+        }
+
+        [Fact, Trait("Category", "Explicit")]
+        public async Task ShouldNotDeadlockWhenASubscriberDisposesTheBus()
+        {
+            var host = await StartHostAsync("obvs-activemq-dispose-in-callback");
+            var bus = host.Services.GetRequiredService<IServiceBus>();
+            var disposed = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+            using var subscription = bus.Commands.OfType<PlaceOrder>().Subscribe(_ =>
+            {
+                host.Dispose();
+                disposed.SetResult(true);
+            });
+
+            await bus.SendAsync(new PlaceOrder { OrderId = "o-1" });
+
+            Assert.True(await disposed.Task.WaitAsync(Timeout, TestContext.Current.CancellationToken));
+        }
+
+        [Fact, Trait("Category", "Explicit")]
+        public async Task ShouldContinueTheSendersTraceAcrossActiveMq()
+        {
+            const string serviceName = "obvs-activemq-tracing";
+            using var activities = new ActivityRecorder();
+            using var receiver = await StartHostAsync(serviceName, service => service.AsServer()
+                .HandleCommand<PlaceOrder>((command, context, ct) => Task.CompletedTask));
+            using var sender = await StartHostAsync(serviceName, service => service.AsClient());
+
+            await sender.Services.GetRequiredService<IServiceBusClient>().SendAsync(new PlaceOrder { OrderId = "o-1" });
+
+            var send = await WaitForAsync(() => activities.For(serviceName).SingleOrDefault(activity => activity.DisplayName == $"send {serviceName}"));
+            var process = await WaitForAsync(() => activities.For(serviceName).SingleOrDefault(activity => activity.DisplayName == $"process {serviceName}"));
+            Assert.Equal(send.TraceId, process.TraceId);
+            Assert.Equal(send.SpanId, process.ParentSpanId);
+        }
+
+        [Fact, Trait("Category", "Explicit")]
+        public async Task ShouldCarryHeadersTheAppSetsOverActiveMq()
+        {
+            const string serviceName = "obvs-activemq-headers";
+            var received = new TaskCompletionSource<IReadOnlyDictionary<string, string>>(TaskCreationOptions.RunContinuationsAsynchronously);
+            using var receiver = await StartHostAsync(serviceName, service => service.AsServer()
+                .HandleCommand<PlaceOrder>((command, context, ct) =>
+                {
+                    received.TrySetResult(ObvsMessageHeaders.Get(command));
+                    return Task.CompletedTask;
+                }));
+            using var sender = await StartHostAsync(serviceName, service => service.AsClient());
+            var placeOrder = new PlaceOrder { OrderId = "o-1" };
+            ObvsMessageHeaders.Set(placeOrder, new Dictionary<string, string> { ["x-tenant"] = "acme" });
+
+            await sender.Services.GetRequiredService<IServiceBusClient>().SendAsync(placeOrder);
+
+            var headers = await received.Task.WaitAsync(Timeout, TestContext.Current.CancellationToken);
+            Assert.Equal("acme", headers["x-tenant"]);
+        }
+
+        private static async Task<IHost> StartHostAsync(string serviceName, Action<ObvsServiceBuilder> configure = null)
+        {
+            var builder = Host.CreateApplicationBuilder(new HostApplicationBuilderSettings { DisableDefaults = true });
+            builder.Services.AddObvs(obvs => obvs.AddService<IOrderMessage>(serviceName, service =>
+            {
+                service
+                    .UseActiveMq(options => options.BrokerUri = TestBroker.Uri)
+                    .SerializedWith(new JsonMessageSerializer(), new JsonMessageDeserializerFactory(typeof(JsonMessageDeserializer<>)));
+                configure?.Invoke(service);
+            }));
+            var host = builder.Build();
+            await host.StartAsync(TestContext.Current.CancellationToken);
+            return host;
+        }
+
+        private static async Task<T> WaitForAsync<T>(Func<T> find) where T : class
+        {
+            var deadline = DateTime.UtcNow + Timeout;
+            T found;
+            while ((found = find()) == null && DateTime.UtcNow < deadline)
+            {
+                await Task.Delay(20, TestContext.Current.CancellationToken);
+            }
+            return found ?? throw new TimeoutException($"The expected {typeof(T).Name} wasn't recorded in time.");
+        }
+
+        private sealed class ActivityRecorder : IDisposable
+        {
+            private readonly ConcurrentQueue<Activity> _stopped = new ConcurrentQueue<Activity>();
+            private readonly ActivityListener _listener;
+
+            public ActivityRecorder()
+            {
+                _listener = new ActivityListener
+                {
+                    ShouldListenTo = source => source.Name == ObvsTelemetry.ActivitySourceName,
+                    Sample = (ref ActivityCreationOptions<ActivityContext> _) => ActivitySamplingResult.AllDataAndRecorded,
+                    ActivityStopped = activity => _stopped.Enqueue(activity),
+                };
+                ActivitySource.AddActivityListener(_listener);
+            }
+
+            public IReadOnlyList<Activity> For(string serviceName)
+            {
+                return _stopped.Where(activity => (activity.GetTagItem("messaging.destination.name") as string) == serviceName).ToList();
+            }
+
+            public void Dispose() => _listener.Dispose();
+        }
+    }
+}

Obvs.ActiveMQ.Tests/Hosting/TestUseActiveMq.cs

@@ -0,0 +1,104 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Options;
+using Obvs.Hosting;
+using Obvs.Serialization.Json;
+using Xunit;
+
+namespace Obvs.ActiveMQ.Tests.Hosting
+{
+    // Configuration and validation; none of these connect to a broker
+    public class TestUseActiveMq
+    {
+        [Fact]
+        public void ShouldBindOptionsFromConfiguration()
+        {
+            var configuration = Configuration(new Dictionary<string, string>
+            {
+                ["Obvs:Orders:ActiveMq:BrokerUri"] = "activemq:tcp://activemq.example:61616",
+                ["Obvs:Orders:ActiveMq:UserName"] = "orders-app",
+                ["Obvs:Orders:ActiveMq:Password"] = "not-a-real-password",
+            });
+            using var host = BuildHost(services => services.AddObvs(obvs => obvs
+                .AddService<IOrderMessage>("orders", service => service
+                    .UseActiveMq(configuration.GetSection("Obvs:Orders:ActiveMq"))
+                    .SerializedWith(new JsonMessageSerializer(), new JsonMessageDeserializerFactory(typeof(JsonMessageDeserializer<>))))));
+
+            var options = host.Services.GetRequiredService<IOptionsMonitor<ActiveMqOptions>>().Get("orders");
+            Assert.Equal("activemq:tcp://activemq.example:61616", options.BrokerUri);
+            Assert.Equal("orders-app", options.UserName);
+            Assert.Equal("not-a-real-password", options.Password);
+        }
+
+        [Fact]
+        public void ShouldConfigureOptionsInCode()
+        {
+            using var host = BuildHost(services => services.AddObvs(obvs => obvs
+                .AddService<IOrderMessage>("orders", service => service
+                    .UseActiveMq(options => options.BrokerUri = "activemq:tcp://activemq.example:61616")
+                    .SerializedWith(new JsonMessageSerializer(), new JsonMessageDeserializerFactory(typeof(JsonMessageDeserializer<>))))));
+
+            Assert.Equal("activemq:tcp://activemq.example:61616", host.Services.GetRequiredService<IOptionsMonitor<ActiveMqOptions>>().Get("orders").BrokerUri);
+        }
+
+        [Fact]
+        public async Task ShouldFailToStartWhenTheBrokerUriIsMissingFromConfiguration()
+        {
+            var configuration = Configuration(new Dictionary<string, string>());
+            using var host = BuildHost(services => services.AddObvs(obvs => obvs
+                .AddService<IOrderMessage>("orders", service => service
+                    .UseActiveMq(configuration.GetSection("Obvs:Orders:ActiveMq"))
+                    .SerializedWith(new JsonMessageSerializer(), new JsonMessageDeserializerFactory(typeof(JsonMessageDeserializer<>))))));
+
+            var exception = await Assert.ThrowsAsync<ObvsConfigurationException>(() => host.StartAsync(TestContext.Current.CancellationToken));
+
+            var error = Assert.Single(exception.Errors);
+            Assert.Contains("'orders'", error);
+            Assert.Contains("Obvs:Orders:ActiveMq:BrokerUri", error);
+        }
+
+        [Fact]
+        public async Task ShouldFailToStartWhenTheBrokerUriIsMissingFromCode()
+        {
+            using var host = BuildHost(services => services.AddObvs(obvs => obvs
+                .AddService<IOrderMessage>("orders", service => service
+                    .UseActiveMq(options => { })
+                    .SerializedWith(new JsonMessageSerializer(), new JsonMessageDeserializerFactory(typeof(JsonMessageDeserializer<>))))));
+
+            var exception = await Assert.ThrowsAsync<ObvsConfigurationException>(() => host.StartAsync(TestContext.Current.CancellationToken));
+
+            var error = Assert.Single(exception.Errors);
+            Assert.Contains("'orders'", error);
+            Assert.Contains("BrokerUri", error);
+        }
+
+        [Fact]
+        public async Task ShouldFailToStartWithoutASerializer()
+        {
+            using var host = BuildHost(services => services.AddObvs(obvs => obvs
+                .AddService<IOrderMessage>("orders", service => service.UseActiveMq(options => options.BrokerUri = "activemq:tcp://activemq.example:61616"))));
+
+            var exception = await Assert.ThrowsAsync<ObvsConfigurationException>(() => host.StartAsync(TestContext.Current.CancellationToken));
+
+            var error = Assert.Single(exception.Errors);
+            Assert.Contains("'orders'", error);
+            Assert.Contains("SerializedWith", error);
+        }
+
+        private static IConfiguration Configuration(IDictionary<string, string> values)
+        {
+            return new ConfigurationBuilder().AddInMemoryCollection(values).Build();
+        }
+
+        private static IHost BuildHost(Action<IServiceCollection> configure)
+        {
+            var builder = Host.CreateApplicationBuilder(new HostApplicationBuilderSettings { DisableDefaults = true });
+            configure(builder.Services);
+            return builder.Build();
+        }
+    }
+}

Obvs.ActiveMQ.Tests/Obvs.ActiveMQ.Tests.csproj

@@ -9,6 +9,7 @@
   </PropertyGroup>
   <ItemGroup>
     <PackageReference Include="FakeItEasy" />
+    <PackageReference Include="Microsoft.Extensions.Hosting" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" />
     <PackageReference Include="Microsoft.Reactive.Testing" />
     <PackageReference Include="xunit.v3" />
@@ -19,5 +20,6 @@
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\Obvs.ActiveMQ\Obvs.ActiveMQ.csproj" />
+    <ProjectReference Include="..\Obvs.Serialization.Json\Obvs.Serialization.Json.csproj" />
   </ItemGroup>
 </Project>

Obvs.ActiveMQ.Tests/TestBroker.cs

@@ -0,0 +1,10 @@
+using System;
+
+namespace Obvs.ActiveMQ.Tests
+{
+    internal static class TestBroker
+    {
+        // Set ACTIVEMQ_URI to run the Explicit tests against a broker other than a default local install
+        public static string Uri => Environment.GetEnvironmentVariable("ACTIVEMQ_URI") ?? "activemq:tcp://localhost:61616";
+    }
+}