2022-04-14 19:23:57 +00:00
// (c) Copyright Ascensio System SIA 2010-2022
//
// This program is a free software product.
// You can redistribute it and/or modify it under the terms
// of the GNU Affero General Public License (AGPL) version 3 as published by the Free Software
// Foundation. In accordance with Section 7(a) of the GNU AGPL its Section 15 shall be amended
// to the effect that Ascensio System SIA expressly excludes the warranty of non-infringement of
// any third-party rights.
//
// This program is distributed WITHOUT ANY WARRANTY, without even the implied warranty
// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. For details, see
// the GNU AGPL at: http://www.gnu.org/licenses/agpl-3.0.html
//
// You can contact Ascensio System SIA at Lubanas st. 125a-25, Riga, Latvia, EU, LV-1021.
//
// The interactive user interfaces in modified source and object code versions of the Program must
// display Appropriate Legal Notices, as required under Section 5 of the GNU AGPL version 3.
//
// Pursuant to Section 7(b) of the License you must retain the original Product logo when
// distributing the program. Pursuant to Section 7(e) we decline to grant you any rights under
// trademark law for use of our trademarks.
//
// All the Product's GUI elements, including illustrations and icon sets, as well as technical writing
// content are licensed under the terms of the Creative Commons Attribution-ShareAlike 4.0
// International. See the License terms at http://creativecommons.org/licenses/by-sa/4.0/legalcode
namespace ASC.EventBus.RabbitMQ ;
2022-02-10 14:42:21 +00:00
public class DefaultRabbitMQPersistentConnection
: IRabbitMQPersistentConnection
{
private readonly IConnectionFactory _connectionFactory ;
private readonly ILog _logger ;
private readonly int _retryCount ;
private IConnection _connection ;
2022-04-14 19:23:57 +00:00
private bool _disposed ;
readonly object sync_root = new object ( ) ;
2022-02-10 14:42:21 +00:00
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 ( )
{
2022-04-14 19:23:57 +00:00
if ( _disposed )
{
return ;
}
2022-02-10 14:42:21 +00:00
_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 ) = >
{
2022-03-04 15:12:48 +00:00
_logger . WarnFormat ( "RabbitMQ Client could not connect after {TimeOut}s ({ExceptionMessage})" , $"{time.TotalSeconds:n1}" , ex . Message ) ;
2022-02-10 14:42:21 +00:00
}
) ;
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 ;
}
}
}
2022-02-14 11:18:58 +00:00
private void OnConnectionBlocked ( object sender , ConnectionBlockedEventArgs e )
2022-02-10 14:42:21 +00:00
{
2022-04-14 19:23:57 +00:00
if ( _disposed )
{
return ;
}
2022-02-10 14:42:21 +00:00
_logger . Warn ( "A RabbitMQ connection is shutdown. Trying to re-connect..." ) ;
TryConnect ( ) ;
}
2022-02-14 11:18:58 +00:00
void OnCallbackException ( object sender , CallbackExceptionEventArgs e )
2022-02-10 14:42:21 +00:00
{
2022-04-14 19:23:57 +00:00
if ( _disposed )
{
return ;
}
2022-02-10 14:42:21 +00:00
_logger . Warn ( "A RabbitMQ connection throw exception. Trying to re-connect..." ) ;
TryConnect ( ) ;
}
2022-02-14 11:18:58 +00:00
void OnConnectionShutdown ( object sender , ShutdownEventArgs reason )
2022-02-10 14:42:21 +00:00
{
2022-04-14 19:23:57 +00:00
if ( _disposed )
{
return ;
}
2022-02-10 14:42:21 +00:00
_logger . Warn ( "A RabbitMQ connection is on shutdown. Trying to re-connect..." ) ;
TryConnect ( ) ;
}
}