99
1010namespace RabbitMQ . Stream . Client . Reliable ;
1111
12+ public record ReliableConfig
13+ {
14+ public IReconnectStrategy ReconnectStrategy { get ; set ; } = new BackOffReconnectStrategy ( ) ;
15+ public StreamSystem StreamSystem { get ; set ; }
16+ public string Stream { get ; set ; }
17+ }
18+
1219/// <summary>
1320/// Base class for Reliable producer/ consumer
21+ /// With the term Entity we mean a Producer or a Consumer
1422/// </summary>
1523public abstract class ReliableBase
1624{
1725 protected readonly SemaphoreSlim SemaphoreSlim = new ( 1 ) ;
18- protected bool _needReconnect = true ;
26+
27+ protected bool _isOpen ;
1928 protected bool _inReconnection ;
2029
21- protected async Task Init ( )
30+ public async Task Init ( IReconnectStrategy reconnectStrategy )
2231 {
23- await GetNewReliable ( true ) ;
32+ await Init ( true , reconnectStrategy ) ;
2433 }
2534
26- // boot is the first time is called.
27- // used to init the producer/consumer
28- protected abstract Task GetNewReliable ( bool boot ) ;
29-
30- protected async Task TryToReconnect ( IReconnectStrategy reconnectStrategy )
35+ // <summary>
36+ /// Init the reliable client
37+ /// <param name="boot"> If it is the First boot for the reliable P/C </param>
38+ /// <param name="reconnectStrategy">IReconnectStrategy</param>
39+ /// Try to Init the Entity, if it fails, it will try to reconnect
40+ /// only if the exception is a known exception
41+ // </summary>
42+ private async Task Init ( bool boot , IReconnectStrategy reconnectStrategy )
3143 {
32- _inReconnection = true ;
44+ var reconnect = false ;
45+ await SemaphoreSlim . WaitAsync ( ) ;
3346 try
3447 {
35- var reconnect = reconnectStrategy . WhenDisconnected ( ToString ( ) ) ;
36- var hasToReconnect = reconnect && _needReconnect ;
37- var addInfo = "Client won't reconnect" ;
38- if ( hasToReconnect )
48+ _isOpen = true ;
49+ await CreateNewEntity ( boot ) ;
50+ }
51+
52+ catch ( Exception e )
53+ {
54+ reconnect = IsAKnownException ( e ) ;
55+ LogException ( e ) ;
56+ if ( ! reconnect )
3957 {
40- addInfo = "Client will try reconnect" ;
58+ // We consider the client as closed
59+ // since the exception is raised to the caller
60+ _isOpen = false ;
61+ throw ;
4162 }
63+ }
64+ finally
65+ {
66+ SemaphoreSlim . Release ( ) ;
67+ }
4268
43- LogEventSource . Log . LogInformation (
44- $ "{ ToString ( ) } is disconnected. { addInfo } ") ;
69+ if ( reconnect )
70+ {
71+ await TryToReconnect ( reconnectStrategy ) ;
72+ }
73+ }
4574
46- if ( hasToReconnect )
47- {
48- await GetNewReliable ( false ) ;
49- }
50- else
75+ // <summary>
76+ /// Init the a new Entity (Producer/Consumer)
77+ /// <param name="boot"> If it is the First boot for the reliable P/C </param>
78+ /// Called by Init method
79+ // </summary>
80+ internal abstract Task CreateNewEntity ( bool boot ) ;
81+
82+ // <summary>
83+ /// Try to reconnect to the broker
84+ /// Based on the retry strategy
85+ /// <param name="reconnectStrategy"> The Strategy for the reconnection
86+ /// by default it is exponential backoff.
87+ /// It it possible to change implementing the IReconnectStrategy interface </param>
88+ // </summary>
89+ protected async Task TryToReconnect ( IReconnectStrategy reconnectStrategy )
90+ {
91+ _inReconnection = true ;
92+ try
93+ {
94+ // The reconnected strategy cloud decide to stop the reconnection
95+ // for some reason
96+ // _isOpen==true is when the user specified that the client must stop
97+
98+ switch ( reconnectStrategy . WhenDisconnected ( ToString ( ) ) && _isOpen )
5199 {
52- _needReconnect = false ;
53- await Close ( ) ;
100+ case true :
101+ LogEventSource . Log . LogInformation (
102+ $ "{ ToString ( ) } is disconnected. Client will try reconnect") ;
103+ await Init ( false , reconnectStrategy ) ;
104+ break ;
105+ case false :
106+ LogEventSource . Log . LogInformation (
107+ $ "{ ToString ( ) } is asked to be closed") ;
108+ await Close ( ) ;
109+ break ;
54110 }
55111 }
56112 finally
@@ -83,7 +139,7 @@ internal async Task HandleMetaDataMaybeReconnect(string stream, StreamSystem sys
83139 // Here we just close the producer connection
84140 // the func TryToReconnect/0 will be called.
85141
86- await CloseReliable ( ) ;
142+ await CloseEntity ( ) ;
87143 }
88144 else
89145 {
@@ -96,23 +152,49 @@ internal async Task HandleMetaDataMaybeReconnect(string stream, StreamSystem sys
96152 }
97153 }
98154
155+ // <summary>
156+ /// IsAKnownException returns true if the exception is a known exception
157+ /// We need it to reconnect when the producer/consumer.
158+ /// - LeaderNotFoundException is a temporary exception
159+ /// It means that the leader is not available and the client can't reconnect.
160+ /// Especially the Producer that needs to know the leader.
161+ /// - SocketException
162+ /// Client is trying to connect in a not ready endpoint.
163+ /// It is usually a temporary situation.
164+ /// - TimeoutException
165+ /// Some call went in timeout. Maybe a temporary DNS problem.
166+ /// In this case we can try to reconnect.
167+ ///
168+ /// For the other kind of exception, we just throw back the exception.
169+ //</summary>
99170 internal static bool IsAKnownException ( Exception exception )
100171 {
101172 return exception is ( SocketException or TimeoutException or LeaderNotFoundException ) ;
102173 }
103174
104- internal void LogException ( Exception exception )
175+ private void LogException ( Exception exception )
105176 {
106177 LogEventSource . Log . LogError ( IsAKnownException ( exception )
107178 ? $ "Trying to reconnect { ToString ( ) } due of: { exception . Message } "
108179 : $ "Error during initialization { ToString ( ) } due of: { exception . Message } ") ;
109180 }
110181
111- protected abstract Task CloseReliable ( ) ;
182+ // <summary>
183+ /// ONLY close the current Entity (Producer/Consumer)
184+ /// without closing the Reliable(Producer/Consumer) instance.
185+ /// It happens when the stream change topology, and the entity
186+ /// must be recreated. In the producer case for example when the
187+ /// leader changes.
188+ // </summary>
189+ protected abstract Task CloseEntity ( ) ;
190+
191+ // <summary>
192+ /// Close the Reliable(Producer/Consumer) instance.
193+ // </summary>
112194 public abstract Task Close ( ) ;
113195
114196 public bool IsOpen ( )
115197 {
116- return _needReconnect ;
198+ return _isOpen ;
117199 }
118200}
0 commit comments