From 65d50eab2b82af6335cf329f7af67638e9223496 Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sun, 25 Oct 2026 08:11:18 -0700 Subject: [PATCH 1/7] Run broker-backed transport tests in CI (#82) Fixes #74 Co-Authored-By: Claude Opus 5.5 --- .github/workflows/dotnet.yml | 50 +++++++++++++++++++ Obvs.Kafka.Tests/TestBroker.cs | 10 ++++ Obvs.Kafka.Tests/TestMessageSendAndReceive.cs | 9 +--- Obvs.NATS.Tests/TestBroker.cs | 10 ++++ Obvs.NATS.Tests/TestServiceBus.cs | 2 +- Obvs.NATS.Tests/TestSourceAndPublisher.cs | 2 +- 6 files changed, 74 insertions(+), 9 deletions(-) create mode 100644 Obvs.Kafka.Tests/TestBroker.cs create mode 100644 Obvs.NATS.Tests/TestBroker.cs diff --git a/.github/workflows/dotnet.yml b/.github/workflows/dotnet.yml index 18ee687..76a115e 100644 --- a/.github/workflows/dotnet.yml +++ b/.github/workflows/dotnet.yml @@ -17,6 +17,41 @@ 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 + # Assign partitions to the test's new consumer group straight away + KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 + 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 +75,21 @@ 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: Prepare Kafka + # The test consumer starts at the latest offset and doesn't wait for partition assignment, so create its + # topic and warm up the group coordinator (which creates __consumer_offsets) before the tests run + 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 + docker exec ${{ job.services.kafka.id }} /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic-3 --group ci-warmup --timeout-ms 10000 || true + - 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/Obvs.Kafka.Tests/TestBroker.cs b/Obvs.Kafka.Tests/TestBroker.cs new file mode 100644 index 0000000..674d5b3 --- /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 714639e..1d24e30 100644 --- a/Obvs.Kafka.Tests/TestMessageSendAndReceive.cs +++ b/Obvs.Kafka.Tests/TestMessageSendAndReceive.cs @@ -44,13 +44,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"; diff --git a/Obvs.NATS.Tests/TestBroker.cs b/Obvs.NATS.Tests/TestBroker.cs new file mode 100644 index 0000000..e5cf445 --- /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 be60588..805eb11 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 => diff --git a/Obvs.NATS.Tests/TestSourceAndPublisher.cs b/Obvs.NATS.Tests/TestSourceAndPublisher.cs index ea880a1..0ed1521 100644 --- a/Obvs.NATS.Tests/TestSourceAndPublisher.cs +++ b/Obvs.NATS.Tests/TestSourceAndPublisher.cs @@ -19,7 +19,7 @@ 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 + return cf.CreateConnection(TestBroker.Url); }, LazyThreadSafetyMode.ExecutionAndPublication); const string subjectPrefix = "Obvs.NATS.Tests"; From f8e920ae7e8beac5b438942325fafa261515ea95 Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sun, 25 Oct 2026 08:11:18 -0700 Subject: [PATCH 2/7] Use the official ActiveMQ Classic image in the examples (#83) Fixes #76 Co-Authored-By: Claude Opus 5.5 --- Examples/docker-compose.yml | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/Examples/docker-compose.yml b/Examples/docker-compose.yml index 3cce701..5e69a09 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 From d261d482d468f773b29a1911cae1bac0e46773c9 Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sun, 25 Oct 2026 08:11:18 -0700 Subject: [PATCH 3/7] Deprecate Obvs.Serialization.MessagePack (#96) Fixes #79 Co-Authored-By: Claude Opus 5.5 --- .../Configuration/FluentConfigExtensions.cs | 4 ++- .../MsgPackMessageDeserializer.cs | 2 ++ .../MsgPackMessageDeserializerFactory.cs | 1 + .../MsgPackMessageSerializer.cs | 4 ++- .../Obvs.Serialization.MessagePack.csproj | 4 +-- .../TestMsgPackSerialization.cs | 28 +++++++++++++++++++ README.md | 2 +- 7 files changed, 40 insertions(+), 5 deletions(-) diff --git a/Obvs.Serialization.MessagePack/Configuration/FluentConfigExtensions.cs b/Obvs.Serialization.MessagePack/Configuration/FluentConfigExtensions.cs index 4f2aeb8..5e0d06c 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 0c27685..12bd8c9 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 0d9fe04..f25cde9 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 04c6cbd..d678bd6 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 8d147a5..9663136 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 1e8cb80..77516b2 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 7b37ded..01410af 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 From bf99a81f34bb4d7ef251fc34328a129f13a583ea Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sun, 25 Oct 2026 08:11:18 -0700 Subject: [PATCH 4/7] Port Obvs.Monitoring.ElasticSearch from NEST to Elastic.Clients.Elasticsearch (#98) Fixes #77 Co-Authored-By: Claude Opus 5.5 --- Directory.Packages.props | 2 +- .../ElasticSearchMonitoringConfig.cs | 12 +-- .../ElasticSearchMonitor.cs | 10 +-- .../ElasticSearchMonitorFactory.cs | 6 +- .../Obvs.Monitoring.ElasticSearch.csproj | 2 +- Obvs.Monitoring.ElasticSearch/ObvsCounter.cs | 1 - .../TestElasticSearchMonitoring.cs | 90 ++++++++++++++----- 7 files changed, 85 insertions(+), 38 deletions(-) diff --git a/Directory.Packages.props b/Directory.Packages.props index 1398065..c295a08 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -11,7 +11,7 @@ - + diff --git a/Obvs.Monitoring.ElasticSearch/Configuration/ElasticSearchMonitoringConfig.cs b/Obvs.Monitoring.ElasticSearch/Configuration/ElasticSearchMonitoringConfig.cs index 9c25cab..3c17c37 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 5d2d18d..b94da2f 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 cd61666..ffc09d8 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 cc9be8a..ab011d8 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 88cc7f3..0996de6 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 3835d8b..31231d8 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"); From db6747eaff6564bbc86fac35b9625f5aa1837276 Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sun, 25 Oct 2026 08:12:05 -0700 Subject: [PATCH 5/7] Port Obvs.NATS from NATS.Client to NATS.Net (#99) Fixes #78 Co-Authored-By: Claude Opus 5.5 --- Directory.Packages.props | 4 +- Obvs.NATS.Tests/TestServiceBus.cs | 9 ++- Obvs.NATS.Tests/TestSourceAndPublisher.cs | 19 ++--- Obvs.NATS/AsyncSubscriptionExtensions.cs | 26 ------ .../NatsServiceEndpointProvider.cs | 24 ++---- .../Configuration/SharedConnectionFactory.cs | 22 +++--- Obvs.NATS/MessagePublisher.cs | 13 ++- Obvs.NATS/MessageSource.cs | 79 ++++++++++++++----- Obvs.NATS/Obvs.NATS.csproj | 4 +- 9 files changed, 104 insertions(+), 96 deletions(-) delete mode 100644 Obvs.NATS/AsyncSubscriptionExtensions.cs diff --git a/Directory.Packages.props b/Directory.Packages.props index c295a08..f75f379 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -7,11 +7,11 @@ + - - + diff --git a/Obvs.NATS.Tests/TestServiceBus.cs b/Obvs.NATS.Tests/TestServiceBus.cs index 805eb11..c86af93 100644 --- a/Obvs.NATS.Tests/TestServiceBus.cs +++ b/Obvs.NATS.Tests/TestServiceBus.cs @@ -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 0ed1521..256937b 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(TestBroker.Url); - }, 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 45f475f..0000000 --- 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 09de932..239952d 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 fe23658..0fbeb19 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 ec06f35..4470cdc 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 ed65f18..64f4da1 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 7691021..a5ad1c4 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 - + From 61612e774630597dce814c991472fcc1cf2e295a Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sun, 25 Oct 2026 08:13:47 -0700 Subject: [PATCH 6/7] Test that a Kafka message published straight after subscribing is received Also drop the CI broker warm-up that hid the problem; only the topic is created up front now. Expected to fail until #103 is fixed. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/dotnet.yml | 10 ++------ Obvs.Kafka.Tests/TestMessageSendAndReceive.cs | 24 +++++++++++++++++++ 2 files changed, 26 insertions(+), 8 deletions(-) diff --git a/.github/workflows/dotnet.yml b/.github/workflows/dotnet.yml index 76a115e..a844f32 100644 --- a/.github/workflows/dotnet.yml +++ b/.github/workflows/dotnet.yml @@ -46,8 +46,6 @@ jobs: KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 KAFKA_SHARE_COORDINATOR_STATE_TOPIC_REPLICATION_FACTOR: 1 KAFKA_SHARE_COORDINATOR_STATE_TOPIC_MIN_ISR: 1 - # Assign partitions to the test's new consumer group straight away - KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 options: >- --health-cmd "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092" --health-interval 5s --health-timeout 10s --health-retries 20 @@ -75,12 +73,8 @@ 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: Prepare Kafka - # The test consumer starts at the latest offset and doesn't wait for partition assignment, so create its - # topic and warm up the group coordinator (which creates __consumer_offsets) before the tests run - 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 - docker exec ${{ job.services.kafka.id }} /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic-3 --group ci-warmup --timeout-ms 10000 || 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 diff --git a/Obvs.Kafka.Tests/TestMessageSendAndReceive.cs b/Obvs.Kafka.Tests/TestMessageSendAndReceive.cs index 1d24e30..f62b1af 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; @@ -95,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 From 78e360486f9470d19698e26e67f6b81265085b1d Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sun, 25 Oct 2026 08:18:54 -0700 Subject: [PATCH 7/7] Kafka: don't miss messages published straight after subscribing Subscribe returned as soon as the consumer called Subscribe, before Kafka assigned partitions, and "latest" was only resolved later, so messages published in between were skipped. Now Subscribe waits (up to 30s) for partition assignment, and the assignment handler fixes each partition's start offset there and then: the group's committed offset, or else the current end of the log. Fixes #103 Co-Authored-By: Claude Opus 5.5 --- Obvs.Kafka/MessageSource.cs | 25 +++++++++++++++++++++++-- 1 file changed, 23 insertions(+), 2 deletions(-) diff --git a/Obvs.Kafka/MessageSource.cs b/Obvs.Kafka/MessageSource.cs index 2a76fd0..405b85e 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