diff --git a/.github/workflows/dotnet.yml b/.github/workflows/dotnet.yml index f83f630..be575d3 100644 --- a/.github/workflows/dotnet.yml +++ b/.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. diff --git a/Obvs.ActiveMQ.Tests/Hosting/HostingMessages.cs b/Obvs.ActiveMQ.Tests/Hosting/HostingMessages.cs new file mode 100644 index 0000000..a1d86d2 --- /dev/null +++ b/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; } + } +} diff --git a/Obvs.ActiveMQ.Tests/Hosting/TestActiveMqHosting.cs b/Obvs.ActiveMQ.Tests/Hosting/TestActiveMqHosting.cs new file mode 100644 index 0000000..502fd45 --- /dev/null +++ b/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(); + + var command = bus.Commands.OfType().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().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((request, context, ct) => Task.FromResult(new Quote { Price = 9.99m }))); + + var quote = await host.Services.GetRequiredService() + .RequestAsync(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(); + var busB = hostB.Services.GetRequiredService(); + + var receivedByA = busA.Commands.OfType().FirstAsync().ToTask(); + var receivedByB = busB.Commands.OfType().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(); + + var command = busA.Commands.OfType().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(); + var disposed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var subscription = bus.Commands.OfType().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((command, context, ct) => Task.CompletedTask)); + using var sender = await StartHostAsync(serviceName, service => service.AsClient()); + + await sender.Services.GetRequiredService().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>(TaskCreationOptions.RunContinuationsAsynchronously); + using var receiver = await StartHostAsync(serviceName, service => service.AsServer() + .HandleCommand((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 { ["x-tenant"] = "acme" }); + + await sender.Services.GetRequiredService().SendAsync(placeOrder); + + var headers = await received.Task.WaitAsync(Timeout, TestContext.Current.CancellationToken); + Assert.Equal("acme", headers["x-tenant"]); + } + + private static async Task StartHostAsync(string serviceName, Action configure = null) + { + var builder = Host.CreateApplicationBuilder(new HostApplicationBuilderSettings { DisableDefaults = true }); + builder.Services.AddObvs(obvs => obvs.AddService(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 WaitForAsync(Func 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 _stopped = new ConcurrentQueue(); + private readonly ActivityListener _listener; + + public ActivityRecorder() + { + _listener = new ActivityListener + { + ShouldListenTo = source => source.Name == ObvsTelemetry.ActivitySourceName, + Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllDataAndRecorded, + ActivityStopped = activity => _stopped.Enqueue(activity), + }; + ActivitySource.AddActivityListener(_listener); + } + + public IReadOnlyList For(string serviceName) + { + return _stopped.Where(activity => (activity.GetTagItem("messaging.destination.name") as string) == serviceName).ToList(); + } + + public void Dispose() => _listener.Dispose(); + } + } +} diff --git a/Obvs.ActiveMQ.Tests/Hosting/TestUseActiveMq.cs b/Obvs.ActiveMQ.Tests/Hosting/TestUseActiveMq.cs new file mode 100644 index 0000000..5775f79 --- /dev/null +++ b/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 + { + ["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("orders", service => service + .UseActiveMq(configuration.GetSection("Obvs:Orders:ActiveMq")) + .SerializedWith(new JsonMessageSerializer(), new JsonMessageDeserializerFactory(typeof(JsonMessageDeserializer<>)))))); + + var options = host.Services.GetRequiredService>().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("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>().Get("orders").BrokerUri); + } + + [Fact] + public async Task ShouldFailToStartWhenTheBrokerUriIsMissingFromConfiguration() + { + var configuration = Configuration(new Dictionary()); + using var host = BuildHost(services => services.AddObvs(obvs => obvs + .AddService("orders", service => service + .UseActiveMq(configuration.GetSection("Obvs:Orders:ActiveMq")) + .SerializedWith(new JsonMessageSerializer(), new JsonMessageDeserializerFactory(typeof(JsonMessageDeserializer<>)))))); + + var exception = await Assert.ThrowsAsync(() => 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("orders", service => service + .UseActiveMq(options => { }) + .SerializedWith(new JsonMessageSerializer(), new JsonMessageDeserializerFactory(typeof(JsonMessageDeserializer<>)))))); + + var exception = await Assert.ThrowsAsync(() => 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("orders", service => service.UseActiveMq(options => options.BrokerUri = "activemq:tcp://activemq.example:61616")))); + + var exception = await Assert.ThrowsAsync(() => host.StartAsync(TestContext.Current.CancellationToken)); + + var error = Assert.Single(exception.Errors); + Assert.Contains("'orders'", error); + Assert.Contains("SerializedWith", error); + } + + private static IConfiguration Configuration(IDictionary values) + { + return new ConfigurationBuilder().AddInMemoryCollection(values).Build(); + } + + private static IHost BuildHost(Action configure) + { + var builder = Host.CreateApplicationBuilder(new HostApplicationBuilderSettings { DisableDefaults = true }); + configure(builder.Services); + return builder.Build(); + } + } +} diff --git a/Obvs.ActiveMQ.Tests/Obvs.ActiveMQ.Tests.csproj b/Obvs.ActiveMQ.Tests/Obvs.ActiveMQ.Tests.csproj index 185f3b4..af1a473 100644 --- a/Obvs.ActiveMQ.Tests/Obvs.ActiveMQ.Tests.csproj +++ b/Obvs.ActiveMQ.Tests/Obvs.ActiveMQ.Tests.csproj @@ -9,6 +9,7 @@ + @@ -19,5 +20,6 @@ + \ No newline at end of file diff --git a/Obvs.ActiveMQ.Tests/TestBroker.cs b/Obvs.ActiveMQ.Tests/TestBroker.cs new file mode 100644 index 0000000..f2c7f64 --- /dev/null +++ b/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"; + } +} diff --git a/Obvs.ActiveMQ.Tests/TestMessageSource.cs b/Obvs.ActiveMQ.Tests/TestMessageSource.cs index b549cef..f4efe40 100644 --- a/Obvs.ActiveMQ.Tests/TestMessageSource.cs +++ b/Obvs.ActiveMQ.Tests/TestMessageSource.cs @@ -1,7 +1,9 @@ using System; using System.Collections; using System.IO; +using System.Linq; using System.Text; +using System.Threading; using System.Xml.Serialization; using Apache.NMS; using Apache.NMS.Util; @@ -58,6 +60,13 @@ public TestMessageSource() _acknowledgementMode); } + // Messages reach subscribers on the task pool, so wait for delivery rather than expecting it on the listener's thread + private void AssertDelivered(ITestMessage message) + { + SpinWait.SpinUntil(() => Fake.GetCalls(_observer).Any(call => call.Method.Name == nameof(_observer.OnNext) && ReferenceEquals(call.Arguments[0], message)), TimeSpan.FromSeconds(5)); + A.CallTo(() => _observer.OnNext(message)).MustHaveHappened(); + } + [Fact] public void ShouldConnectToBrokerWhenSubscribedTo() { @@ -143,7 +152,7 @@ public void ShouldDeserializeAndPublishMessageWhenReceived() _source.Messages.Subscribe(_observer); consumer.Listener += Raise.FreeForm.With((Apache.NMS.IMessage) bytesMessage); - A.CallTo(() => _observer.OnNext(message)).MustHaveHappened(); + AssertDelivered(message); } [Fact] @@ -161,7 +170,7 @@ public void ShouldDeserializeByteMessagesAndPublishMessageWhenReceived() _source.Messages.Subscribe(_observer); consumer.Listener += Raise.FreeForm.With((Apache.NMS.IMessage) bytesMessage); - A.CallTo(() => _observer.OnNext(message)).MustHaveHappened(); + AssertDelivered(message); } [Fact] @@ -188,7 +197,7 @@ public void ShouldDeserializeAndPublishMessageOfRightTypeName() _source.Messages.Subscribe(_observer); consumer.Listener += Raise.FreeForm.With((Apache.NMS.IMessage) bytesMessage); - A.CallTo(() => _observer.OnNext(message)).MustHaveHappened(); + AssertDelivered(message); } [Fact] @@ -214,7 +223,7 @@ public void ShouldProcessMessagesWithoutTypeNameIfOnlyOneDeserializerSupplied() consumer.Listener += Raise.FreeForm.With((Apache.NMS.IMessage) bytesMessage); A.CallTo(() => _deserializer.Deserialize(A._)).MustHaveHappened(); - A.CallTo(() => _observer.OnNext(message)).MustHaveHappened(); + AssertDelivered(message); } [Fact] @@ -248,7 +257,7 @@ public void ShouldProcessMessagesAccordingToFilter() consumer.Listener += Raise.FreeForm.With((Apache.NMS.IMessage) bytesMessage); A.CallTo(() => _deserializer.Deserialize(A._)).MustHaveHappened(); - A.CallTo(() => _observer.OnNext(message)).MustHaveHappened(); + AssertDelivered(message); } } } \ No newline at end of file diff --git a/Obvs.ActiveMQ/ActiveMqOptions.cs b/Obvs.ActiveMQ/ActiveMqOptions.cs new file mode 100644 index 0000000..490e03d --- /dev/null +++ b/Obvs.ActiveMQ/ActiveMqOptions.cs @@ -0,0 +1,18 @@ +namespace Obvs.ActiveMQ +{ + /// + /// A service's ActiveMQ settings. Each service has its own, as named options keyed by the service's + /// name: IOptionsMonitor<ActiveMqOptions>.Get("orders"). + /// + public sealed class ActiveMqOptions + { + /// The broker to connect to, for example activemq:tcp://localhost:61616. Required. + public string BrokerUri { get; set; } + + /// The user name to connect as, if the broker requires one. + public string UserName { get; set; } + + /// The password for . + public string Password { get; set; } + } +} diff --git a/Obvs.ActiveMQ/ActiveMqTransportExtensions.cs b/Obvs.ActiveMQ/ActiveMqTransportExtensions.cs new file mode 100644 index 0000000..50f67a8 --- /dev/null +++ b/Obvs.ActiveMQ/ActiveMqTransportExtensions.cs @@ -0,0 +1,96 @@ +using System; +using System.Collections.Generic; +using Apache.NMS.ActiveMQ; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using Obvs.ActiveMQ; +using Obvs.ActiveMQ.Configuration; +using Obvs.Configuration; +using Obvs.Types; +using AcknowledgementMode = Obvs.ActiveMQ.AcknowledgementMode; + +// In the builder's namespace, so UseActiveMq() is found without an extra using +namespace Obvs.Hosting +{ + public static class ActiveMqTransportExtensions + { + /// + /// Carries this service's messages over ActiveMQ topics, with settings bound from , + /// for example builder.Configuration.GetSection("Obvs:Orders:ActiveMq"). + /// The service also needs a serializer: call SerializedWith(...). + /// + public static ObvsServiceBuilder UseActiveMq(this ObvsServiceBuilder service, IConfiguration configuration) + { + if (service == null) + { + throw new ArgumentNullException(nameof(service)); + } + if (configuration == null) + { + throw new ArgumentNullException(nameof(configuration)); + } + + service.Services.AddOptions(service.Name).Bind(configuration); + var brokerUriKey = configuration is IConfigurationSection section ? section.Path + ":BrokerUri" : "BrokerUri"; + return UseActiveMq(service, $"configuration key {brokerUriKey}"); + } + + /// + /// Carries this service's messages over ActiveMQ topics, with settings set in code. + /// The service also needs a serializer: call SerializedWith(...). + /// + public static ObvsServiceBuilder UseActiveMq(this ObvsServiceBuilder service, Action configure) + { + if (service == null) + { + throw new ArgumentNullException(nameof(service)); + } + if (configure == null) + { + throw new ArgumentNullException(nameof(configure)); + } + + service.Services.AddOptions(service.Name).Configure(configure); + return UseActiveMq(service, "options.BrokerUri in UseActiveMq(...)"); + } + + private static ObvsServiceBuilder UseActiveMq(ObvsServiceBuilder service, string whereToSetBrokerUri) + { + return service + .ValidateWith(serviceProvider => Validate(service, GetOptions(serviceProvider, service), whereToSetBrokerUri)) + .UseEndpointProvider(serviceProvider => CreateProvider(service, GetOptions(serviceProvider, service))); + } + + private static ActiveMqOptions GetOptions(IServiceProvider serviceProvider, ObvsServiceBuilder service) + { + return serviceProvider.GetRequiredService>().Get(service.Name); + } + + private static IEnumerable Validate(ObvsServiceBuilder service, ActiveMqOptions options, string whereToSetBrokerUri) + { + if (string.IsNullOrWhiteSpace(options.BrokerUri)) + { + yield return $"Service '{service.Name}' has no ActiveMQ broker. Set {whereToSetBrokerUri}, for example activemq:tcp://localhost:61616."; + } + if (service.Serializer == null || service.DeserializerFactory == null) + { + yield return $"Service '{service.Name}' uses ActiveMQ but has no serializer. Call SerializedWith(...) on it inside AddService(\"{service.Name}\", ...)."; + } + } + + // The provider is generic in the service's message type, which the builder only knows at run time. + // Client IDs get the bus's instance ID, as on .NET Framework they're otherwise the same for every bus + // in the process, and the broker refuses a second connection with the same client ID. + private static IServiceEndpointProvider CreateProvider(ObvsServiceBuilder service, ActiveMqOptions options) + { + var providerType = typeof(ActiveMQServiceEndpointProvider<,,,,,>).MakeGenericType( + service.ServiceMessageType, typeof(IMessage), typeof(ICommand), typeof(IEvent), typeof(IRequest), typeof(IResponse)); + Action configureConnectionFactory = factory => factory.ClientId = $"{factory.ClientId}-{service.InstanceId}"; + + return (IServiceEndpointProvider)Activator.CreateInstance(providerType, + service.Name, options.BrokerUri, service.Serializer, service.DeserializerFactory, new List>(), + null, null, null, null, null, null, options.UserName, options.Password, configureConnectionFactory, false); + } + } +} diff --git a/Obvs.ActiveMQ/MessagePublisher.cs b/Obvs.ActiveMQ/MessagePublisher.cs index 7eb9dbc..8ea69d1 100644 --- a/Obvs.ActiveMQ/MessagePublisher.cs +++ b/Obvs.ActiveMQ/MessagePublisher.cs @@ -5,6 +5,7 @@ using System.Threading.Tasks; using Apache.NMS; using Obvs.ActiveMQ.Extensions; +using Obvs.Hosting; using Obvs.MessageProperties; using Obvs.Serialization; @@ -108,6 +109,7 @@ private void Publish(TMessage message, Dictionary properties) var msg = GenerateMessage(message, _producer, _serializer); msg.Properties.AddProperties(properties); + SetHeaders(msg, message); _producer.Send(msg, _deliveryMode(message), _priority(message), _timeToLive(message)); } @@ -128,6 +130,19 @@ protected virtual IMessage GenerateMessage(TMessage message, IMessageProducer pr return bytesMessage; } + // The message's Obvs headers (such as trace context) as message properties, which v6 receivers ignore. + // The type name property Obvs relies on is never overwritten. + private static void SetHeaders(IMessage msg, TMessage message) + { + foreach (var header in ObvsMessageHeaders.Get(message)) + { + if (header.Key != MessagePropertyNames.TypeName) + { + msg.Properties.SetString(header.Key, header.Value); + } + } + } + private static void AppendTypeNameProperty(TMessage message, Dictionary properties) { try diff --git a/Obvs.ActiveMQ/MessageSource.cs b/Obvs.ActiveMQ/MessageSource.cs index e9876fe..d64bd61 100644 --- a/Obvs.ActiveMQ/MessageSource.cs +++ b/Obvs.ActiveMQ/MessageSource.cs @@ -4,11 +4,14 @@ using System.Collections.Generic; using System.IO; using System.Linq; +using System.Reactive.Concurrency; using System.Reactive.Disposables; using System.Reactive.Linq; using Apache.NMS; using Obvs.ActiveMQ.Extensions; using Obvs.ActiveMQ.Utils; +using Obvs.Hosting; +using Obvs.MessageProperties; using Obvs.Serialization; namespace Obvs.ActiveMQ @@ -50,6 +53,10 @@ public MessageSource(Lazy lazyConnection, .Select(message => new { message, deserializer = GetDeserializer(message) }) .Where(msg => msg.deserializer != null) .Select(msg => DeserializeAndAcknowledge(msg.message, msg.deserializer)) + // Hand messages to subscribers on the task pool, in order. Subscriber code running on the + // session's dispatch thread could otherwise deadlock it, for example by disposing the bus, + // which stops the connection and waits for that thread. + .ObserveOn(TaskPoolScheduler.Default) .Subscribe(observer); return Disposable.Create(() => @@ -81,10 +88,33 @@ private bool PassesFilter(IMessage message) private TMessage DeserializeAndAcknowledge(IMessage message, IMessageDeserializer deserializer) { var deserializedMessage = DeserializeMessage(message, deserializer); + SetHeaders(deserializedMessage, message); Acknowledge(message); return deserializedMessage; } + // Text properties become the message's Obvs headers, apart from the type name Obvs sets itself + private static void SetHeaders(TMessage deserializedMessage, IMessage message) + { + if (deserializedMessage == null) + { + return; + } + + var headers = new Dictionary(); + foreach (string key in message.Properties.Keys) + { + if (key != MessagePropertyNames.TypeName && message.Properties[key] is string value) + { + headers[key] = value; + } + } + if (headers.Count > 0) + { + ObvsMessageHeaders.Set(deserializedMessage, headers); + } + } + private IMessageDeserializer GetDeserializer(IMessage message) { string typeName; diff --git a/Obvs.ActiveMQ/Obvs.ActiveMQ.csproj b/Obvs.ActiveMQ/Obvs.ActiveMQ.csproj index 1d8e454..759cd18 100644 --- a/Obvs.ActiveMQ/Obvs.ActiveMQ.csproj +++ b/Obvs.ActiveMQ/Obvs.ActiveMQ.csproj @@ -11,6 +11,7 @@ +