| | | 1 | | // Copyright (c) Microsoft. All rights reserved. |
| | | 2 | | // Licensed under the MIT license. See LICENSE file in the project root for full license information. |
| | | 3 | | |
| | | 4 | | namespace Microsoft.Azure.ServiceBus |
| | | 5 | | { |
| | | 6 | | using System; |
| | | 7 | | using System.Net; |
| | | 8 | | using System.Threading.Tasks; |
| | | 9 | | using Microsoft.Azure.Amqp; |
| | | 10 | | using Microsoft.Azure.Amqp.Framing; |
| | | 11 | | using Microsoft.Azure.Amqp.Transaction; |
| | | 12 | | using Microsoft.Azure.Amqp.Transport; |
| | | 13 | | using Microsoft.Azure.ServiceBus.Amqp; |
| | | 14 | | using Microsoft.Azure.ServiceBus.Primitives; |
| | | 15 | | |
| | | 16 | | /// <summary> |
| | | 17 | | /// Connection object to service bus namespace |
| | | 18 | | /// </summary> |
| | | 19 | | public class ServiceBusConnection |
| | | 20 | | { |
| | 0 | 21 | | static readonly Version AmqpVersion = new Version(1, 0, 0, 0); |
| | | 22 | | readonly object syncLock; |
| | | 23 | | bool isClosedOrClosing; |
| | | 24 | | |
| | | 25 | | /// <summary> |
| | | 26 | | /// Creates a new connection to service bus. |
| | | 27 | | /// </summary> |
| | | 28 | | /// <param name="connectionStringBuilder"><see cref="ServiceBusConnectionStringBuilder"/> having namespace infor |
| | | 29 | | /// <remarks>It is the responsibility of the user to close the connection after use through <see cref="CloseAsyn |
| | | 30 | | public ServiceBusConnection(ServiceBusConnectionStringBuilder connectionStringBuilder) |
| | 18 | 31 | | : this(connectionStringBuilder?.GetNamespaceConnectionString()) |
| | | 32 | | { |
| | 18 | 33 | | } |
| | | 34 | | |
| | | 35 | | /// <summary> |
| | | 36 | | /// Creates a new connection to service bus. |
| | | 37 | | /// </summary> |
| | | 38 | | /// <param name="namespaceConnectionString">Namespace connection string</param> |
| | | 39 | | /// <remarks>It is the responsibility of the user to close the connection after use through <see cref="CloseAsyn |
| | | 40 | | public ServiceBusConnection(string namespaceConnectionString) |
| | 26 | 41 | | : this(namespaceConnectionString, RetryPolicy.Default) |
| | | 42 | | { |
| | 26 | 43 | | } |
| | | 44 | | |
| | | 45 | | /// <summary> |
| | | 46 | | /// Creates a new connection to service bus. |
| | | 47 | | /// </summary> |
| | | 48 | | /// <param name="namespaceConnectionString">Namespace connection string.</param> |
| | | 49 | | /// <param name="retryPolicy">Retry policy for operations. Defaults to <see cref="RetryPolicy.Default"/></param> |
| | | 50 | | /// <remarks>It is the responsibility of the user to close the connection after use through <see cref="CloseAsyn |
| | | 51 | | public ServiceBusConnection(string namespaceConnectionString, RetryPolicy retryPolicy = null) |
| | 26 | 52 | | : this(retryPolicy) |
| | | 53 | | { |
| | 26 | 54 | | if (string.IsNullOrWhiteSpace(namespaceConnectionString)) |
| | | 55 | | { |
| | 0 | 56 | | throw Fx.Exception.ArgumentNullOrWhiteSpace(nameof(namespaceConnectionString)); |
| | | 57 | | } |
| | | 58 | | |
| | 26 | 59 | | var serviceBusConnectionStringBuilder = new ServiceBusConnectionStringBuilder(namespaceConnectionString); |
| | 26 | 60 | | if (!string.IsNullOrWhiteSpace(serviceBusConnectionStringBuilder.EntityPath)) |
| | | 61 | | { |
| | 0 | 62 | | throw Fx.Exception.Argument(nameof(namespaceConnectionString), "NamespaceConnectionString should not con |
| | | 63 | | } |
| | | 64 | | |
| | 26 | 65 | | this.InitializeConnection(serviceBusConnectionStringBuilder); |
| | 26 | 66 | | } |
| | | 67 | | |
| | | 68 | | /// <summary> |
| | | 69 | | /// Creates a new connection to service bus. |
| | | 70 | | /// </summary> |
| | | 71 | | /// <param name="namespaceConnectionString">Namespace connection string.</param> |
| | | 72 | | /// <param name="operationTimeout">Duration after which individual operations will timeout.</param> |
| | | 73 | | /// <param name="retryPolicy">Retry policy for operations. Defaults to <see cref="RetryPolicy.Default"/></param> |
| | | 74 | | /// <remarks>It is the responsibility of the user to close the connection after use through <see cref="CloseAsyn |
| | | 75 | | [Obsolete("This constructor is obsolete. Use ServiceBusConnection(string namespaceConnectionString, RetryPolicy |
| | | 76 | | public ServiceBusConnection(string namespaceConnectionString, TimeSpan operationTimeout, RetryPolicy retryPolicy |
| | 0 | 77 | | : this(retryPolicy) |
| | | 78 | | { |
| | 0 | 79 | | if (string.IsNullOrWhiteSpace(namespaceConnectionString)) |
| | | 80 | | { |
| | 0 | 81 | | throw Fx.Exception.ArgumentNullOrWhiteSpace(nameof(namespaceConnectionString)); |
| | | 82 | | } |
| | | 83 | | |
| | 0 | 84 | | var serviceBusConnectionStringBuilder = new ServiceBusConnectionStringBuilder(namespaceConnectionString); |
| | 0 | 85 | | if (!string.IsNullOrWhiteSpace(serviceBusConnectionStringBuilder.EntityPath)) |
| | | 86 | | { |
| | 0 | 87 | | throw Fx.Exception.Argument(nameof(namespaceConnectionString), "NamespaceConnectionString should not con |
| | | 88 | | } |
| | | 89 | | |
| | 0 | 90 | | this.InitializeConnection(serviceBusConnectionStringBuilder); |
| | | 91 | | // operationTimeout argument explicitly provided by caller should take precedence over OperationTimeout foun |
| | 0 | 92 | | this.OperationTimeout = operationTimeout; |
| | 0 | 93 | | } |
| | | 94 | | |
| | | 95 | | /// <summary> |
| | | 96 | | /// Creates a new connection to service bus. |
| | | 97 | | /// </summary> |
| | | 98 | | /// <param name="endpoint">Fully qualified domain name for Service Bus. Most likely, {yournamespace}.servicebus. |
| | | 99 | | /// <param name="transportType">Transport type.</param> |
| | | 100 | | /// <param name="retryPolicy">Retry policy for operations. Defaults to <see cref="RetryPolicy.Default"/></param> |
| | | 101 | | public ServiceBusConnection(string endpoint, TransportType transportType, RetryPolicy retryPolicy = null) |
| | 0 | 102 | | : this(retryPolicy) |
| | | 103 | | { |
| | 0 | 104 | | if (string.IsNullOrWhiteSpace(endpoint)) |
| | | 105 | | { |
| | 0 | 106 | | throw Fx.Exception.ArgumentNullOrWhiteSpace(nameof(endpoint)); |
| | | 107 | | } |
| | | 108 | | |
| | 0 | 109 | | var serviceBusConnectionStringBuilder = new ServiceBusConnectionStringBuilder() |
| | 0 | 110 | | { |
| | 0 | 111 | | Endpoint = endpoint, |
| | 0 | 112 | | TransportType = transportType |
| | 0 | 113 | | }; |
| | | 114 | | |
| | 0 | 115 | | this.InitializeConnection(serviceBusConnectionStringBuilder); |
| | 0 | 116 | | } |
| | | 117 | | |
| | 26 | 118 | | internal ServiceBusConnection(RetryPolicy retryPolicy = null) |
| | | 119 | | { |
| | 26 | 120 | | this.RetryPolicy = retryPolicy ?? RetryPolicy.Default; |
| | 26 | 121 | | this.syncLock = new object(); |
| | 26 | 122 | | } |
| | | 123 | | |
| | | 124 | | /// <summary> |
| | | 125 | | /// Fully qualified domain name for Service Bus. |
| | | 126 | | /// </summary> |
| | 64 | 127 | | public Uri Endpoint { get; set; } |
| | | 128 | | |
| | | 129 | | /// <summary> |
| | | 130 | | /// OperationTimeout is applied in erroneous situations to notify the caller about the relevant <see cref="Servi |
| | | 131 | | /// </summary> |
| | | 132 | | /// <remarks>Defaults to 1 minute.</remarks> |
| | 38 | 133 | | public TimeSpan OperationTimeout { get; set; } |
| | | 134 | | |
| | | 135 | | /// <summary> |
| | | 136 | | /// Retry policy for operations performed on the connection. |
| | | 137 | | /// </summary> |
| | | 138 | | /// <remarks>Defaults to <see cref="RetryPolicy.Default"/></remarks> |
| | 0 | 139 | | public RetryPolicy RetryPolicy { get; set; } |
| | | 140 | | |
| | | 141 | | /// <summary> |
| | | 142 | | /// Get the transport type from the connection string. |
| | | 143 | | /// <remarks>Available options: Amqp and AmqpWebSockets.</remarks> |
| | | 144 | | /// </summary> |
| | 30 | 145 | | public TransportType TransportType { get; set; } |
| | | 146 | | |
| | | 147 | | /// <summary> |
| | | 148 | | /// Token provider for authentication. <see cref="TokenProvider"/> |
| | | 149 | | /// </summary> |
| | 58 | 150 | | public ITokenProvider TokenProvider { get; set; } |
| | | 151 | | |
| | | 152 | | /// <summary> |
| | | 153 | | /// Returns true if the Service Bus Connection is closed or closing. |
| | | 154 | | /// </summary> |
| | | 155 | | public bool IsClosedOrClosing |
| | | 156 | | { |
| | | 157 | | get |
| | | 158 | | { |
| | 12 | 159 | | lock (syncLock) |
| | | 160 | | { |
| | 12 | 161 | | return isClosedOrClosing; |
| | | 162 | | } |
| | 12 | 163 | | } |
| | | 164 | | internal set |
| | | 165 | | { |
| | 0 | 166 | | lock (syncLock) |
| | | 167 | | { |
| | 0 | 168 | | isClosedOrClosing = value; |
| | 0 | 169 | | } |
| | 0 | 170 | | } |
| | | 171 | | } |
| | | 172 | | |
| | 0 | 173 | | internal FaultTolerantAmqpObject<AmqpConnection> ConnectionManager { get; set; } |
| | | 174 | | |
| | 0 | 175 | | internal FaultTolerantAmqpObject<Controller> TransactionController { get; set; } |
| | | 176 | | |
| | | 177 | | /// <summary> |
| | | 178 | | /// Throw an OperationCanceledException if the object is Closing. |
| | | 179 | | /// </summary> |
| | | 180 | | internal virtual void ThrowIfClosed() |
| | | 181 | | { |
| | 12 | 182 | | if (this.IsClosedOrClosing) |
| | | 183 | | { |
| | 0 | 184 | | throw new ObjectDisposedException($"{nameof(ServiceBusConnection)} has already been closed. Please creat |
| | | 185 | | } |
| | 12 | 186 | | } |
| | | 187 | | |
| | | 188 | | /// <summary> |
| | | 189 | | /// Closes the connection. |
| | | 190 | | /// </summary> |
| | | 191 | | public async Task CloseAsync() |
| | | 192 | | { |
| | 0 | 193 | | var callClose = false; |
| | 0 | 194 | | lock (this.syncLock) |
| | | 195 | | { |
| | 0 | 196 | | if (!this.IsClosedOrClosing) |
| | | 197 | | { |
| | 0 | 198 | | this.IsClosedOrClosing = true; |
| | 0 | 199 | | callClose = true; |
| | | 200 | | } |
| | 0 | 201 | | } |
| | | 202 | | |
| | 0 | 203 | | if (callClose) |
| | | 204 | | { |
| | 0 | 205 | | await this.ConnectionManager.CloseAsync().ConfigureAwait(false); |
| | | 206 | | } |
| | 0 | 207 | | } |
| | | 208 | | |
| | | 209 | | void InitializeConnection(ServiceBusConnectionStringBuilder builder) |
| | | 210 | | { |
| | 26 | 211 | | this.Endpoint = new Uri(builder.Endpoint); |
| | | 212 | | |
| | 26 | 213 | | if (builder.SasToken != null) |
| | | 214 | | { |
| | 0 | 215 | | this.TokenProvider = new SharedAccessSignatureTokenProvider(builder.SasToken); |
| | | 216 | | } |
| | 26 | 217 | | else if (builder.SasKeyName != null || builder.SasKey != null) |
| | | 218 | | { |
| | 14 | 219 | | this.TokenProvider = new SharedAccessSignatureTokenProvider(builder.SasKeyName, builder.SasKey); |
| | | 220 | | } |
| | 12 | 221 | | else if (builder.Authentication.Equals(ServiceBusConnectionStringBuilder.AuthenticationType.ManagedIdentity) |
| | | 222 | | { |
| | 6 | 223 | | this.TokenProvider = new ManagedIdentityTokenProvider(); |
| | | 224 | | } |
| | | 225 | | |
| | 26 | 226 | | this.OperationTimeout = builder.OperationTimeout; |
| | 26 | 227 | | this.TransportType = builder.TransportType; |
| | 26 | 228 | | this.ConnectionManager = new FaultTolerantAmqpObject<AmqpConnection>(this.CreateConnectionAsync, CloseConnec |
| | 26 | 229 | | this.TransactionController = new FaultTolerantAmqpObject<Controller>(this.CreateControllerAsync, CloseContro |
| | 26 | 230 | | } |
| | | 231 | | |
| | | 232 | | static void CloseConnection(AmqpConnection connection) |
| | | 233 | | { |
| | 0 | 234 | | MessagingEventSource.Log.AmqpConnectionClosed(connection); |
| | 0 | 235 | | connection.SafeClose(); |
| | 0 | 236 | | } |
| | | 237 | | |
| | | 238 | | static void CloseController(Controller controller) |
| | | 239 | | { |
| | 0 | 240 | | controller.Close(); |
| | 0 | 241 | | } |
| | | 242 | | |
| | | 243 | | async Task<AmqpConnection> CreateConnectionAsync(TimeSpan timeout) |
| | | 244 | | { |
| | 0 | 245 | | var hostName = this.Endpoint.Host; |
| | | 246 | | |
| | 0 | 247 | | var timeoutHelper = new TimeoutHelper(timeout, true); |
| | 0 | 248 | | var amqpSettings = AmqpConnectionHelper.CreateAmqpSettings( |
| | 0 | 249 | | amqpVersion: AmqpVersion, |
| | 0 | 250 | | useSslStreamSecurity: true, |
| | 0 | 251 | | hasTokenProvider: true, |
| | 0 | 252 | | useWebSockets: TransportType == TransportType.AmqpWebSockets); |
| | | 253 | | |
| | 0 | 254 | | var transportSettings = CreateTransportSettings(); |
| | 0 | 255 | | var amqpTransportInitiator = new AmqpTransportInitiator(amqpSettings, transportSettings); |
| | 0 | 256 | | var transport = await amqpTransportInitiator.ConnectTaskAsync(timeoutHelper.RemainingTime()).ConfigureAwait( |
| | | 257 | | |
| | 0 | 258 | | var containerId = Guid.NewGuid().ToString(); |
| | 0 | 259 | | var amqpConnectionSettings = AmqpConnectionHelper.CreateAmqpConnectionSettings(AmqpConstants.DefaultMaxFrame |
| | 0 | 260 | | var connection = new AmqpConnection(transport, amqpSettings, amqpConnectionSettings); |
| | 0 | 261 | | await connection.OpenAsync(timeoutHelper.RemainingTime()).ConfigureAwait(false); |
| | | 262 | | |
| | | 263 | | // Always create the CBS Link + Session |
| | 0 | 264 | | var cbsLink = new AmqpCbsLink(connection); |
| | 0 | 265 | | if (connection.Extensions.Find<AmqpCbsLink>() == null) |
| | | 266 | | { |
| | 0 | 267 | | connection.Extensions.Add(cbsLink); |
| | | 268 | | } |
| | | 269 | | |
| | 0 | 270 | | MessagingEventSource.Log.AmqpConnectionCreated(hostName, connection); |
| | | 271 | | |
| | 0 | 272 | | return connection; |
| | 0 | 273 | | } |
| | | 274 | | |
| | | 275 | | async Task<Controller> CreateControllerAsync(TimeSpan timeout) |
| | | 276 | | { |
| | 0 | 277 | | var timeoutHelper = new TimeoutHelper(timeout, true); |
| | 0 | 278 | | var connection = await this.ConnectionManager.GetOrCreateAsync(timeoutHelper.RemainingTime()).ConfigureAwait |
| | | 279 | | |
| | 0 | 280 | | var sessionSettings = new AmqpSessionSettings { Properties = new Fields() }; |
| | 0 | 281 | | AmqpSession amqpSession = null; |
| | | 282 | | Controller controller; |
| | | 283 | | |
| | | 284 | | try |
| | | 285 | | { |
| | 0 | 286 | | amqpSession = connection.CreateSession(sessionSettings); |
| | 0 | 287 | | await amqpSession.OpenAsync(timeoutHelper.RemainingTime()).ConfigureAwait(false); |
| | | 288 | | |
| | 0 | 289 | | controller = new Controller(amqpSession, timeoutHelper.RemainingTime()); |
| | 0 | 290 | | await controller.OpenAsync(timeoutHelper.RemainingTime()).ConfigureAwait(false); |
| | 0 | 291 | | } |
| | 0 | 292 | | catch (Exception exception) |
| | | 293 | | { |
| | 0 | 294 | | if (amqpSession != null) |
| | | 295 | | { |
| | 0 | 296 | | await amqpSession.CloseAsync(timeout).ConfigureAwait(false); |
| | | 297 | | } |
| | | 298 | | |
| | 0 | 299 | | MessagingEventSource.Log.AmqpCreateControllerException(this.ConnectionManager.ToString(), exception); |
| | 0 | 300 | | throw; |
| | 0 | 301 | | } |
| | | 302 | | |
| | 0 | 303 | | return controller; |
| | 0 | 304 | | } |
| | | 305 | | |
| | | 306 | | TransportSettings CreateTransportSettings() |
| | | 307 | | { |
| | 0 | 308 | | var hostName = this.Endpoint.Host; |
| | 0 | 309 | | var networkHost = this.Endpoint.Host; |
| | 0 | 310 | | var port = this.Endpoint.Port; |
| | | 311 | | |
| | 0 | 312 | | if (TransportType == TransportType.AmqpWebSockets) |
| | | 313 | | { |
| | 0 | 314 | | return AmqpConnectionHelper.CreateWebSocketTransportSettings( |
| | 0 | 315 | | networkHost: networkHost, |
| | 0 | 316 | | hostName: hostName, |
| | 0 | 317 | | port: port, |
| | 0 | 318 | | proxy: WebRequest.DefaultWebProxy); |
| | | 319 | | } |
| | | 320 | | |
| | 0 | 321 | | return AmqpConnectionHelper.CreateTcpTransportSettings( |
| | 0 | 322 | | networkHost: networkHost, |
| | 0 | 323 | | hostName: hostName, |
| | 0 | 324 | | port: port, |
| | 0 | 325 | | useSslStreamSecurity: true); |
| | | 326 | | } |
| | | 327 | | |
| | | 328 | | } |
| | | 329 | | } |