diff --git a/.github/workflows/dotnet.yml b/.github/workflows/dotnet.yml
index 4ac797ec..b95a8e17 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
run: dotnet test --no-build --configuration Release --verbosity normal --filter Category!=Explicit
- name: Pack
diff --git a/Directory.Packages.props b/Directory.Packages.props
index ec3fc538..8cb50885 100644
--- a/Directory.Packages.props
+++ b/Directory.Packages.props
@@ -10,7 +10,7 @@
-
+
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 39243007..b6061d75 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 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 e7bb5cad..d13178c0 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));
((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 1838bf6b..a4ad065a 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));
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
-
+