1+ namespace NServiceBus . Transport . RabbitMQ . Tests . ConnectionString
2+ {
3+ using System ;
4+ using System . Collections . Generic ;
5+ using System . Threading ;
6+ using System . Threading . Tasks ;
7+ using global RabbitMQ . Client ;
8+ using global RabbitMQ . Client . Events ;
9+ using NUnit . Framework ;
10+
11+ [ TestFixture ]
12+ public class ChannelProviderTests
13+ {
14+ [ Test ]
15+ public async Task Should_recover_connection_and_dispose_old_one_when_connection_shutdown ( )
16+ {
17+ var channelProvider = new TestableChannelProvider ( ) ;
18+ channelProvider . CreateConnection ( ) ;
19+
20+ var publishConnection = channelProvider . PublishConnections . Dequeue ( ) ;
21+ publishConnection . RaiseConnectionShutdown ( new ShutdownEventArgs ( ShutdownInitiator . Library , 0 , "Test" ) ) ;
22+
23+ channelProvider . DelayTaskCompletionSource . SetResult ( ) ;
24+
25+ await channelProvider . FireAndForgetAction ( CancellationToken . None ) ;
26+
27+ var recoveredConnection = channelProvider . PublishConnections . Dequeue ( ) ;
28+
29+ Assert . That ( publishConnection . WasDisposed , Is . True ) ;
30+ Assert . That ( recoveredConnection . WasDisposed , Is . False ) ;
31+ }
32+
33+ [ Test ]
34+ public void Should_dispose_connection_when_disposed ( )
35+ {
36+ var channelProvider = new TestableChannelProvider ( ) ;
37+ channelProvider . CreateConnection ( ) ;
38+
39+ var publishConnection = channelProvider . PublishConnections . Dequeue ( ) ;
40+ channelProvider . Dispose ( ) ;
41+
42+ Assert . That ( publishConnection . WasDisposed , Is . True ) ;
43+ }
44+
45+ [ Test ]
46+ public async Task Should_not_attempt_to_recover_during_dispose_when_retry_delay_still_pending ( )
47+ {
48+ var channelProvider = new TestableChannelProvider ( ) ;
49+ channelProvider . CreateConnection ( ) ;
50+
51+ var publishConnection = channelProvider . PublishConnections . Dequeue ( ) ;
52+ publishConnection . RaiseConnectionShutdown ( new ShutdownEventArgs ( ShutdownInitiator . Library , 0 , "Test" ) ) ;
53+
54+ // Deliberately not completing the delay task with channelProvider.DelayTaskCompletionSource.SetResult(); before disposing
55+ // to simulate a pending delay task
56+ channelProvider . Dispose ( ) ;
57+
58+ await channelProvider . FireAndForgetAction ( CancellationToken . None ) ;
59+
60+ Assert . That ( publishConnection . WasDisposed , Is . True ) ;
61+ Assert . That ( channelProvider . PublishConnections . TryDequeue ( out _ ) , Is . False ) ;
62+ }
63+
64+ [ Test ]
65+ public async Task Should_dispose_newly_established_connection ( )
66+ {
67+ var channelProvider = new TestableChannelProvider ( ) ;
68+ channelProvider . CreateConnection ( ) ;
69+
70+ var publishConnection = channelProvider . PublishConnections . Dequeue ( ) ;
71+ publishConnection . RaiseConnectionShutdown ( new ShutdownEventArgs ( ShutdownInitiator . Library , 0 , "Test" ) ) ;
72+
73+ // This simulates the race of the reconnection loop being fired off with the delay task completed during
74+ // the disposal of the channel provider. To achieve that it is necessary to kick off the reconnection loop
75+ // and await its completion after the channel provider has been disposed.
76+ var fireAndForgetTask = channelProvider . FireAndForgetAction ( CancellationToken . None ) ;
77+ channelProvider . DelayTaskCompletionSource . SetResult ( ) ;
78+ channelProvider . Dispose ( ) ;
79+
80+ await fireAndForgetTask ;
81+
82+ var recoveredConnection = channelProvider . PublishConnections . Dequeue ( ) ;
83+
84+ Assert . That ( publishConnection . WasDisposed , Is . True ) ;
85+ Assert . That ( recoveredConnection . WasDisposed , Is . True ) ;
86+ }
87+
88+ class TestableChannelProvider ( ) : ChannelProvider ( null ! , TimeSpan . Zero , null ! )
89+ {
90+ public Queue < FakeConnection > PublishConnections { get ; } = new ( ) ;
91+
92+ public TaskCompletionSource DelayTaskCompletionSource { get ; } = new ( TaskCreationOptions . RunContinuationsAsynchronously ) ;
93+
94+ public Func < CancellationToken , Task > FireAndForgetAction { get ; private set ; }
95+
96+ protected override IConnection CreatePublishConnection ( )
97+ {
98+ var connection = new FakeConnection ( ) ;
99+ PublishConnections . Enqueue ( connection ) ;
100+ return connection ;
101+ }
102+
103+ protected override void FireAndForget ( Func < CancellationToken , Task > action , CancellationToken cancellationToken = default )
104+ => FireAndForgetAction = _ => action ( cancellationToken ) ;
105+
106+ protected override async Task DelayReconnect ( CancellationToken cancellationToken = default )
107+ {
108+ await using var _ = cancellationToken . Register ( ( ) => DelayTaskCompletionSource . TrySetCanceled ( cancellationToken ) ) ;
109+ await DelayTaskCompletionSource . Task ;
110+ }
111+ }
112+
113+ class FakeConnection : IConnection
114+ {
115+ public int LocalPort { get ; }
116+ public int RemotePort { get ; }
117+
118+ public void Dispose ( ) => WasDisposed = true ;
119+
120+ public bool WasDisposed { get ; private set ; }
121+
122+ public void UpdateSecret ( string newSecret , string reason ) => throw new NotImplementedException ( ) ;
123+
124+ public void Abort ( ) => throw new NotImplementedException ( ) ;
125+
126+ public void Abort ( ushort reasonCode , string reasonText ) => throw new NotImplementedException ( ) ;
127+
128+ public void Abort ( TimeSpan timeout ) => throw new NotImplementedException ( ) ;
129+
130+ public void Abort ( ushort reasonCode , string reasonText , TimeSpan timeout ) => throw new NotImplementedException ( ) ;
131+
132+ public void Close ( ) => throw new NotImplementedException ( ) ;
133+
134+ public void Close ( ushort reasonCode , string reasonText ) => throw new NotImplementedException ( ) ;
135+
136+ public void Close ( TimeSpan timeout ) => throw new NotImplementedException ( ) ;
137+
138+ public void Close ( ushort reasonCode , string reasonText , TimeSpan timeout ) => throw new NotImplementedException ( ) ;
139+
140+ public IModel CreateModel ( ) => throw new NotImplementedException ( ) ;
141+
142+ public void HandleConnectionBlocked ( string reason ) => throw new NotImplementedException ( ) ;
143+
144+ public void HandleConnectionUnblocked ( ) => throw new NotImplementedException ( ) ;
145+
146+ public ushort ChannelMax { get ; }
147+ public IDictionary < string , object > ClientProperties { get ; }
148+ public ShutdownEventArgs CloseReason { get ; }
149+ public AmqpTcpEndpoint Endpoint { get ; }
150+ public uint FrameMax { get ; }
151+ public TimeSpan Heartbeat { get ; }
152+ public bool IsOpen { get ; }
153+ public AmqpTcpEndpoint [ ] KnownHosts { get ; }
154+ public IProtocol Protocol { get ; }
155+ public IDictionary < string , object > ServerProperties { get ; }
156+ public IList < ShutdownReportEntry > ShutdownReport { get ; }
157+ public string ClientProvidedName { get ; } = $ "FakeConnection{ Interlocked . Increment ( ref connectionCounter ) } ";
158+ public event EventHandler < CallbackExceptionEventArgs > CallbackException = ( _ , _ ) => { } ;
159+ public event EventHandler < ConnectionBlockedEventArgs > ConnectionBlocked = ( _ , _ ) => { } ;
160+ public event EventHandler < ShutdownEventArgs > ConnectionShutdown = ( _ , _ ) => { } ;
161+ public event EventHandler < EventArgs > ConnectionUnblocked = ( _ , _ ) => { } ;
162+
163+ public void RaiseConnectionShutdown ( ShutdownEventArgs args ) => ConnectionShutdown ? . Invoke ( this , args ) ;
164+
165+ static int connectionCounter ;
166+ }
167+ }
168+ }
0 commit comments