-
Notifications
You must be signed in to change notification settings - Fork 15
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* add NBB.Messaging.JetStream
- Loading branch information
Showing
8 changed files
with
627 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
24 changes: 24 additions & 0 deletions
24
src/Messaging/NBB.Messaging.JetStream/DependencyInjectionExtensions.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,24 @@ | ||
| // Copyright (c) TotalSoft. | ||
| // This source code is licensed under the MIT license. | ||
|
|
||
| using Microsoft.Extensions.Configuration; | ||
| using NBB.Messaging.Abstractions; | ||
| using NBB.Messaging.JetStream; | ||
| using NBB.Messaging.JetStream.Internal; | ||
|
|
||
| // ReSharper disable once CheckNamespace | ||
| namespace Microsoft.Extensions.DependencyInjection | ||
| { | ||
| public static class DependencyInjectionExtensions | ||
| { | ||
| public static IServiceCollection AddJetStreamTransport(this IServiceCollection services, IConfiguration configuration) | ||
| { | ||
| services.Configure<JetStreamOptions>(configuration.GetSection("Messaging").GetSection("JetStream")); | ||
| services.AddSingleton<JetStreamConnectionProvider>(); | ||
| services.AddSingleton<IMessagingTransport, JetStreamMessagingTransport>(); | ||
| services.AddSingleton<ITransportMonitor>(sp => sp.GetRequiredService<JetStreamConnectionProvider>()); | ||
|
|
||
| return services; | ||
| } | ||
| } | ||
| } |
104 changes: 104 additions & 0 deletions
104
src/Messaging/NBB.Messaging.JetStream/Internal/JetStreamConnectionProvider.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,104 @@ | ||
| // Copyright (c) TotalSoft. | ||
| // This source code is licensed under the MIT license. | ||
|
|
||
| using Microsoft.Extensions.Logging; | ||
| using Microsoft.Extensions.Options; | ||
| using NATS.Client; | ||
| using NBB.Messaging.Abstractions; | ||
| using System; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
|
|
||
| namespace NBB.Messaging.JetStream.Internal | ||
| { | ||
| public class JetStreamConnectionProvider : IDisposable, ITransportMonitor | ||
| { | ||
| private readonly IOptions<JetStreamOptions> _natsOptions; | ||
| private readonly ILogger<JetStreamConnectionProvider> _logger; | ||
| private IConnection _connection; | ||
| private static readonly object InstanceLoker = new(); | ||
| private Exception _unrecoverableException; | ||
|
|
||
| public event TransportErrorHandler OnError; | ||
|
|
||
| public JetStreamConnectionProvider(IOptions<JetStreamOptions> natsOptions, ILogger<JetStreamConnectionProvider> logger) | ||
| { | ||
| _natsOptions = natsOptions; | ||
| _logger = logger; | ||
| } | ||
|
|
||
| public async Task ExecuteAsync(Func<IConnection, Task> action) | ||
| { | ||
| var connection = GetAndCheckConnection(); | ||
|
|
||
| await action(connection); | ||
| } | ||
|
|
||
| public void Execute(Action<IConnection> action) | ||
| { | ||
| var connection = GetAndCheckConnection(); | ||
|
|
||
| action(connection); | ||
| } | ||
|
|
||
| private IConnection GetAndCheckConnection() | ||
| { | ||
| if (_connection == null) | ||
| lock (InstanceLoker) | ||
| { | ||
| if (_connection == null) | ||
| _connection = CreateConnection(); | ||
| } | ||
| return _connection; | ||
| } | ||
|
|
||
| private IConnection CreateConnection() | ||
| { | ||
| var options = ConnectionFactory.GetDefaultOptions(); | ||
| options.Url = _natsOptions.Value.NatsUrl; | ||
|
|
||
| //https://github.com/nats-io/nats.net/issues/804 | ||
| options.AllowReconnect = false; | ||
|
|
||
| options.ClosedEventHandler += (_, args) => | ||
| { | ||
| SetConnectionLostState(args.Error ?? new Exception("NATS Jetstream connection was lost")); | ||
| }; | ||
|
|
||
| _connection = new ConnectionFactory().CreateConnection(options); | ||
| _logger.LogInformation($"NATS Jetstream connection to {_natsOptions.Value.NatsUrl} was established"); | ||
|
|
||
| return _connection; | ||
| } | ||
|
|
||
| private void SetConnectionLostState(Exception exception) | ||
| { | ||
| _connection = null; | ||
|
|
||
| // Set the field to the current exception if not already set | ||
| var existingException = Interlocked.CompareExchange(ref _unrecoverableException, exception, null); | ||
|
|
||
| // Send the application stop signal only once | ||
| if (existingException != null) | ||
| return; | ||
|
|
||
| _logger.LogError(exception, "NATS Jetstream connection unrecoverable"); | ||
|
|
||
| OnError?.Invoke(exception); | ||
| } | ||
|
|
||
| public void Dispose() | ||
| { | ||
| Dispose(true); | ||
| GC.SuppressFinalize(this); | ||
| } | ||
|
|
||
| protected virtual void Dispose(bool disposing) | ||
| { | ||
| if (disposing) | ||
| { | ||
| _connection?.Dispose(); | ||
| } | ||
| } | ||
| } | ||
| } |
97 changes: 97 additions & 0 deletions
97
src/Messaging/NBB.Messaging.JetStream/JetStreamMessagingTransport.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,97 @@ | ||
| // Copyright (c) TotalSoft. | ||
| // This source code is licensed under the MIT license. | ||
|
|
||
| using Microsoft.Extensions.Options; | ||
| using NATS.Client; | ||
| using NATS.Client.JetStream; | ||
| using NBB.Messaging.Abstractions; | ||
| using NBB.Messaging.JetStream.Internal; | ||
| using System; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
|
|
||
| namespace NBB.Messaging.JetStream | ||
| { | ||
| public class JetStreamMessagingTransport : IMessagingTransport | ||
| { | ||
| private readonly IOptions<JetStreamOptions> _natsOptions; | ||
| private readonly JetStreamConnectionProvider _natsConnectionManager; | ||
|
|
||
| public JetStreamMessagingTransport(IOptions<JetStreamOptions> natsOptions, JetStreamConnectionProvider natsConnectionManager) | ||
| { | ||
| _natsOptions = natsOptions; | ||
| _natsConnectionManager = natsConnectionManager; | ||
| } | ||
|
|
||
| public Task PublishAsync(string topic, TransportSendContext sendContext, CancellationToken cancellationToken = default) | ||
| { | ||
| var envelopeData = sendContext.EnvelopeBytesAccessor.Invoke(); | ||
|
|
||
| return _natsConnectionManager.ExecuteAsync(con => | ||
| { | ||
| IJetStream js = con.CreateJetStreamContext(); | ||
| return js.PublishAsync(topic, envelopeData); | ||
| }); | ||
| } | ||
|
|
||
| public Task<IDisposable> SubscribeAsync(string topic, | ||
| Func<TransportReceiveContext, Task> handler, | ||
| SubscriptionTransportOptions options = null, | ||
| CancellationToken cancellationToken = default) | ||
| { | ||
|
|
||
| IDisposable consumer = null; | ||
|
|
||
| _natsConnectionManager.Execute(con => | ||
| { | ||
| IJetStream js = con.CreateJetStreamContext(); | ||
|
|
||
| // set's up the stream | ||
| var isCommand = topic.ToLower().Contains("commands."); | ||
|
|
||
| var stream = isCommand ? _natsOptions.Value.CommandsStream : _natsOptions.Value.EventsStream; | ||
| var jsm = con.CreateJetStreamManagementContext(); | ||
| jsm.GetStreamInfo(stream); | ||
|
|
||
| // get stream context, create consumer and get the consumer context | ||
| var streamContext = con.GetStreamContext(stream); | ||
|
|
||
| var subscriberOptions = options ?? SubscriptionTransportOptions.Default; | ||
| var ccb = ConsumerConfiguration.Builder(); | ||
|
|
||
| if (subscriberOptions.IsDurable) | ||
| { | ||
| var clientId = (_natsOptions.Value.ClientId + topic).Replace(".", "_"); | ||
| ccb.WithDurable(clientId); | ||
| } | ||
|
|
||
| if (subscriberOptions.DeliverNewMessagesOnly) | ||
| ccb.WithDeliverPolicy(DeliverPolicy.New); | ||
| else | ||
| ccb.WithDeliverPolicy(DeliverPolicy.All); | ||
|
|
||
| ccb.WithAckWait(subscriberOptions.AckWait ?? _natsOptions.Value.AckWait ?? 50000); | ||
|
|
||
| //https://docs.nats.io/nats-concepts/jetstream/consumers#maxackpending | ||
| ccb.WithMaxAckPending(subscriberOptions.MaxConcurrentMessages); | ||
| ccb.WithFilterSubject(topic); | ||
|
|
||
| var consumerContext = streamContext.CreateOrUpdateConsumer(ccb.Build()); | ||
|
|
||
| void NatsMsgHandler(object obj, MsgHandlerEventArgs args) | ||
| { | ||
| if (cancellationToken.IsCancellationRequested) | ||
| return; | ||
|
|
||
| var receiveContext = new TransportReceiveContext(new TransportReceivedData.EnvelopeBytes(args.Message.Data)); | ||
|
|
||
| // Fire and forget | ||
| _ = handler(receiveContext).ContinueWith(_ => args.Message.Ack(), cancellationToken); | ||
| } | ||
| consumer = consumerContext.Consume(NatsMsgHandler); | ||
|
|
||
| }); | ||
| return Task.FromResult(consumer); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,26 @@ | ||
| // Copyright (c) TotalSoft. | ||
| // This source code is licensed under the MIT license. | ||
|
|
||
| namespace NBB.Messaging.JetStream | ||
| { | ||
| public class JetStreamOptions | ||
| { | ||
| /// <summary> | ||
| /// URL of the Streaming NATS cluster | ||
| /// </summary> | ||
| public string NatsUrl { get; set; } | ||
|
|
||
| /// <summary> | ||
| /// Identifier of the Streaming NATS client | ||
| /// </summary> | ||
| public string ClientId { get; set; } | ||
|
|
||
| /// <summary> | ||
| /// The time the server awaits for acknowledgement from the client before redelivering the message (in milliseconds) | ||
| /// </summary> | ||
| public int? AckWait { get; set; } | ||
| public string CommandsStream { get; set; } | ||
| public string EventsStream { get; set; } | ||
|
|
||
| } | ||
| } |
Oops, something went wrong.