Skip to content

Commit 35b836d

Browse files
lukebakkenlukebakken
committed
Add CreateConnectionAsync
1 parent 5c936ff commit 35b836d

File tree

6 files changed

+284
-29
lines changed

6 files changed

+284
-29
lines changed

‎projects/RabbitMQ.Client/client/api/ConnectionFactory.cs‎

Lines changed: 164 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
using System.Reflection;
3737
using System.Security.Authentication;
3838
using System.Text;
39+
using System.Threading.Tasks;
3940
using RabbitMQ.Client.Exceptions;
4041
using RabbitMQ.Client.Framing.Impl;
4142
using RabbitMQ.Client.Impl;
@@ -410,6 +411,19 @@ public IConnection CreateConnection()
410411
return CreateConnection(ClientProvidedName);
411412
}
412413

414+
/// <summary>
415+
/// Asynchronously reate a connection to one of the endpoints provided by the IEndpointResolver
416+
/// returned by the EndpointResolverFactory. By default the configured
417+
/// hostname and port are used.
418+
/// </summary>
419+
/// <exception cref="BrokerUnreachableException">
420+
/// When the configured hostname was not reachable.
421+
/// </exception>
422+
public ValueTask<IConnection> CreateConnectionAsync()
423+
{
424+
return CreateConnectionAsync(ClientProvidedName);
425+
}
426+
413427
/// <summary>
414428
/// Create a connection to one of the endpoints provided by the IEndpointResolver
415429
/// returned by the EndpointResolverFactory. By default the configured
@@ -429,6 +443,25 @@ public IConnection CreateConnection(string clientProvidedName)
429443
return CreateConnection(EndpointResolverFactory(LocalEndpoints()), clientProvidedName);
430444
}
431445

446+
/// <summary>
447+
/// Asynchronously create a connection to one of the endpoints provided by the IEndpointResolver
448+
/// returned by the EndpointResolverFactory. By default the configured
449+
/// hostname and port are used.
450+
/// </summary>
451+
/// <param name="clientProvidedName">
452+
/// Application-specific connection name, will be displayed in the management UI
453+
/// if RabbitMQ server supports it. This value doesn't have to be unique and cannot
454+
/// be used as a connection identifier, e.g. in HTTP API requests.
455+
/// This value is supposed to be human-readable.
456+
/// </param>
457+
/// <exception cref="BrokerUnreachableException">
458+
/// When the configured hostname was not reachable.
459+
/// </exception>
460+
public ValueTask<IConnection> CreateConnectionAsync(string clientProvidedName)
461+
{
462+
return CreateConnectionAsync(EndpointResolverFactory(LocalEndpoints()), clientProvidedName);
463+
}
464+
432465
/// <summary>
433466
/// Create a connection using a list of hostnames using the configured port.
434467
/// By default each hostname is tried in a random order until a successful connection is
@@ -448,6 +481,25 @@ public IConnection CreateConnection(IList<string> hostnames)
448481
return CreateConnection(hostnames, ClientProvidedName);
449482
}
450483

484+
/// <summary>
485+
/// Asynchronously create a connection using a list of hostnames using the configured port.
486+
/// By default each hostname is tried in a random order until a successful connection is
487+
/// found or the list is exhausted using the DefaultEndpointResolver.
488+
/// The selection behaviour can be overridden by configuring the EndpointResolverFactory.
489+
/// </summary>
490+
/// <param name="hostnames">
491+
/// List of hostnames to use for the initial
492+
/// connection and recovery.
493+
/// </param>
494+
/// <returns>Open connection</returns>
495+
/// <exception cref="BrokerUnreachableException">
496+
/// When no hostname was reachable.
497+
/// </exception>
498+
public ValueTask<IConnection> CreateConnectionAsync(IList<string> hostnames)
499+
{
500+
return CreateConnectionAsync(hostnames, ClientProvidedName);
501+
}
502+
451503
/// <summary>
452504
/// Create a connection using a list of hostnames using the configured port.
453505
/// By default each endpoint is tried in a random order until a successful connection is
@@ -474,6 +526,32 @@ public IConnection CreateConnection(IList<string> hostnames, string clientProvid
474526
return CreateConnection(EndpointResolverFactory(endpoints), clientProvidedName);
475527
}
476528

529+
/// <summary>
530+
/// Asynchronously create a connection using a list of hostnames using the configured port.
531+
/// By default each endpoint is tried in a random order until a successful connection is
532+
/// found or the list is exhausted.
533+
/// The selection behaviour can be overridden by configuring the EndpointResolverFactory.
534+
/// </summary>
535+
/// <param name="hostnames">
536+
/// List of hostnames to use for the initial
537+
/// connection and recovery.
538+
/// </param>
539+
/// <param name="clientProvidedName">
540+
/// Application-specific connection name, will be displayed in the management UI
541+
/// if RabbitMQ server supports it. This value doesn't have to be unique and cannot
542+
/// be used as a connection identifier, e.g. in HTTP API requests.
543+
/// This value is supposed to be human-readable.
544+
/// </param>
545+
/// <returns>Open connection</returns>
546+
/// <exception cref="BrokerUnreachableException">
547+
/// When no hostname was reachable.
548+
/// </exception>
549+
public ValueTask<IConnection> CreateConnectionAsync(IList<string> hostnames, string clientProvidedName)
550+
{
551+
IEnumerable<AmqpTcpEndpoint> endpoints = hostnames.Select(h => new AmqpTcpEndpoint(h, Port, Ssl, MaxMessageSize));
552+
return CreateConnectionAsync(EndpointResolverFactory(endpoints), clientProvidedName);
553+
}
554+
477555
/// <summary>
478556
/// Create a connection using a list of endpoints. By default each endpoint will be tried
479557
/// in a random order until a successful connection is found or the list is exhausted.
@@ -492,6 +570,24 @@ public IConnection CreateConnection(IList<AmqpTcpEndpoint> endpoints)
492570
return CreateConnection(endpoints, ClientProvidedName);
493571
}
494572

573+
/// <summary>
574+
/// Asynchronously create a connection using a list of endpoints. By default each endpoint will be tried
575+
/// in a random order until a successful connection is found or the list is exhausted.
576+
/// The selection behaviour can be overridden by configuring the EndpointResolverFactory.
577+
/// </summary>
578+
/// <param name="endpoints">
579+
/// List of endpoints to use for the initial
580+
/// connection and recovery.
581+
/// </param>
582+
/// <returns>Open connection</returns>
583+
/// <exception cref="BrokerUnreachableException">
584+
/// When no hostname was reachable.
585+
/// </exception>
586+
public ValueTask<IConnection> CreateConnectionAsync(IList<AmqpTcpEndpoint> endpoints)
587+
{
588+
return CreateConnectionAsync(endpoints, ClientProvidedName);
589+
}
590+
495591
/// <summary>
496592
/// Create a connection using a list of endpoints. By default each endpoint will be tried
497593
/// in a random order until a successful connection is found or the list is exhausted.
@@ -516,6 +612,30 @@ public IConnection CreateConnection(IList<AmqpTcpEndpoint> endpoints, string cli
516612
return CreateConnection(EndpointResolverFactory(endpoints), clientProvidedName);
517613
}
518614

615+
/// <summary>
616+
/// Asynchronously create a connection using a list of endpoints. By default each endpoint will be tried
617+
/// in a random order until a successful connection is found or the list is exhausted.
618+
/// The selection behaviour can be overridden by configuring the EndpointResolverFactory.
619+
/// </summary>
620+
/// <param name="endpoints">
621+
/// List of endpoints to use for the initial
622+
/// connection and recovery.
623+
/// </param>
624+
/// <param name="clientProvidedName">
625+
/// Application-specific connection name, will be displayed in the management UI
626+
/// if RabbitMQ server supports it. This value doesn't have to be unique and cannot
627+
/// be used as a connection identifier, e.g. in HTTP API requests.
628+
/// This value is supposed to be human-readable.
629+
/// </param>
630+
/// <returns>Open connection</returns>
631+
/// <exception cref="BrokerUnreachableException">
632+
/// When no hostname was reachable.
633+
/// </exception>
634+
public ValueTask<IConnection> CreateConnectionAsync(IList<AmqpTcpEndpoint> endpoints, string clientProvidedName)
635+
{
636+
return CreateConnectionAsync(EndpointResolverFactory(endpoints), clientProvidedName);
637+
}
638+
519639
/// <summary>
520640
/// Create a connection using an IEndpointResolver.
521641
/// </summary>
@@ -539,10 +659,52 @@ public IConnection CreateConnection(IEndpointResolver endpointResolver, string c
539659
{
540660
if (AutomaticRecoveryEnabled)
541661
{
542-
return new AutorecoveringConnection(config, endpointResolver);
662+
var c = new AutorecoveringConnection(config, endpointResolver);
663+
return c.Open();
664+
}
665+
else
666+
{
667+
var c = new Connection(config, endpointResolver.SelectOne(CreateFrameHandler));
668+
return c.Open();
543669
}
670+
}
671+
catch (Exception e)
672+
{
673+
throw new BrokerUnreachableException(e);
674+
}
675+
}
544676

545-
return new Connection(config, endpointResolver.SelectOne(CreateFrameHandler));
677+
/// <summary>
678+
/// Asynchronously create a connection using an IEndpointResolver.
679+
/// </summary>
680+
/// <param name="endpointResolver">
681+
/// The endpointResolver that returns the endpoints to use for the connection attempt.
682+
/// </param>
683+
/// <param name="clientProvidedName">
684+
/// Application-specific connection name, will be displayed in the management UI
685+
/// if RabbitMQ server supports it. This value doesn't have to be unique and cannot
686+
/// be used as a connection identifier, e.g. in HTTP API requests.
687+
/// This value is supposed to be human-readable.
688+
/// </param>
689+
/// <returns>Open connection</returns>
690+
/// <exception cref="BrokerUnreachableException">
691+
/// When no hostname was reachable.
692+
/// </exception>
693+
public ValueTask<IConnection> CreateConnectionAsync(IEndpointResolver endpointResolver, string clientProvidedName)
694+
{
695+
ConnectionConfig config = CreateConfig(clientProvidedName);
696+
try
697+
{
698+
if (AutomaticRecoveryEnabled)
699+
{
700+
var c = new AutorecoveringConnection(config, endpointResolver);
701+
return c.OpenAsync();
702+
}
703+
else
704+
{
705+
var c = new Connection(config, endpointResolver.SelectOne(CreateFrameHandler));
706+
return c.OpenAsync();
707+
}
546708
}
547709
catch (Exception e)
548710
{

‎projects/RabbitMQ.Client/client/api/IConnectionFactory.cs‎

Lines changed: 72 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@
3131

3232
using System;
3333
using System.Collections.Generic;
34-
34+
using System.Threading.Tasks;
3535
using RabbitMQ.Client.Exceptions;
3636

3737
namespace RabbitMQ.Client
@@ -102,6 +102,11 @@ public interface IConnectionFactory
102102
/// </summary>
103103
IConnection CreateConnection();
104104

105+
/// <summary>
106+
/// Asynchronously create a connection to the specified endpoint.
107+
/// </summary>
108+
ValueTask<IConnection> CreateConnectionAsync();
109+
105110
/// <summary>
106111
/// Create a connection to the specified endpoint.
107112
/// </summary>
@@ -114,13 +119,32 @@ public interface IConnectionFactory
114119
/// <returns>Open connection</returns>
115120
IConnection CreateConnection(string clientProvidedName);
116121

122+
/// <summary>
123+
/// Asynchronously create a connection to the specified endpoint.
124+
/// </summary>
125+
/// <param name="clientProvidedName">
126+
/// Application-specific connection name, will be displayed in the management UI
127+
/// if RabbitMQ server supports it. This value doesn't have to be unique and cannot
128+
/// be used as a connection identifier, e.g. in HTTP API requests.
129+
/// This value is supposed to be human-readable.
130+
/// </param>
131+
/// <returns>Open connection</returns>
132+
ValueTask<IConnection> CreateConnectionAsync(string clientProvidedName);
133+
117134
/// <summary>
118135
/// Connects to the first reachable hostname from the list.
119136
/// </summary>
120137
/// <param name="hostnames">List of host names to use</param>
121138
/// <returns>Open connection</returns>
122139
IConnection CreateConnection(IList<string> hostnames);
123140

141+
/// <summary>
142+
/// Asynchronously connects to the first reachable hostname from the list.
143+
/// </summary>
144+
/// <param name="hostnames">List of host names to use</param>
145+
/// <returns>Open connection</returns>
146+
ValueTask<IConnection> CreateConnectionAsync(IList<string> hostnames);
147+
124148
/// <summary>
125149
/// Connects to the first reachable hostname from the list.
126150
/// </summary>
@@ -134,6 +158,19 @@ public interface IConnectionFactory
134158
/// <returns>Open connection</returns>
135159
IConnection CreateConnection(IList<string> hostnames, string clientProvidedName);
136160

161+
/// <summary>
162+
/// Asynchronously connects to the first reachable hostname from the list.
163+
/// </summary>
164+
/// <param name="hostnames">List of host names to use</param>
165+
/// <param name="clientProvidedName">
166+
/// Application-specific connection name, will be displayed in the management UI
167+
/// if RabbitMQ server supports it. This value doesn't have to be unique and cannot
168+
/// be used as a connection identifier, e.g. in HTTP API requests.
169+
/// This value is supposed to be human-readable.
170+
/// </param>
171+
/// <returns>Open connection</returns>
172+
ValueTask<IConnection> CreateConnectionAsync(IList<string> hostnames, string clientProvidedName);
173+
137174
/// <summary>
138175
/// Create a connection using a list of endpoints.
139176
/// The selection behaviour can be overridden by configuring the EndpointResolverFactory.
@@ -148,6 +185,20 @@ public interface IConnectionFactory
148185
/// </exception>
149186
IConnection CreateConnection(IList<AmqpTcpEndpoint> endpoints);
150187

188+
/// <summary>
189+
/// Asynchronously create a connection using a list of endpoints.
190+
/// The selection behaviour can be overridden by configuring the EndpointResolverFactory.
191+
/// </summary>
192+
/// <param name="endpoints">
193+
/// List of endpoints to use for the initial
194+
/// connection and recovery.
195+
/// </param>
196+
/// <returns>Open connection</returns>
197+
/// <exception cref="BrokerUnreachableException">
198+
/// When no hostname was reachable.
199+
/// </exception>
200+
ValueTask<IConnection> CreateConnectionAsync(IList<AmqpTcpEndpoint> endpoints);
201+
151202
/// <summary>
152203
/// Create a connection using a list of endpoints.
153204
/// The selection behaviour can be overridden by configuring the EndpointResolverFactory.
@@ -168,6 +219,26 @@ public interface IConnectionFactory
168219
/// </exception>
169220
IConnection CreateConnection(IList<AmqpTcpEndpoint> endpoints, string clientProvidedName);
170221

222+
/// <summary>
223+
/// Asynchronously create a connection using a list of endpoints.
224+
/// The selection behaviour can be overridden by configuring the EndpointResolverFactory.
225+
/// </summary>
226+
/// <param name="endpoints">
227+
/// List of endpoints to use for the initial
228+
/// connection and recovery.
229+
/// </param>
230+
/// <param name="clientProvidedName">
231+
/// Application-specific connection name, will be displayed in the management UI
232+
/// if RabbitMQ server supports it. This value doesn't have to be unique and cannot
233+
/// be used as a connection identifier, e.g. in HTTP API requests.
234+
/// This value is supposed to be human-readable.
235+
/// </param>
236+
/// <returns>Open connection</returns>
237+
/// <exception cref="BrokerUnreachableException">
238+
/// When no hostname was reachable.
239+
/// </exception>
240+
ValueTask<IConnection> CreateConnectionAsync(IList<AmqpTcpEndpoint> endpoints, string clientProvidedName);
241+
171242
/// <summary>
172243
/// Amount of time protocol handshake operations are allowed to take before
173244
/// timing out.

‎projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.cs‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,7 @@ private Connection InnerConnection
5858
}
5959
}
6060

61-
public AutorecoveringConnection(ConnectionConfig config, IEndpointResolver endpoints)
61+
internal AutorecoveringConnection(ConnectionConfig config, IEndpointResolver endpoints)
6262
{
6363
_config = config;
6464
_endpoints = endpoints;
@@ -78,6 +78,16 @@ public AutorecoveringConnection(ConnectionConfig config, IEndpointResolver endpo
7878
ConnectionShutdown += HandleConnectionShutdown;
7979
}
8080

81+
internal IConnection Open()
82+
{
83+
return InnerConnection.Open();
84+
}
85+
86+
internal ValueTask<IConnection> OpenAsync()
87+
{
88+
return InnerConnection.OpenAsync();
89+
}
90+
8191
public event EventHandler<EventArgs> RecoverySucceeded
8292
{
8393
add => _recoverySucceededWrapper.AddHandler(value);

‎projects/RabbitMQ.Client/client/impl/Connection.Commands.cs‎

Lines changed: 0 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,6 @@
3636
using RabbitMQ.Client.Events;
3737
using RabbitMQ.Client.Exceptions;
3838
using RabbitMQ.Client.Impl;
39-
using RabbitMQ.Client.Logging;
4039

4140
namespace RabbitMQ.Client.Framing.Impl
4241
{
@@ -70,13 +69,6 @@ internal void HandleConnectionUnblocked()
7069
}
7170
}
7271

73-
private async ValueTask OpenAsync()
74-
{
75-
RabbitMqClientEventSource.Log.ConnectionOpened();
76-
await StartAndTuneAsync().ConfigureAwait(false);
77-
await _channel0.ConnectionOpenAsync(_config.VirtualHost);
78-
}
79-
8072
private async ValueTask StartAndTuneAsync()
8173
{
8274
var connectionStartCell = new TaskCompletionSource<ConnectionStartDetails>(TaskCreationOptions.RunContinuationsAsynchronously);

0 commit comments

Comments
 (0)