124 lines
3.6 KiB
C#
124 lines
3.6 KiB
C#
namespace ASC.EventBus.RabbitMQ;
|
|
|
|
public class DefaultRabbitMQPersistentConnection
|
|
: IRabbitMQPersistentConnection
|
|
{
|
|
private readonly IConnectionFactory _connectionFactory;
|
|
private readonly ILog _logger;
|
|
private readonly int _retryCount;
|
|
private IConnection _connection;
|
|
private bool _disposed;
|
|
|
|
object sync_root = new object();
|
|
|
|
public DefaultRabbitMQPersistentConnection(IConnectionFactory connectionFactory, IOptionsMonitor<ILog> options, int retryCount = 5)
|
|
{
|
|
_connectionFactory = connectionFactory ?? throw new ArgumentNullException(nameof(connectionFactory));
|
|
_logger = options.CurrentValue ?? throw new ArgumentNullException(nameof(options.CurrentValue));
|
|
_retryCount = retryCount;
|
|
}
|
|
|
|
public bool IsConnected
|
|
{
|
|
get
|
|
{
|
|
return _connection != null && _connection.IsOpen && !_disposed;
|
|
}
|
|
}
|
|
|
|
public IModel CreateModel()
|
|
{
|
|
if (!IsConnected)
|
|
{
|
|
throw new InvalidOperationException("No RabbitMQ connections are available to perform this action");
|
|
}
|
|
|
|
return _connection.CreateModel();
|
|
}
|
|
|
|
public void Dispose()
|
|
{
|
|
if (_disposed) return;
|
|
|
|
_disposed = true;
|
|
|
|
try
|
|
{
|
|
_connection.ConnectionShutdown -= OnConnectionShutdown;
|
|
_connection.CallbackException -= OnCallbackException;
|
|
_connection.ConnectionBlocked -= OnConnectionBlocked;
|
|
_connection.Dispose();
|
|
}
|
|
catch (IOException ex)
|
|
{
|
|
_logger.Fatal(ex.ToString());
|
|
}
|
|
}
|
|
|
|
public bool TryConnect()
|
|
{
|
|
_logger.Info("RabbitMQ Client is trying to connect");
|
|
|
|
lock (sync_root)
|
|
{
|
|
var policy = RetryPolicy.Handle<SocketException>()
|
|
.Or<BrokerUnreachableException>()
|
|
.WaitAndRetry(_retryCount, retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)), (ex, time) =>
|
|
{
|
|
_logger.Warn(String.Format("RabbitMQ Client could not connect after {TimeOut}s ({ExceptionMessage})", $"{time.TotalSeconds:n1}", ex.Message), ex);
|
|
}
|
|
);
|
|
|
|
policy.Execute(() =>
|
|
{
|
|
_connection = _connectionFactory
|
|
.CreateConnection();
|
|
});
|
|
|
|
if (IsConnected)
|
|
{
|
|
_connection.ConnectionShutdown += OnConnectionShutdown;
|
|
_connection.CallbackException += OnCallbackException;
|
|
_connection.ConnectionBlocked += OnConnectionBlocked;
|
|
|
|
_logger.InfoFormat("RabbitMQ Client acquired a persistent connection to '{HostName}' and is subscribed to failure events", _connection.Endpoint.HostName);
|
|
|
|
return true;
|
|
}
|
|
else
|
|
{
|
|
_logger.Fatal("FATAL ERROR: RabbitMQ connections could not be created and opened");
|
|
|
|
return false;
|
|
}
|
|
}
|
|
}
|
|
|
|
private void OnConnectionBlocked(object? sender, ConnectionBlockedEventArgs e)
|
|
{
|
|
if (_disposed) return;
|
|
|
|
_logger.Warn("A RabbitMQ connection is shutdown. Trying to re-connect...");
|
|
|
|
TryConnect();
|
|
}
|
|
|
|
void OnCallbackException(object? sender, CallbackExceptionEventArgs e)
|
|
{
|
|
if (_disposed) return;
|
|
|
|
_logger.Warn("A RabbitMQ connection throw exception. Trying to re-connect...");
|
|
|
|
TryConnect();
|
|
}
|
|
|
|
void OnConnectionShutdown(object? sender, ShutdownEventArgs reason)
|
|
{
|
|
if (_disposed) return;
|
|
|
|
_logger.Warn("A RabbitMQ connection is on shutdown. Trying to re-connect...");
|
|
|
|
TryConnect();
|
|
}
|
|
}
|