-
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.
Merge branch 'master' of https://github.com/osstotalsoft/nbb
- Loading branch information
Showing
5 changed files
with
175 additions
and
80 deletions.
There are no files selected for viewing
146 changes: 146 additions & 0 deletions
146
src/Messaging/NBB.Messaging.Nats/Internal/StanConnectionManager.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,146 @@ | ||
| using System; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
| using Microsoft.Extensions.Configuration; | ||
| using Microsoft.Extensions.Hosting; | ||
| using Microsoft.Extensions.Logging; | ||
| using NATS.Client; | ||
| using STAN.Client; | ||
|
|
||
| namespace NBB.Messaging.Nats.Internal | ||
| { | ||
| public class StanConnectionProvider : IDisposable | ||
| { | ||
| private readonly IConfiguration _configuration; | ||
| private readonly ILogger<StanConnectionProvider> _logger; | ||
| private readonly IApplicationLifetime _applicationLifetime; | ||
| private IStanConnection _connection; | ||
| private Exception _unrecoverableException; | ||
| private readonly Lazy<IStanConnection> _lazyConnection; | ||
|
|
||
| public StanConnectionProvider(IConfiguration configuration, ILogger<StanConnectionProvider> logger, | ||
| IApplicationLifetime applicationLifetime) | ||
| { | ||
| _configuration = configuration; | ||
| _logger = logger; | ||
| _applicationLifetime = applicationLifetime; | ||
| _lazyConnection = new Lazy<IStanConnection>(GetConnection); | ||
| } | ||
|
|
||
| public async Task ExecuteAsync(Func<IStanConnection, Task> action) | ||
| { | ||
| var connection = GetAndCheckConnection(); | ||
| try | ||
| { | ||
| await action(connection); | ||
| } | ||
| catch (Exception ex) | ||
| when (IsUnrecoverableException(ex)) | ||
| { | ||
| SetUnrecoverableState(ex); | ||
| throw; | ||
| } | ||
| } | ||
|
|
||
| public void Execute(Action<IStanConnection> action) | ||
| { | ||
| var connection = GetAndCheckConnection(); | ||
| try | ||
| { | ||
| action(connection); | ||
| } | ||
| catch (Exception ex) | ||
| when (IsUnrecoverableException(ex)) | ||
| { | ||
| SetUnrecoverableState(ex); | ||
| throw; | ||
| } | ||
| } | ||
|
|
||
| private IStanConnection GetAndCheckConnection() | ||
| { | ||
| ThrowIfUnrecoverableState(); | ||
| try | ||
| { | ||
| // Exception from the lazy factory method is cached | ||
| return _lazyConnection.Value; | ||
| } | ||
| catch (Exception ex) | ||
| { | ||
| SetUnrecoverableState(ex); | ||
| throw; | ||
| } | ||
| } | ||
|
|
||
| private IStanConnection GetConnection() | ||
| { | ||
| var natsUrl = _configuration.GetSection("Messaging").GetSection("Nats")["natsUrl"]; | ||
| var cluster = _configuration.GetSection("Messaging").GetSection("Nats")["cluster"]; | ||
| var clientId = _configuration.GetSection("Messaging").GetSection("Nats")["clientId"] | ||
| ?.Replace(".", "_"); | ||
| var options = StanOptions.GetDefaultOptions(); | ||
| options.NatsURL = natsUrl; | ||
|
|
||
| options.ConnectionLostEventHandler = (obj, args) => | ||
| { | ||
| SetUnrecoverableState(args.ConnectionException ?? new Exception("NATS connection was lost")); | ||
| }; | ||
|
|
||
| //fix https://github.com/nats-io/csharp-nats-streaming/issues/28 | ||
| options.PubAckWait = 30000; | ||
|
|
||
| var cf = new StanConnectionFactory(); | ||
| _connection = cf.CreateConnection(cluster, clientId + Guid.NewGuid(), options); | ||
|
|
||
| return _connection; | ||
| } | ||
|
|
||
| private void SetUnrecoverableState(Exception exception) | ||
| { | ||
| // 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.LogCritical(exception, "NATS connection unrecoverable"); | ||
| _applicationLifetime.StopApplication(); | ||
|
|
||
| Dispose(); | ||
| } | ||
|
|
||
| private void ThrowIfUnrecoverableState() | ||
| { | ||
| // For consistency, read the field using the same primitive used for writing instead of using Thread.VolatileRead | ||
| var exception = Interlocked.CompareExchange(ref _unrecoverableException, null, null); | ||
| if (exception != null) | ||
| { | ||
| throw new Exception("NATS connection encountered an unrecoverable exception", exception); | ||
| } | ||
| } | ||
|
|
||
| private static bool IsUnrecoverableException(Exception ex) | ||
| { | ||
| return | ||
| ex is NATSConnectionClosedException || | ||
| ex is StanConnectionClosedException || | ||
| ex is NATSConnectionException || | ||
| ex is StanConnectionException || | ||
| ex is NATSBadSubscriptionException || | ||
| ex is StanBadSubscriptionException || | ||
| ex is StanTimeoutException || | ||
| ex is NATSTimeoutException || | ||
| ex is NATSStaleConnectionException || | ||
| ex is NATSNoServersException || | ||
| ex is StanConnectRequestException || | ||
| ex is StanMaxPingsException; | ||
| } | ||
|
|
||
| public void Dispose() | ||
| { | ||
| _connection?.Dispose(); | ||
| } | ||
| } | ||
| } |
49 changes: 0 additions & 49 deletions
49
src/Messaging/NBB.Messaging.Nats/Internal/StanConnectionProvider.cs
This file was deleted.
Oops, something went wrong.
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
23 changes: 12 additions & 11 deletions
23
src/Messaging/NBB.Messaging.Nats/NatsMessagingTopicPublisher.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 |
|---|---|---|
| @@ -1,34 +1,35 @@ | ||
| using System; | ||
| using System.Diagnostics; | ||
| using System.Diagnostics; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
| using Microsoft.Extensions.Logging; | ||
| using NBB.Messaging.Abstractions; | ||
| using NBB.Messaging.DataContracts; | ||
| using NBB.Messaging.Nats.Internal; | ||
|
|
||
| namespace NBB.Messaging.Nats | ||
| { | ||
| public class NatsMessagingTopicPublisher : IMessagingTopicPublisher | ||
| { | ||
| private readonly StanConnectionProvider _stanConnectionProvider; | ||
| private readonly StanConnectionProvider _stanConnectionManager; | ||
| private readonly ILogger<NatsMessagingTopicPublisher> _logger; | ||
|
|
||
| public NatsMessagingTopicPublisher(StanConnectionProvider stanConnectionProvider, ILogger<NatsMessagingTopicPublisher> logger) | ||
| public NatsMessagingTopicPublisher(StanConnectionProvider stanConnectionManager, | ||
| ILogger<NatsMessagingTopicPublisher> logger) | ||
| { | ||
| _stanConnectionProvider = stanConnectionProvider; | ||
| _stanConnectionManager = stanConnectionManager; | ||
| _logger = logger; | ||
| } | ||
|
|
||
| public async Task PublishAsync(string topic, string key, string message, CancellationToken cancellationToken = default(CancellationToken)) | ||
| { | ||
| public async Task PublishAsync(string topic, string key, string message, | ||
| CancellationToken cancellationToken = default(CancellationToken)) | ||
| { | ||
| var stopWatch = new Stopwatch(); | ||
| stopWatch.Start(); | ||
| var connection = _stanConnectionProvider.GetConnection(); | ||
| var result = await connection.PublishAsync(topic, System.Text.Encoding.UTF8.GetBytes(message)); | ||
| await _stanConnectionManager.ExecuteAsync(async connection => | ||
| await connection.PublishAsync(topic, System.Text.Encoding.UTF8.GetBytes(message))); | ||
| stopWatch.Stop(); | ||
|
|
||
| _logger.LogDebug("Nats message published to subject {Subject} in {ElapsedMilliseconds} ms", topic, stopWatch.ElapsedMilliseconds); | ||
| _logger.LogDebug("Nats message published to subject {Subject} in {ElapsedMilliseconds} ms", topic, | ||
| stopWatch.ElapsedMilliseconds); | ||
| } | ||
| } | ||
| } |
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