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 ( true ) ;
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 , Has . Count . Zero ) ;
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 ( true ) ;
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
89+ {
90+ public TestableChannelProvider ( ) : base ( null , TimeSpan . Zero , null )
91+ {
92+ }
93+
94+ public Queue < FakeConnection > PublishConnections { get ; } = new Queue < FakeConnection > ( ) ;
95+
96+ public TaskCompletionSource < bool > DelayTaskCompletionSource { get ; } = new TaskCompletionSource < bool > ( TaskCreationOptions . RunContinuationsAsynchronously ) ;
97+
98+ public Func < CancellationToken , Task > FireAndForgetAction { get ; private set ; }
99+
100+ protected override IConnection CreatePublishConnection ( )
101+ {
102+ var connection = new FakeConnection ( ) ;
103+ PublishConnections . Enqueue ( connection ) ;
104+ return connection ;
105+ }
106+
107+ protected override void FireAndForget ( Func < CancellationToken , Task > action , CancellationToken cancellationToken = default )
108+ => FireAndForgetAction = _ => action ( cancellationToken ) ;
109+
110+ protected override async Task DelayReconnect ( CancellationToken cancellationToken = default )
111+ {
112+ using ( var _ = cancellationToken . Register ( ( ) => DelayTaskCompletionSource . TrySetCanceled ( cancellationToken ) ) )
113+ {
114+ await DelayTaskCompletionSource . Task ;
115+ }
116+ }
117+ }
118+
119+ class FakeConnection : IConnection
120+ {
121+ public int LocalPort { get ; }
122+ public int RemotePort { get ; }
123+
124+ public void Dispose ( ) => WasDisposed = true ;
125+
126+ public bool WasDisposed { get ; private set ; }
127+
128+ public void UpdateSecret ( string newSecret , string reason ) => throw new NotImplementedException ( ) ;
129+
130+ public void Abort ( ) => throw new NotImplementedException ( ) ;
131+
132+ public void Abort ( ushort reasonCode , string reasonText ) => throw new NotImplementedException ( ) ;
133+
134+ public void Abort ( TimeSpan timeout ) => throw new NotImplementedException ( ) ;
135+
136+ public void Abort ( ushort reasonCode , string reasonText , TimeSpan timeout ) => throw new NotImplementedException ( ) ;
137+
138+ public void Close ( ) => throw new NotImplementedException ( ) ;
139+
140+ public void Close ( ushort reasonCode , string reasonText ) => throw new NotImplementedException ( ) ;
141+
142+ public void Close ( TimeSpan timeout ) => throw new NotImplementedException ( ) ;
143+
144+ public void Close ( ushort reasonCode , string reasonText , TimeSpan timeout ) => throw new NotImplementedException ( ) ;
145+
146+ public IModel CreateModel ( ) => throw new NotImplementedException ( ) ;
147+
148+ public void HandleConnectionBlocked ( string reason ) => throw new NotImplementedException ( ) ;
149+
150+ public void HandleConnectionUnblocked ( ) => throw new NotImplementedException ( ) ;
151+
152+ public ushort ChannelMax { get ; }
153+ public IDictionary < string , object > ClientProperties { get ; }
154+ public ShutdownEventArgs CloseReason { get ; }
155+ public AmqpTcpEndpoint Endpoint { get ; }
156+ public uint FrameMax { get ; }
157+ public TimeSpan Heartbeat { get ; }
158+ public bool IsOpen { get ; }
159+ public AmqpTcpEndpoint [ ] KnownHosts { get ; }
160+ public IProtocol Protocol { get ; }
161+ public IDictionary < string , object > ServerProperties { get ; }
162+ public IList < ShutdownReportEntry > ShutdownReport { get ; }
163+ public string ClientProvidedName { get ; } = $ "FakeConnection{ Interlocked . Increment ( ref connectionCounter ) } ";
164+ public event EventHandler < CallbackExceptionEventArgs > CallbackException = ( sender , args ) => { } ;
165+ public event EventHandler < ConnectionBlockedEventArgs > ConnectionBlocked = ( sender , args ) => { } ;
166+ public event EventHandler < ShutdownEventArgs > ConnectionShutdown = ( sender , args ) => { } ;
167+ public event EventHandler < EventArgs > ConnectionUnblocked = ( sender , args ) => { } ;
168+
169+ public void RaiseConnectionShutdown ( ShutdownEventArgs args ) => ConnectionShutdown ? . Invoke ( this , args ) ;
170+
171+ static int connectionCounter ;
172+ }
173+ }
174+ }
0 commit comments