v7: UseNats with options from configuration, and headers over NATS

Obvs.NATS.Tests/Hosting/HostingMessages.cs

@@ -0,0 +1,30 @@
+using Obvs.Types;
+
+namespace Obvs.NATS.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.NATS.Tests/Hosting/TestNatsHosting.cs

@@ -0,0 +1,171 @@
+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.NATS.Tests.Hosting
+{
+    // End to end through AddObvs and UseNats against a real server (NATS_URL, default nats://localhost:4222).
+    // Each test uses its own service name, so its subjects don't meet the other tests'.
+    public class TestNatsHosting
+    {
+        private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(10);
+
+        [Fact, Trait("Category", "Explicit")]
+        public async Task ShouldSendCommandsAndPublishEventsOverNats()
+        {
+            using var host = await StartHostAsync("obvs-nats-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 ShouldReturnResponsesOverNats()
+        {
+            using var host = await StartHostAsync("obvs-nats-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 ShouldKeepWorkingWhenAnotherHostInTheProcessStops()
+        {
+            // Two hosts in one process, as tests have: stopping one mustn't close the other's connection
+            using var hostA = await StartHostAsync("obvs-nats-two-hosts");
+            var hostB = await StartHostAsync("obvs-nats-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-nats-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 ShouldContinueTheSendersTraceAcrossNats()
+        {
+            const string serviceName = "obvs-nats-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 ShouldCarryHeadersTheAppSetsOverNats()
+        {
+            const string serviceName = "obvs-nats-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
+                    .UseNats(options => options.Url = TestBroker.Url)
+                    .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.NATS.Tests/Hosting/TestUseNats.cs

@@ -0,0 +1,96 @@
+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.NATS.Tests.Hosting
+{
+    // Configuration and validation; none of these connect to a server
+    public class TestUseNats
+    {
+        [Fact]
+        public void ShouldBindOptionsFromConfiguration()
+        {
+            var configuration = Configuration(new Dictionary<string, string> { ["Obvs:Orders:Nats:Url"] = "nats://nats.example:4222" });
+            using var host = BuildHost(services => services.AddObvs(obvs => obvs
+                .AddService<IOrderMessage>("orders", service => service
+                    .UseNats(configuration.GetSection("Obvs:Orders:Nats"))
+                    .SerializedWith(new JsonMessageSerializer(), new JsonMessageDeserializerFactory(typeof(JsonMessageDeserializer<>))))));
+
+            Assert.Equal("nats://nats.example:4222", host.Services.GetRequiredService<IOptionsMonitor<NatsOptions>>().Get("orders").Url);
+        }
+
+        [Fact]
+        public void ShouldConfigureOptionsInCode()
+        {
+            using var host = BuildHost(services => services.AddObvs(obvs => obvs
+                .AddService<IOrderMessage>("orders", service => service
+                    .UseNats(options => options.Url = "nats://nats.example:4222")
+                    .SerializedWith(new JsonMessageSerializer(), new JsonMessageDeserializerFactory(typeof(JsonMessageDeserializer<>))))));
+
+            Assert.Equal("nats://nats.example:4222", host.Services.GetRequiredService<IOptionsMonitor<NatsOptions>>().Get("orders").Url);
+        }
+
+        [Fact]
+        public async Task ShouldFailToStartWhenTheUrlIsMissingFromConfiguration()
+        {
+            var configuration = Configuration(new Dictionary<string, string>());
+            using var host = BuildHost(services => services.AddObvs(obvs => obvs
+                .AddService<IOrderMessage>("orders", service => service
+                    .UseNats(configuration.GetSection("Obvs:Orders:Nats"))
+                    .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:Nats:Url", error);
+        }
+
+        [Fact]
+        public async Task ShouldFailToStartWhenTheUrlIsMissingFromCode()
+        {
+            using var host = BuildHost(services => services.AddObvs(obvs => obvs
+                .AddService<IOrderMessage>("orders", service => service
+                    .UseNats(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("Url", error);
+        }
+
+        [Fact]
+        public async Task ShouldFailToStartWithoutASerializer()
+        {
+            using var host = BuildHost(services => services.AddObvs(obvs => obvs
+                .AddService<IOrderMessage>("orders", service => service.UseNats(options => options.Url = "nats://nats.example:4222"))));
+
+            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.NATS.Tests/Obvs.NATS.Tests.csproj

@@ -5,6 +5,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" />

Obvs.NATS/MessagePublisher.cs

@@ -40,10 +40,27 @@ public async Task PublishAsync(TMessage message)
             using (var stream = new MemoryStream())
             {
                 ProtoBuf.Serializer.Serialize(stream, wrapper);
-                await _connection.PublishAsync(subject, stream.ToArray()).ConfigureAwait(false);
+                await _connection.PublishAsync(subject, stream.ToArray(), GetHeaders(message)).ConfigureAwait(false);
             }
         }
 
+        // The message's Obvs headers (such as trace context) as NATS headers, which v6 receivers ignore
+        private static NatsHeaders GetHeaders(TMessage message)
+        {
+            var headers = ObvsMessageHeaders.Get(message);
+            if (headers.Count == 0)
+            {
+                return null;
+            }
+
+            var natsHeaders = new NatsHeaders();
+            foreach (var header in headers)
+            {
+                natsHeaders[header.Key] = header.Value;
+            }
+            return natsHeaders;
+        }
+
         public void Dispose()
         {
         }

Obvs.NATS/MessageSource.cs

@@ -68,7 +68,9 @@ private async Task ReceiveAsync(INatsSub<byte[]> subscription, IObserver<TMessag
                         var wrapper = Unwrap(msg.Data);
                         if (MatchesFilter(wrapper))
                         {
-                            observer.OnNext(Deserialize(msg.Subject, wrapper));
+                            var message = Deserialize(msg.Subject, wrapper);
+                            SetHeaders(message, msg.Headers);
+                            observer.OnNext(message);
                         }
                     }
                 }
@@ -82,6 +84,21 @@ private async Task ReceiveAsync(INatsSub<byte[]> subscription, IObserver<TMessag
             }
         }
 
+        private static void SetHeaders(TMessage message, NatsHeaders natsHeaders)
+        {
+            if (natsHeaders == null || natsHeaders.Count == 0)
+            {
+                return;
+            }
+
+            var headers = new Dictionary<string, string>();
+            foreach (var header in natsHeaders)
+            {
+                headers[header.Key] = header.Value.ToString();
+            }
+            ObvsMessageHeaders.Set(message, headers);
+        }
+
         private bool MatchesFilter(MessageWrapper wrapper)
         {
             return _filter == null || _filter(wrapper.Properties);

Obvs.NATS/NatsOptions.cs

@@ -0,0 +1,12 @@
+namespace Obvs.NATS
+{
+    /// <summary>
+    /// A service's NATS settings. Each service has its own, as named options keyed by the service's
+    /// name: <c>IOptionsMonitor&lt;NatsOptions&gt;.Get("orders")</c>.
+    /// </summary>
+    public sealed class NatsOptions
+    {
+        /// <summary>The server to connect to, for example <c>nats://localhost:4222</c>. Required.</summary>
+        public string Url { get; set; }
+    }
+}