diff --git a/.github/workflows/dotnet.yml b/.github/workflows/dotnet.yml index 18ee687c..a844f322 100644 --- a/.github/workflows/dotnet.yml +++ b/.github/workflows/dotnet.yml @@ -17,6 +17,39 @@ jobs: contents: read id-token: write # NuGet Trusted Publishing + # Brokers for the transport tests marked Category=Explicit + services: + rabbitmq: + image: rabbitmq:4.4-alpine + ports: ['5672:5672'] + options: >- + --health-cmd "rabbitmq-diagnostics -q check_port_connectivity" + --health-interval 5s --health-timeout 5s --health-retries 20 + nats: + image: nats:2.15 + ports: ['4222:4222'] + kafka: + image: apache/kafka:4.4.0 + ports: ['9092:9092'] + # Setting any KAFKA_* variable replaces the image's defaults, so configure a full single-node KRaft broker + env: + KAFKA_NODE_ID: 1 + KAFKA_PROCESS_ROLES: broker,controller + KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093 + KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 + KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER + KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT + KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093 + KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 + KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS: 1 + KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 + KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 + KAFKA_SHARE_COORDINATOR_STATE_TOPIC_REPLICATION_FACTOR: 1 + KAFKA_SHARE_COORDINATOR_STATE_TOPIC_MIN_ISR: 1 + options: >- + --health-cmd "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092" + --health-interval 5s --health-timeout 10s --health-retries 20 + steps: - name: Checkout uses: actions/checkout@v7 @@ -40,6 +73,17 @@ jobs: run: dotnet restore - name: Build run: dotnet build --no-restore --configuration Release -p:Version=${{ steps.gitversion.outputs.semVer }} -p:ContinuousIntegrationBuild=true -p:Deterministic=true + - name: Create Kafka test topic + run: docker exec ${{ job.services.kafka.id }} /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --if-not-exists --topic test-topic-3 + - name: Test transports against brokers + env: + RABBITMQ_URI: amqp://localhost:5672 + NATS_URL: nats://localhost:4222 + KAFKA_BOOTSTRAP_SERVERS: localhost:9092 + 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" - 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/Directory.Packages.props b/Directory.Packages.props index 13980655..f75f379e 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -7,11 +7,11 @@ + - - + diff --git a/Examples/docker-compose.yml b/Examples/docker-compose.yml index 3cce7010..5e69a09d 100644 --- a/Examples/docker-compose.yml +++ b/Examples/docker-compose.yml @@ -1,7 +1,10 @@ -version: '3.5' services: amq: - image: rmohr/activemq:latest + image: apache/activemq-classic:6.2.0 + environment: + # Require the credentials the examples connect with + - ACTIVEMQ_CONNECTION_USER=admin + - ACTIVEMQ_CONNECTION_PASSWORD=admin ports: - "61616:61616" - "8161:8161" @@ -20,4 +23,4 @@ services: networks: obvs: - name: obvs \ No newline at end of file + name: obvs diff --git a/Obvs.Kafka.Tests/TestBroker.cs b/Obvs.Kafka.Tests/TestBroker.cs new file mode 100644 index 00000000..674d5b38 --- /dev/null +++ b/Obvs.Kafka.Tests/TestBroker.cs @@ -0,0 +1,10 @@ +using System; + +namespace Obvs.Kafka.Tests +{ + internal static class TestBroker + { + // Set KAFKA_BOOTSTRAP_SERVERS to run the Explicit tests against a broker other than a default local install + public static string BootstrapServers => Environment.GetEnvironmentVariable("KAFKA_BOOTSTRAP_SERVERS") ?? "localhost:9092"; + } +} diff --git a/Obvs.Kafka.Tests/TestMessageSendAndReceive.cs b/Obvs.Kafka.Tests/TestMessageSendAndReceive.cs index 714639e8..f62b1afa 100644 --- a/Obvs.Kafka.Tests/TestMessageSendAndReceive.cs +++ b/Obvs.Kafka.Tests/TestMessageSendAndReceive.cs @@ -1,5 +1,7 @@ using System; +using System.Collections.Concurrent; using System.Collections.Generic; +using System.Linq; using System.Reactive.Linq; using System.Threading.Tasks; using System.Xml.Serialization; @@ -44,13 +46,8 @@ public override string ToString() public TestMessageSendAndReceive() { - // Steps to run local kafka broker: - // - Install: Docker for Windows / Docker Desktop - // - Clone: https://github.com/wurstmeister/kafka-docker - // - Edit: docker-compose-single-broker.yml - // - Change: KAFKA_ADVERTISED_HOST_NAME: localhost - // - Run: docker-compose -f "docker-compose-single-broker.yml" up -d --build - var cfg = new KafkaConfiguration("localhost:9092"); + // Run a local broker with: docker run -p 9092:9092 apache/kafka + var cfg = new KafkaConfiguration(TestBroker.BootstrapServers); var sourceConfig = new KafkaSourceConfiguration(); var producerConfig = new KafkaProducerConfiguration(); const string topicName = "test-topic-3"; @@ -100,5 +97,27 @@ public async Task TestSendAndReceiveMessage() sub.Dispose(); } + + [Fact] + [Trait("Category", "Explicit")] + public async Task TestMessagePublishedStraightAfterSubscribingIsReceived() + { + var data = $"straight after subscribing {Guid.NewGuid()}"; + var messages = new ConcurrentQueue(); + + using (_source.Messages.OfType().Subscribe(messages.Enqueue)) + { + // No delay: once Subscribe returns, a published message must not be missed + await _publisher.PublishAsync(new TestMessage(data)); + + var deadline = DateTime.UtcNow.AddSeconds(10); + while (!messages.Any(message => message.Data == data) && DateTime.UtcNow < deadline) + { + await Task.Delay(100, TestContext.Current.CancellationToken); + } + } + + Assert.Contains(messages, message => message.Data == data); + } } } \ No newline at end of file diff --git a/Obvs.Kafka/MessageSource.cs b/Obvs.Kafka/MessageSource.cs index 2a76fd0e..405b85ec 100644 --- a/Obvs.Kafka/MessageSource.cs +++ b/Obvs.Kafka/MessageSource.cs @@ -24,6 +24,10 @@ public class MessageSource : IMessageSource private readonly ConsumerConfig _consumerConfig; + // How long Subscribe waits for Kafka to assign partitions, and for each offset query while it does + private static readonly TimeSpan PartitionAssignmentTimeout = TimeSpan.FromSeconds(30); + private static readonly TimeSpan OffsetQueryTimeout = TimeSpan.FromSeconds(10); + public MessageSource(KafkaConfiguration kafkaConfig, KafkaSourceConfiguration sourceConfig, string topicName, @@ -60,7 +64,8 @@ public IObservable Messages var task = Task.Run(() => Subscriber(token, observer, subscribedEvent)); - subscribedEvent.Wait(); + // Wait until partitions are assigned, so messages published once Subscribe returns are received + subscribedEvent.Wait(PartitionAssignmentTimeout); return Disposable.Create(() => { @@ -78,10 +83,15 @@ private void Subscriber(CancellationToken token, IObserver observer, M using (var consumer = new ConsumerBuilder(_consumerConfig) .SetValueDeserializer(new ConsumerValueDeserializer()) .SetErrorHandler((_, e) => observer.OnError(new Exception($"Consumer ErrorHandler: {e.Reason}"))) + .SetPartitionsAssignedHandler((c, partitions) => + { + var offsets = StartOffsets(c, partitions); + subscribedEvent.Set(); + return offsets; + }) .Build()) { consumer.Subscribe(_topicName); - subscribedEvent.Set(); try { @@ -114,6 +124,17 @@ private void Subscriber(CancellationToken token, IObserver observer, M } } + // Fix each partition's start position at assignment: the group's committed offset, or else the current + // end of the log. Leaving "latest" to be resolved later would skip messages published in the meantime. + private static IEnumerable StartOffsets(IConsumer consumer, List partitions) + { + return consumer.Committed(partitions, OffsetQueryTimeout) + .Select(committed => committed.Offset != Offset.Unset + ? committed + : new TopicPartitionOffset(committed.TopicPartition, consumer.QueryWatermarkOffsets(committed.TopicPartition, OffsetQueryTimeout).High)) + .ToList(); + } + private void ConsumeMessage(CancellationToken token, IObserver observer, IConsumer consumer) { try diff --git a/Obvs.Monitoring.ElasticSearch/Configuration/ElasticSearchMonitoringConfig.cs b/Obvs.Monitoring.ElasticSearch/Configuration/ElasticSearchMonitoringConfig.cs index 9c25cab0..3c17c37d 100644 --- a/Obvs.Monitoring.ElasticSearch/Configuration/ElasticSearchMonitoringConfig.cs +++ b/Obvs.Monitoring.ElasticSearch/Configuration/ElasticSearchMonitoringConfig.cs @@ -1,8 +1,8 @@ using System; using System.Collections.Generic; using System.Reactive.Concurrency; -using Elasticsearch.Net; -using Nest; +using Elastic.Clients.Elasticsearch; +using Elastic.Transport; using Obvs.Configuration; namespace Obvs.Monitoring.ElasticSearch.Configuration @@ -57,7 +57,7 @@ public interface ICanSpecifyElasticSearchMonitoringUri ConnectToServer(string uri); - ICanAddEndpointOrLoggingOrCorrelationOrCreate ConnectToServer(IConnectionPool elasticConnectionPool); + ICanAddEndpointOrLoggingOrCorrelationOrCreate ConnectToServer(NodePool elasticConnectionPool); } internal class ElasticSearchMonitoringConfig : @@ -94,18 +94,18 @@ public ICanAddEndpointOrLoggingOrCorrelationOrCreate( _indexPrefix, _types, _instanceName, _samplePeriod, Scheduler.Default, - new ElasticClient(new ConnectionSettings(new Uri(uri)))); + new ElasticsearchClient(new ElasticsearchClientSettings(new Uri(uri)))); _config.UsingMonitor(monitorFactory); return _config; } - public ICanAddEndpointOrLoggingOrCorrelationOrCreate ConnectToServer(IConnectionPool elasticConnectionPool) + public ICanAddEndpointOrLoggingOrCorrelationOrCreate ConnectToServer(NodePool elasticConnectionPool) { var monitorFactory = new ElasticSearchMonitorFactory( _indexPrefix, _types, _instanceName, _samplePeriod, Scheduler.Default, - new ElasticClient(new ConnectionSettings(elasticConnectionPool))); + new ElasticsearchClient(new ElasticsearchClientSettings(elasticConnectionPool))); _config.UsingMonitor(monitorFactory); return _config; diff --git a/Obvs.Monitoring.ElasticSearch/ElasticSearchMonitor.cs b/Obvs.Monitoring.ElasticSearch/ElasticSearchMonitor.cs index 5d2d18df..b94da2f2 100644 --- a/Obvs.Monitoring.ElasticSearch/ElasticSearchMonitor.cs +++ b/Obvs.Monitoring.ElasticSearch/ElasticSearchMonitor.cs @@ -6,7 +6,7 @@ using System.Reactive.Concurrency; using System.Reactive.Linq; using System.Reactive.Subjects; -using Nest; +using Elastic.Clients.Elasticsearch; namespace Obvs.Monitoring.ElasticSearch { @@ -22,7 +22,7 @@ static class Directions private readonly string _counterName; private readonly string _indexPrefix; private readonly IList _types; - private readonly IElasticClient _client; + private readonly ElasticsearchClient _client; private readonly Subject> _queueSent = new Subject>(); private readonly Subject> _queueReceived = new Subject>(); private readonly TimeSpan _samplePeriod; @@ -34,7 +34,7 @@ static class Directions private readonly string _processName; private readonly IList _emptyList = new List(); - public ElasticSearchMonitor(string instanceName, string counterName, string indexPrefix, IList types, TimeSpan samplePeriod, IScheduler scheduler, IElasticClient client) + public ElasticSearchMonitor(string instanceName, string counterName, string indexPrefix, IList types, TimeSpan samplePeriod, IScheduler scheduler, ElasticsearchClient client) { _messageTypeName = typeof(TMessage).Name; _instanceName = instanceName; @@ -69,9 +69,9 @@ private void Save(IList items) { string indexName = string.Format("{0}-{1}", _indexPrefix, DateTime.Today.ToString("yyyy.MM.dd")); var response = _client.IndexMany(items, indexName); - if (!response.IsValid && response.ServerError != null) + if (!response.IsValidResponse && response.ElasticsearchServerError != null) { - Console.WriteLine(response.ServerError.Error); + Console.WriteLine(response.ElasticsearchServerError.Error); } } } diff --git a/Obvs.Monitoring.ElasticSearch/ElasticSearchMonitorFactory.cs b/Obvs.Monitoring.ElasticSearch/ElasticSearchMonitorFactory.cs index cd61666e..ffc09d89 100644 --- a/Obvs.Monitoring.ElasticSearch/ElasticSearchMonitorFactory.cs +++ b/Obvs.Monitoring.ElasticSearch/ElasticSearchMonitorFactory.cs @@ -1,7 +1,7 @@ using System; using System.Collections.Generic; using System.Reactive.Concurrency; -using Nest; +using Elastic.Clients.Elasticsearch; namespace Obvs.Monitoring.ElasticSearch { @@ -12,9 +12,9 @@ public class ElasticSearchMonitorFactory : IMonitorFactory private readonly string _instanceName; private readonly TimeSpan _samplePeriod; private readonly IScheduler _scheduler; - private readonly IElasticClient _client; + private readonly ElasticsearchClient _client; - public ElasticSearchMonitorFactory(string indexPrefx, IList types, string instanceName, TimeSpan samplePeriod, IScheduler scheduler, IElasticClient client) + public ElasticSearchMonitorFactory(string indexPrefx, IList types, string instanceName, TimeSpan samplePeriod, IScheduler scheduler, ElasticsearchClient client) { _indexPrefx = indexPrefx; _types = types; diff --git a/Obvs.Monitoring.ElasticSearch/Obvs.Monitoring.ElasticSearch.csproj b/Obvs.Monitoring.ElasticSearch/Obvs.Monitoring.ElasticSearch.csproj index cc9be8ab..ab011d8e 100644 --- a/Obvs.Monitoring.ElasticSearch/Obvs.Monitoring.ElasticSearch.csproj +++ b/Obvs.Monitoring.ElasticSearch/Obvs.Monitoring.ElasticSearch.csproj @@ -11,7 +11,7 @@ Monitoring extension for Obvs, using Elastic. - + diff --git a/Obvs.Monitoring.ElasticSearch/ObvsCounter.cs b/Obvs.Monitoring.ElasticSearch/ObvsCounter.cs index 88cc7f3f..0996de6b 100644 --- a/Obvs.Monitoring.ElasticSearch/ObvsCounter.cs +++ b/Obvs.Monitoring.ElasticSearch/ObvsCounter.cs @@ -1,5 +1,4 @@ using System; -using Nest; namespace Obvs.Monitoring.ElasticSearch { diff --git a/Obvs.Monitoring.Tests/TestElasticSearchMonitoring.cs b/Obvs.Monitoring.Tests/TestElasticSearchMonitoring.cs index 3835d8b8..31231d89 100644 --- a/Obvs.Monitoring.Tests/TestElasticSearchMonitoring.cs +++ b/Obvs.Monitoring.Tests/TestElasticSearchMonitoring.cs @@ -1,8 +1,10 @@ using System; using System.Collections.Generic; -using FakeItEasy; +using System.Linq; +using System.Text; +using Elastic.Clients.Elasticsearch; +using Elastic.Transport; using Microsoft.Reactive.Testing; -using Nest; using Obvs.Monitoring.ElasticSearch; using Xunit; @@ -10,13 +12,40 @@ namespace Obvs.Monitoring.Tests { public class TestElasticSearchMonitoring { + private readonly List _calls = new List(); + private readonly ElasticsearchClient _elasticClient; + + public TestElasticSearchMonitoring() + { + // Send requests to an in-memory transport that answers like a real server, and record them + var invoker = new InMemoryRequestInvoker( + Encoding.UTF8.GetBytes("{\"took\":1,\"errors\":false,\"items\":[]}"), 200, null, "application/json", + new Dictionary> { ["x-elastic-product"] = new[] { "Elasticsearch" } }); + var settings = new ElasticsearchClientSettings(invoker) + .DisableDirectStreaming() + .OnRequestCompleted(_calls.Add); + _elasticClient = new ElasticsearchClient(settings); + } + + private IEnumerable BulkRequests() + { + return _calls.Where(call => call.Uri.AbsolutePath.EndsWith("/_bulk")); + } + + // A bulk request body has an action line then a document line for each indexed counter + private static int IndexOperations(ApiCallDetails call) + { + return Encoding.UTF8.GetString(call.RequestBodyInBytes) + .Split('\n') + .Count(line => line.StartsWith("{\"index\"")); + } + [Fact] public void ShouldBeAbleToCreateMonitor() { - IElasticClient elasticClient = A.Fake(); TestScheduler testScheduler = new TestScheduler(); - IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient); + IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient); IMonitor monitor = factory.Create("SomeName"); @@ -26,10 +55,9 @@ public void ShouldBeAbleToCreateMonitor() [Fact] public void ShouldNotAttemptToSaveIfNoMessages() { - IElasticClient elasticClient = A.Fake(); TestScheduler testScheduler = new TestScheduler(); - IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient); + IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient); IMonitor monitor = factory.Create("SomeName"); @@ -37,16 +65,15 @@ public void ShouldNotAttemptToSaveIfNoMessages() testScheduler.AdvanceBy(TimeSpan.FromMinutes(1).Ticks); - A.CallTo(() => elasticClient.Bulk(A._)).MustNotHaveHappened(); + Assert.Empty(BulkRequests()); } [Fact] public void ShouldAttemptToSaveIfMessagesSent() { - IElasticClient elasticClient = A.Fake(); TestScheduler testScheduler = new TestScheduler(); - IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient); + IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient); IMonitor monitor = factory.Create("SomeName"); @@ -54,16 +81,15 @@ public void ShouldAttemptToSaveIfMessagesSent() testScheduler.AdvanceBy(TimeSpan.FromMinutes(1).Ticks); - A.CallTo(() => elasticClient.Bulk(A._)).MustHaveHappened(1, Times.Exactly); + Assert.Single(BulkRequests()); } [Fact] public void ShouldAttemptToSaveIfMessagesReceived() { - IElasticClient elasticClient = A.Fake(); TestScheduler testScheduler = new TestScheduler(); - IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient); + IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient); IMonitor monitor = factory.Create("SomeName"); @@ -71,18 +97,17 @@ public void ShouldAttemptToSaveIfMessagesReceived() testScheduler.AdvanceBy(TimeSpan.FromMinutes(1).Ticks); - A.CallTo(() => elasticClient.Bulk(A._)).MustHaveHappened(1, Times.Exactly); + Assert.Single(BulkRequests()); } [Fact] public void ShouldAttemptToSaveMultipleCountersIfMessagesSent() { - IElasticClient elasticClient = A.Fake(); TestScheduler testScheduler = new TestScheduler(); var types = new List { typeof(TestMessage) }; - IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", types, "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient); + IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", types, "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient); IMonitor monitor = factory.Create("SomeName"); @@ -90,17 +115,17 @@ public void ShouldAttemptToSaveMultipleCountersIfMessagesSent() testScheduler.AdvanceBy(TimeSpan.FromMinutes(1).Ticks); - A.CallTo(() => elasticClient.Bulk(A.That.Matches(request => request.Operations.Count == 1 + types.Count))).MustHaveHappened(1, Times.Exactly); + var bulk = Assert.Single(BulkRequests()); + Assert.Equal(1 + types.Count, IndexOperations(bulk)); } [Fact] public void ShouldAttemptToSaveMultipleCountersIfMessagesReceived() { - IElasticClient elasticClient = A.Fake(); TestScheduler testScheduler = new TestScheduler(); var types = new List {typeof(TestMessage)}; - IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", types, "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient); + IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", types, "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient); IMonitor monitor = factory.Create("SomeName"); @@ -108,16 +133,39 @@ public void ShouldAttemptToSaveMultipleCountersIfMessagesReceived() testScheduler.AdvanceBy(TimeSpan.FromMinutes(1).Ticks); - A.CallTo(() => elasticClient.Bulk(A.That.Matches(request => request.Operations.Count == 1 + types.Count))).MustHaveHappened(1, Times.Exactly); + var bulk = Assert.Single(BulkRequests()); + Assert.Equal(1 + types.Count, IndexOperations(bulk)); } + [Fact] + public void ShouldIndexCountersInDailyIndexWithCamelCaseFields() + { + TestScheduler testScheduler = new TestScheduler(); + + IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient); + + IMonitor monitor = factory.Create("SomeName"); + + monitor.MessageSent(new TestMessage(), TimeSpan.FromMilliseconds(1)); + + testScheduler.AdvanceBy(TimeSpan.FromMinutes(1).Ticks); + + // Same index naming and document shape as the NEST-based versions, so existing indexes keep working + var bulk = Assert.Single(BulkRequests()); + Assert.Equal($"/indexName-{DateTime.Today:yyyy.MM.dd}/_bulk", bulk.Uri.AbsolutePath); + Assert.True(bulk.HasSuccessfulStatusCode); + var body = Encoding.UTF8.GetString(bulk.RequestBodyInBytes); + Assert.Contains("\"counterName\":\"SomeName\"", body); + Assert.Contains("\"instanceName\":\"instanceName\"", body); + Assert.Contains("\"direction\":\"send\"", body); + } + [Fact] public void ShouldDisposeCleanly() { - IElasticClient elasticClient = A.Fake(); TestScheduler testScheduler = new TestScheduler(); - IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, elasticClient); + IMonitorFactory factory = new ElasticSearchMonitorFactory("indexName", new List(), "instanceName", TimeSpan.FromSeconds(1), testScheduler, _elasticClient); IMonitor monitor = factory.Create("SomeName"); diff --git a/Obvs.NATS.Tests/TestBroker.cs b/Obvs.NATS.Tests/TestBroker.cs new file mode 100644 index 00000000..e5cf4453 --- /dev/null +++ b/Obvs.NATS.Tests/TestBroker.cs @@ -0,0 +1,10 @@ +using System; + +namespace Obvs.NATS.Tests +{ + internal static class TestBroker + { + // Set NATS_URL to run the Explicit tests against a server other than a default local install + public static string Url => Environment.GetEnvironmentVariable("NATS_URL") ?? "nats://localhost:4222"; + } +} diff --git a/Obvs.NATS.Tests/TestServiceBus.cs b/Obvs.NATS.Tests/TestServiceBus.cs index be60588c..c86af939 100644 --- a/Obvs.NATS.Tests/TestServiceBus.cs +++ b/Obvs.NATS.Tests/TestServiceBus.cs @@ -21,7 +21,7 @@ public async Task TestBasicSendReceiveOfAllMessageTypes() settings.ServiceName = "Obvs.NATS.TestService"; settings.Configure(connection => { - connection.Url = "nats://192.168.99.100:32774"; // change to local Docker address:port that maps onto 4222 + connection.Url = TestBroker.Url; connection.IsShared = true; }); settings.Configure(messageProperty => @@ -42,16 +42,19 @@ public async Task TestBasicSendReceiveOfAllMessageTypes() Console.WriteLine(request); bus.ReplyAsync(request, new TestResponse {Id = request.Id}); }); - bus.Events.Subscribe(Console.WriteLine); + var events = new List(); + bus.Events.Subscribe(events.Add); await bus.SendAsync(new TestCommand {Id = 1}); - var response = await bus.GetResponse(new TestRequest()); - Console.WriteLine(response); + var response = await bus.GetResponse(new TestRequest {Id = 2}); + Assert.Equal(2, response.Id); await Task.Delay(TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); ((IDisposable)bus).Dispose(); + + Assert.Contains(events, ev => ev is TestEvent testEvent && testEvent.Id == 1); } } } diff --git a/Obvs.NATS.Tests/TestSourceAndPublisher.cs b/Obvs.NATS.Tests/TestSourceAndPublisher.cs index ea880a1a..256937bd 100644 --- a/Obvs.NATS.Tests/TestSourceAndPublisher.cs +++ b/Obvs.NATS.Tests/TestSourceAndPublisher.cs @@ -1,10 +1,12 @@ using System; +using System.Collections.Concurrent; using System.Collections.Generic; +using System.Linq; using System.Reactive.Concurrency; using System.Reactive.Linq; using System.Threading; using System.Threading.Tasks; -using NATS.Client; +using NATS.Client.Core; using Obvs.Serialization; using Obvs.Serialization.Json; using Xunit; @@ -16,11 +18,7 @@ public class TestSourceAndPublisher [Fact, Trait("Category", "Explicit")] public async Task TestSendAndReceiveMessage() { - var lazyConnection = new Lazy(() => - { - var cf = new ConnectionFactory(); - return cf.CreateConnection("nats://192.168.99.100:32774"); // change to local Docker address:port that maps onto 4222 - }, LazyThreadSafetyMode.ExecutionAndPublication); + await using var connection = new NatsConnection(new NatsOpts { Url = TestBroker.Url }); const string subjectPrefix = "Obvs.NATS.Tests"; var deserializers = new List> @@ -29,10 +27,11 @@ public async Task TestSendAndReceiveMessage() }; IMessageSerializer serializer = new JsonMessageSerializer(); - var source = new MessageSource(lazyConnection, subjectPrefix, deserializers); - var publisher = new MessagePublisher(lazyConnection, subjectPrefix, serializer); + var source = new MessageSource(connection, subjectPrefix, deserializers); + var publisher = new MessagePublisher(connection, subjectPrefix, serializer); - var subscription = source.Messages.ObserveOn(Scheduler.Default).Subscribe(Console.WriteLine); + var received = new ConcurrentQueue(); + var subscription = source.Messages.ObserveOn(Scheduler.Default).Subscribe(received.Enqueue); for (var i = 0; i < 10; i++) { @@ -41,6 +40,8 @@ public async Task TestSendAndReceiveMessage() await Task.Delay(TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); subscription.Dispose(); + + Assert.Equal(Enumerable.Range(0, 10), received.Select(message => message.Id)); } } } diff --git a/Obvs.NATS/AsyncSubscriptionExtensions.cs b/Obvs.NATS/AsyncSubscriptionExtensions.cs deleted file mode 100644 index 45f475f5..00000000 --- a/Obvs.NATS/AsyncSubscriptionExtensions.cs +++ /dev/null @@ -1,26 +0,0 @@ -using System; -using System.Reactive.Disposables; -using System.Reactive.Linq; -using NATS.Client; - -namespace Obvs.NATS -{ - internal static class AsyncSubscriptionExtensions - { - public static IObservable ToObservable(this IAsyncSubscription sub) - { - return Observable.Create( - observer => - { - EventHandler handler = (sender, args) => observer.OnNext(args); - sub.MessageHandler += handler; - sub.Start(); - return Disposable.Create(() => - { - sub.MessageHandler -= handler; - sub.Unsubscribe(); - }); - }); - } - } -} \ No newline at end of file diff --git a/Obvs.NATS/Configuration/NatsServiceEndpointProvider.cs b/Obvs.NATS/Configuration/NatsServiceEndpointProvider.cs index 09de932e..239952df 100644 --- a/Obvs.NATS/Configuration/NatsServiceEndpointProvider.cs +++ b/Obvs.NATS/Configuration/NatsServiceEndpointProvider.cs @@ -2,7 +2,7 @@ using System.Collections.Generic; using System.Reactive.Disposables; using System.Reflection; -using NATS.Client; +using NATS.Client.Core; using Obvs.Configuration; using Obvs.Serialization; @@ -24,7 +24,7 @@ public class NatsServiceEndpointProvider, bool> _propertyFilter; private readonly Func> _propertyProvider; - private readonly Lazy _connection; + private readonly INatsConnection _connection; public NatsServiceEndpointProvider(NatsEndpointSettings settings, IMessageSerializer serializer, IMessageDeserializerFactory deserializerFactory, Func assemblyFilter = null, Func typeFilter = null, Func, bool> propertyFilter = null, Func> propertyProvider = null) : base(settings.ServiceName) @@ -56,14 +56,14 @@ public override IServiceEndpoint CreateSource(Lazy connection, string subject) where T : class, TMessage + private IMessageSource CreateSource(INatsConnection connection, string subject) where T : class, TMessage { return new MessageSource(connection, subject, _deserializerFactory.Create(_assemblyFilter, _typeFilter), _propertyFilter); } - private IMessagePublisher CreatePublisher(Lazy connection, string destination) where T : class, TMessage + private IMessagePublisher CreatePublisher(INatsConnection connection, string destination) where T : class, TMessage { return new MessagePublisher(connection, destination, _serializer, _propertyProvider); } @@ -80,19 +80,11 @@ public override IServiceEndpointClient lazyConnection) + private static IDisposable GetConnectionDisposable(INatsConnection connection) { - return Disposable.Create(() => - { - if (lazyConnection.IsValueCreated) - { - if (!lazyConnection.Value.IsClosed()) - { - lazyConnection.Value.Close(); - } - lazyConnection.Value.Dispose(); - } - }); +#pragma warning disable VSTHRD002 // IDisposable.Dispose is synchronous + return Disposable.Create(() => connection.DisposeAsync().AsTask().GetAwaiter().GetResult()); +#pragma warning restore VSTHRD002 } } } \ No newline at end of file diff --git a/Obvs.NATS/Configuration/SharedConnectionFactory.cs b/Obvs.NATS/Configuration/SharedConnectionFactory.cs index fe23658f..0fbeb19c 100644 --- a/Obvs.NATS/Configuration/SharedConnectionFactory.cs +++ b/Obvs.NATS/Configuration/SharedConnectionFactory.cs @@ -1,33 +1,31 @@ -using System; using System.Collections.Generic; -using System.Threading; -using NATS.Client; +using NATS.Client.Core; namespace Obvs.NATS.Configuration { internal static class SharedConnectionFactory { - private static readonly Dictionary> SharedConnections = new Dictionary>(); + private static readonly Dictionary SharedConnections = new Dictionary(); - public static Lazy Get(string url, bool isShared) + public static INatsConnection Get(string url, bool isShared) { if (!isShared) { - return CreateLazyConnection(url); + return CreateConnection(url); } - Lazy connection; + INatsConnection connection; if (!SharedConnections.TryGetValue(url, out connection)) { - connection = CreateLazyConnection(url); + connection = CreateConnection(url); SharedConnections.Add(url, connection); } return connection; } - private static Lazy CreateLazyConnection(string url) + // A NatsConnection only connects when it's first used + private static INatsConnection CreateConnection(string url) { - var factory = new ConnectionFactory(); - return new Lazy(() => factory.CreateConnection(url), LazyThreadSafetyMode.ExecutionAndPublication); + return new NatsConnection(new NatsOpts { Url = url }); } } -} \ No newline at end of file +} diff --git a/Obvs.NATS/MessagePublisher.cs b/Obvs.NATS/MessagePublisher.cs index ec06f356..4470cdca 100644 --- a/Obvs.NATS/MessagePublisher.cs +++ b/Obvs.NATS/MessagePublisher.cs @@ -2,30 +2,30 @@ using System.Collections.Generic; using System.IO; using System.Threading.Tasks; -using NATS.Client; +using NATS.Client.Core; using Obvs.Serialization; namespace Obvs.NATS { public class MessagePublisher : IMessagePublisher where TMessage : class { - private readonly Lazy _lazyConnection; + private readonly INatsConnection _connection; private readonly string _subjectPrefix; private readonly IMessageSerializer _serializer; private readonly Func> _properties; - public MessagePublisher(Lazy lazyConnection, + public MessagePublisher(INatsConnection connection, string subjectPrefix, IMessageSerializer serializer, Func> properties = null) { - _lazyConnection = lazyConnection; + _connection = connection; _subjectPrefix = subjectPrefix; _serializer = serializer; _properties = properties ?? (message => null); } - public Task PublishAsync(TMessage message) + public async Task PublishAsync(TMessage message) { byte[] body; using (var stream = new MemoryStream()) @@ -40,9 +40,8 @@ public Task PublishAsync(TMessage message) using (var stream = new MemoryStream()) { ProtoBuf.Serializer.Serialize(stream, wrapper); - _lazyConnection.Value.Publish(subject, stream.ToArray()); + await _connection.PublishAsync(subject, stream.ToArray()).ConfigureAwait(false); } - return Task.FromResult(true); } public void Dispose() diff --git a/Obvs.NATS/MessageSource.cs b/Obvs.NATS/MessageSource.cs index ed65f189..64f4da11 100644 --- a/Obvs.NATS/MessageSource.cs +++ b/Obvs.NATS/MessageSource.cs @@ -2,8 +2,11 @@ using System.Collections.Generic; using System.IO; using System.Linq; +using System.Reactive.Disposables; using System.Reactive.Linq; -using NATS.Client; +using System.Threading; +using System.Threading.Tasks; +using NATS.Client.Core; using Obvs.Serialization; namespace Obvs.NATS @@ -12,17 +15,17 @@ public class MessageSource : IMessageSource where TMessage : class { private readonly IDictionary> _deserializers; - private readonly Lazy _lazyConnection; + private readonly INatsConnection _connection; private readonly Func, bool> _filter; private readonly string _subject; - public MessageSource(Lazy lazyConnection, + public MessageSource(INatsConnection connection, string subjectPrefix, IEnumerable> deserializers, Func, bool> filter = null) { _deserializers = deserializers.ToDictionary(d => d.GetTypeName()); - _lazyConnection = lazyConnection; + _connection = connection; _filter = filter; _subject = string.Format("{0}.*", subjectPrefix); } @@ -31,35 +34,71 @@ public IObservable Messages { get { +#pragma warning disable VSTHRD002 // IObservable.Subscribe and IDisposable.Dispose are synchronous return Observable.Create(observer => - _lazyConnection.Value - .SubscribeAsync(_subject) - .ToObservable() - .Select(Unwrap) - .Where(MatchesFilter) - .Select(Deserialize) - .Subscribe(observer)); + { + // Block until the server has the subscription, so messages published straight after + // Subscribe returns aren't missed. NATS.Net doesn't capture the synchronization context, + // so blocking here can't deadlock. + var subscription = _connection.SubscribeCoreAsync(_subject).AsTask().GetAwaiter().GetResult(); + var cancellation = new CancellationTokenSource(); + // ReceiveAsync reports its own failures to the observer + _ = Task.Run(() => ReceiveAsync(subscription, observer, cancellation.Token)); + + return Disposable.Create(() => + { + cancellation.Cancel(); + subscription.DisposeAsync().AsTask().GetAwaiter().GetResult(); + cancellation.Dispose(); + }); + }); +#pragma warning restore VSTHRD002 + } + } + + private async Task ReceiveAsync(INatsSub subscription, IObserver observer, CancellationToken token) + { + try + { + while (await subscription.Msgs.WaitToReadAsync(token).ConfigureAwait(false)) + { + NatsMsg msg; + while (subscription.Msgs.TryRead(out msg)) + { + var wrapper = Unwrap(msg.Data); + if (MatchesFilter(wrapper)) + { + observer.OnNext(Deserialize(msg.Subject, wrapper)); + } + } + } + } + catch (OperationCanceledException) + { + } + catch (Exception exception) + { + observer.OnError(exception); } } - private bool MatchesFilter(Tuple tuple) + private bool MatchesFilter(MessageWrapper wrapper) { - return _filter == null || _filter(tuple.Item2.Properties); + return _filter == null || _filter(wrapper.Properties); } - private static Tuple Unwrap(MsgHandlerEventArgs args) + private static MessageWrapper Unwrap(byte[] data) { - using (var stream = new MemoryStream(args.Message.Data)) + using (var stream = new MemoryStream(data)) { - var wrapper = ProtoBuf.Serializer.Deserialize(stream); - return Tuple.Create(args.Message.Subject, wrapper); + return ProtoBuf.Serializer.Deserialize(stream); } } - private TMessage Deserialize(Tuple tuple) + private TMessage Deserialize(string subject, MessageWrapper wrapper) { IMessageDeserializer deserializer; - var messageType = GetMessageType(tuple.Item1); + var messageType = GetMessageType(subject); if (!_deserializers.TryGetValue(messageType, out deserializer)) { @@ -70,7 +109,7 @@ private TMessage Deserialize(Tuple tuple) deserializer = _deserializers.Values.Single(); } - return deserializer.Deserialize(new MemoryStream(tuple.Item2.Body)); + return deserializer.Deserialize(new MemoryStream(wrapper.Body)); } private static string GetMessageType(string subject) diff --git a/Obvs.NATS/Obvs.NATS.csproj b/Obvs.NATS/Obvs.NATS.csproj index 7691021d..a5ad1c44 100644 --- a/Obvs.NATS/Obvs.NATS.csproj +++ b/Obvs.NATS/Obvs.NATS.csproj @@ -1,6 +1,8 @@  net472;netstandard2.0;net10.0 + + 9.0 Copyright © Christopher Read 2017 1.0.0 @@ -10,7 +12,7 @@ NATS transport support for Obvs - + diff --git a/Obvs.Serialization.MessagePack/Configuration/FluentConfigExtensions.cs b/Obvs.Serialization.MessagePack/Configuration/FluentConfigExtensions.cs index 4f2aeb8f..5e0d06cb 100644 --- a/Obvs.Serialization.MessagePack/Configuration/FluentConfigExtensions.cs +++ b/Obvs.Serialization.MessagePack/Configuration/FluentConfigExtensions.cs @@ -1,9 +1,11 @@ -using Obvs.Configuration; +using System; +using Obvs.Configuration; namespace Obvs.Serialization.MessagePack.Configuration { public static class FluentConfigExtensions { + [Obsolete("MsgPack.Cli is no longer maintained; use SerializedAsMessagePackCSharp() from Obvs.Serialization.MessagePack-CSharp instead")] public static ICanCreateEndpointAsClientOrServer SerializedAsMsgPack(this ICanSpecifyEndpointSerializers config) where TMessage : class where TCommand : class, TMessage diff --git a/Obvs.Serialization.MessagePack/MsgPackMessageDeserializer.cs b/Obvs.Serialization.MessagePack/MsgPackMessageDeserializer.cs index 0c27685b..12bd8c9e 100644 --- a/Obvs.Serialization.MessagePack/MsgPackMessageDeserializer.cs +++ b/Obvs.Serialization.MessagePack/MsgPackMessageDeserializer.cs @@ -1,8 +1,10 @@ +using System; using System.IO; using MsgPack.Serialization; namespace Obvs.Serialization.MessagePack { + [Obsolete("MsgPack.Cli is no longer maintained; use Obvs.Serialization.MessagePack-CSharp instead")] public class MsgPackMessageDeserializer : MessageDeserializerBase where TMessage : class { diff --git a/Obvs.Serialization.MessagePack/MsgPackMessageDeserializerFactory.cs b/Obvs.Serialization.MessagePack/MsgPackMessageDeserializerFactory.cs index 0d9fe041..f25cde94 100644 --- a/Obvs.Serialization.MessagePack/MsgPackMessageDeserializerFactory.cs +++ b/Obvs.Serialization.MessagePack/MsgPackMessageDeserializerFactory.cs @@ -6,6 +6,7 @@ namespace Obvs.Serialization.MessagePack { + [Obsolete("MsgPack.Cli is no longer maintained; use Obvs.Serialization.MessagePack-CSharp instead")] public class MsgPackMessageDeserializerFactory : IMessageDeserializerFactory { public IEnumerable> Create(Func assemblyFilter = null, Func typeFilter = null) diff --git a/Obvs.Serialization.MessagePack/MsgPackMessageSerializer.cs b/Obvs.Serialization.MessagePack/MsgPackMessageSerializer.cs index 04c6cbdb..d678bd6d 100644 --- a/Obvs.Serialization.MessagePack/MsgPackMessageSerializer.cs +++ b/Obvs.Serialization.MessagePack/MsgPackMessageSerializer.cs @@ -1,8 +1,10 @@ -using System.IO; +using System; +using System.IO; using MsgPack.Serialization; namespace Obvs.Serialization.MessagePack { + [Obsolete("MsgPack.Cli is no longer maintained; use Obvs.Serialization.MessagePack-CSharp instead")] public class MsgPackMessageSerializer : IMessageSerializer { public void Serialize(Stream destination, object message) diff --git a/Obvs.Serialization.MessagePack/Obvs.Serialization.MessagePack.csproj b/Obvs.Serialization.MessagePack/Obvs.Serialization.MessagePack.csproj index 8d147a57..96631365 100644 --- a/Obvs.Serialization.MessagePack/Obvs.Serialization.MessagePack.csproj +++ b/Obvs.Serialization.MessagePack/Obvs.Serialization.MessagePack.csproj @@ -7,8 +7,8 @@ 1.0.0 http://christopherread.github.io/Obvs - Minor dependency update - MessagePack serialization for Obvs + Deprecated: MsgPack.Cli is no longer maintained. Use Obvs.Serialization.MessagePack-CSharp instead. + Deprecated: use Obvs.Serialization.MessagePack-CSharp. MessagePack serialization for Obvs using MsgPack.Cli, which is no longer maintained. diff --git a/Obvs.Serialization.Tests/TestMsgPackSerialization.cs b/Obvs.Serialization.Tests/TestMsgPackSerialization.cs index 1e8cb804..77516b22 100644 --- a/Obvs.Serialization.Tests/TestMsgPackSerialization.cs +++ b/Obvs.Serialization.Tests/TestMsgPackSerialization.cs @@ -1,4 +1,8 @@ +// These tests exercise the deprecated MsgPack.Cli serializer on purpose +#pragma warning disable CS0618 + using System; +using System.Reflection; using FakeItEasy; using Xunit; using Obvs.Configuration; @@ -11,6 +15,30 @@ namespace Obvs.Serialization.Tests public class TestMsgPackSerialization { + [Theory] + [InlineData(typeof(MsgPackMessageSerializer))] + [InlineData(typeof(MsgPackMessageDeserializer<>))] + [InlineData(typeof(MsgPackMessageDeserializerFactory))] + public void ShouldBeObsoleteInFavourOfMessagePackCSharp(Type type) + { + var obsolete = type.GetCustomAttribute(); + + Assert.NotNull(obsolete); + Assert.False(obsolete.IsError); + Assert.Contains("Obvs.Serialization.MessagePack-CSharp", obsolete.Message); + } + + [Fact] + public void FluentConfigShouldBeObsoleteInFavourOfMessagePackCSharp() + { + // Obvs.Serialization.MessagePack-CSharp has a class with the same full name, so look it up in this package's assembly + var extensions = typeof(MsgPackMessageSerializer).Assembly.GetType("Obvs.Serialization.MessagePack.Configuration.FluentConfigExtensions"); + var obsolete = extensions.GetMethod("SerializedAsMsgPack").GetCustomAttribute(); + + Assert.NotNull(obsolete); + Assert.Contains("SerializedAsMessagePackCSharp", obsolete.Message); + } + [Fact] public void ShouldSerializeToMsgPack() { diff --git a/README.md b/README.md index 7b37ded7..01410af4 100644 --- a/README.md +++ b/README.md @@ -38,7 +38,7 @@ ### Extensions * Transports: ActiveMQ / RabbitMQ / NetMQ / AzureServiceBus / Kafka / EventStore -* Serialization: XML / JSON.Net / NetJson / ProtoBuf / MsgPack +* Serialization: XML / JSON.Net / NetJson / ProtoBuf / MessagePack (via MessagePack-CSharp; the MsgPack.Cli-based package is deprecated) * Logging: NLog / log4net * Monitoring: Performance Counters / ElasticSearch