< Summary

Information
Class: IceRpc.ConnectionCache
Assembly: IceRpc
File(s): /home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc/ConnectionCache.cs
Tag: 1321_24790053727
Line coverage
71%
Covered lines: 263
Uncovered lines: 107
Coverable lines: 370
Total lines: 676
Line coverage: 71%
Branch coverage
67%
Covered branches: 58
Total branches: 86
Branch coverage: 67.4%
Method coverage
90%
Covered methods: 18
Fully covered methods: 4
Total methods: 20
Method coverage: 90%
Full method coverage: 20%

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
.ctor()100%210%
DisposeAsync()100%66100%
PerformDisposeAsync()100%2278.57%
InvokeAsync(...)37.5%14854.16%
PerformInvokeAsync()66.66%12643.75%
ShutdownAsync(...)75%4485.71%
PerformShutdownAsync()50%7440%
CreateConnectTask()50%10866.66%
DisposePendingConnectionAsync()75%5461.11%
ShutdownWhenRequestedAsync()100%11100%
GetActiveConnectionAsync()64.28%191470.9%
RemoveFromActiveAsync(...)50%4481.81%
ShutdownAndDisposeConnectionAsync()75%4482.35%
TryGetActiveConnection(...)80%101084.61%
get_Current()75%44100%
get_Count()100%210%
get_CurrentIndex()100%11100%
MoveNext()83.33%6686.66%
.ctor(...)50%2272.72%

File(s)

/home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc/ConnectionCache.cs

#LineLine coverage
 1// Copyright (c) ZeroC, Inc.
 2
 3using IceRpc.Features;
 4using IceRpc.Transports;
 5using Microsoft.Extensions.Logging;
 6using Microsoft.Extensions.Logging.Abstractions;
 7using System.Diagnostics;
 8using System.Diagnostics.CodeAnalysis;
 9using System.Runtime.ExceptionServices;
 10
 11namespace IceRpc;
 12
 13/// <summary>Represents an invoker that routes outgoing requests to connections it manages.</summary>
 14/// <remarks><para>The connection cache routes requests based on the request's <see cref="IServerAddressFeature" />
 15/// feature or the server addresses of the request's target service.</para>
 16/// <para>The connection cache keeps at most one active connection per server address.</para></remarks>
 17public sealed class ConnectionCache : IInvoker, IAsyncDisposable
 18{
 19    // Connected connections.
 1220    private readonly Dictionary<ServerAddress, IProtocolConnection> _activeConnections =
 1221        new(ServerAddressComparer.OptionalTransport);
 22
 23    private readonly IClientProtocolConnectionFactory _connectionFactory;
 24
 25    private readonly TimeSpan _connectTimeout;
 26
 27    // A detached connection is a protocol connection that is connecting, shutting down or being disposed. Both
 28    // ShutdownAsync and DisposeAsync wait for detached connections to reach 0 using _detachedConnectionsTcs. Such a
 29    // connection is "detached" because it's not in _activeConnections.
 30    private int _detachedConnectionCount;
 31
 1232    private readonly TaskCompletionSource _detachedConnectionsTcs = new();
 33
 34    // A cancellation token source that is canceled when DisposeAsync is called.
 1235    private readonly CancellationTokenSource _disposedCts = new();
 36
 37    private Task? _disposeTask;
 38
 1239    private readonly Lock _mutex = new();
 40
 41    // New connections in the process of connecting.
 1242    private readonly Dictionary<ServerAddress, (IProtocolConnection Connection, Task ConnectTask)> _pendingConnections =
 1243        new(ServerAddressComparer.OptionalTransport);
 44
 45    private readonly bool _preferExistingConnection;
 46
 47    private Task? _shutdownTask;
 48
 49    private readonly TimeSpan _shutdownTimeout;
 50
 51    /// <summary>Constructs a connection cache.</summary>
 52    /// <param name="options">The connection cache options.</param>
 53    /// <param name="duplexClientTransport">The duplex client transport. <see langword="null" /> is equivalent to <see
 54    /// cref="IDuplexClientTransport.Default" />.</param>
 55    /// <param name="multiplexedClientTransport">The multiplexed client transport. <see langword="null" /> is equivalent
 56    /// to <see cref="IMultiplexedClientTransport.Default" />.</param>
 57    /// <param name="logger">The logger. <see langword="null" /> is equivalent to <see cref="NullLogger.Instance"
 58    /// />.</param>
 1259    public ConnectionCache(
 1260        ConnectionCacheOptions options,
 1261        IDuplexClientTransport? duplexClientTransport = null,
 1262        IMultiplexedClientTransport? multiplexedClientTransport = null,
 1263        ILogger? logger = null)
 1264    {
 1265        _connectionFactory = new ClientProtocolConnectionFactory(
 1266            options.ConnectionOptions,
 1267            options.ConnectTimeout,
 1268            options.ClientAuthenticationOptions,
 1269            duplexClientTransport,
 1270            multiplexedClientTransport,
 1271            logger);
 72
 1273        _connectTimeout = options.ConnectTimeout;
 1274        _shutdownTimeout = options.ShutdownTimeout;
 75
 1276        _preferExistingConnection = options.PreferExistingConnection;
 1277    }
 78
 79    /// <summary>Constructs a connection cache using the default options.</summary>
 80    public ConnectionCache()
 081        : this(new ConnectionCacheOptions())
 082    {
 083    }
 84
 85    /// <summary>Releases all resources allocated by the cache. The cache disposes all the connections it
 86    /// created.</summary>
 87    /// <returns>A value task that completes when the disposal of all connections created by this cache has completed.
 88    /// This includes connections that were active when this method is called and connections whose disposal was
 89    /// initiated prior to this call.</returns>
 90    /// <remarks>The disposal of an underlying connection of the cache  aborts invocations, cancels dispatches and
 91    /// disposes the underlying transport connection without waiting for the peer. To wait for invocations and
 92    /// dispatches to complete, call <see cref="ShutdownAsync" /> first. If the configured dispatcher does not complete
 93    /// promptly when its cancellation token is canceled, the disposal can hang.</remarks>
 94    public ValueTask DisposeAsync()
 1395    {
 96        lock (_mutex)
 1397        {
 1398            if (_disposeTask is null)
 1299            {
 12100                _shutdownTask ??= Task.CompletedTask;
 12101                if (_detachedConnectionCount == 0)
 11102                {
 11103                    _ = _detachedConnectionsTcs.TrySetResult();
 11104                }
 105
 12106                _disposeTask = PerformDisposeAsync();
 12107            }
 13108            return new(_disposeTask);
 109        }
 110
 111        async Task PerformDisposeAsync()
 12112        {
 12113            await Task.Yield(); // exit mutex lock
 114
 12115            _disposedCts.Cancel();
 116
 117            // Wait for shutdown before disposing connections.
 118            try
 12119            {
 12120                await _shutdownTask.ConfigureAwait(false);
 12121            }
 0122            catch
 0123            {
 124                // Ignore exceptions.
 0125            }
 126
 127            // Since a pending connection is "detached", it's disposed via the connectTask, not directly by this method.
 12128            await Task.WhenAll(
 8129                _activeConnections.Values.Select(connection => connection.DisposeAsync().AsTask())
 12130                    .Append(_detachedConnectionsTcs.Task)).ConfigureAwait(false);
 131
 12132            _disposedCts.Dispose();
 12133        }
 13134    }
 135
 136    /// <summary>Sends an outgoing request and returns the corresponding incoming response.</summary>
 137    /// <param name="request">The outgoing request being sent.</param>
 138    /// <param name="cancellationToken">A cancellation token that receives the cancellation requests.</param>
 139    /// <returns>The corresponding <see cref="IncomingResponse" />.</returns>
 140    /// <exception cref="InvalidOperationException">Thrown if no <see cref="IServerAddressFeature" /> feature is set and
 141    /// the request's service address has no server addresses.</exception>
 142    /// <exception cref="IceRpcException">Thrown with <see cref="IceRpcError.InvocationRefused" /> if the connection
 143    /// cache is shutdown, or with <see cref="IceRpcError.NoConnection" /> if the request's
 144    /// <see cref="IServerAddressFeature" /> feature has no server addresses.</exception>
 145    /// <exception cref="ObjectDisposedException">Thrown if this connection cache is disposed.</exception>
 146    /// <remarks><para>If the request <see cref="IServerAddressFeature" /> feature is not set, the cache sets it from
 147    /// the server addresses of the target service.</para>
 148    /// <para>It then looks for an active connection. The <see cref="ConnectionCacheOptions.PreferExistingConnection" />
 149    /// property influences how the cache selects this active connection. If no active connection can be found, the
 150    /// cache creates a new connection to one of the server addresses from the <see cref="IServerAddressFeature" />
 151    /// feature.</para>
 152    /// <para>If the connection establishment to <see cref="IServerAddressFeature.ServerAddress" /> fails, <see
 153    /// cref="IServerAddressFeature.ServerAddress" /> is appended at the end of <see
 154    /// cref="IServerAddressFeature.AltServerAddresses" /> and the first address from <see
 155    /// cref="IServerAddressFeature.AltServerAddresses" /> replaces <see cref="IServerAddressFeature.ServerAddress" />.
 156    /// The cache tries again to find or establish a connection to <see cref="IServerAddressFeature.ServerAddress" />.
 157    /// If unsuccessful, the cache repeats this process until success or until it tried all the addresses. If all the
 158    /// attempts fail, this method throws the exception from the last attempt.</para></remarks>
 159    public Task<IncomingResponse> InvokeAsync(OutgoingRequest request, CancellationToken cancellationToken)
 94160    {
 94161        if (request.Features.Get<IServerAddressFeature>() is IServerAddressFeature serverAddressFeature)
 0162        {
 0163            if (serverAddressFeature.ServerAddress is null)
 0164            {
 0165                throw new IceRpcException(
 0166                    IceRpcError.NoConnection,
 0167                    $"Could not invoke '{request.Operation}' on '{request.ServiceAddress}': tried all server addresses w
 168            }
 0169        }
 170        else
 94171        {
 94172            if (request.ServiceAddress.ServerAddress is null)
 0173            {
 0174                throw new InvalidOperationException("Cannot send a request to a service without a server address.");
 175            }
 176
 94177            serverAddressFeature = new ServerAddressFeature(request.ServiceAddress);
 94178            request.Features = request.Features.With(serverAddressFeature);
 94179        }
 180
 181        lock (_mutex)
 94182        {
 94183            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 184
 94185            if (_shutdownTask is not null)
 0186            {
 0187                throw new IceRpcException(IceRpcError.InvocationRefused, "The connection cache was shut down.");
 188            }
 94189        }
 190
 94191        return PerformInvokeAsync();
 192
 193        async Task<IncomingResponse> PerformInvokeAsync()
 94194        {
 94195            Debug.Assert(serverAddressFeature.ServerAddress is not null);
 196
 197            // When InvokeAsync (or ConnectAsync) throws an IceRpcException(InvocationRefused) we retry unless the
 198            // cache is being shutdown.
 94199            while (true)
 94200            {
 94201                IProtocolConnection? connection = null;
 94202                if (_preferExistingConnection)
 92203                {
 92204                    _ = TryGetActiveConnection(serverAddressFeature, out connection);
 92205                }
 94206                connection ??= await GetActiveConnectionAsync(serverAddressFeature, cancellationToken)
 94207                    .ConfigureAwait(false);
 208
 209                try
 94210                {
 94211                    return await connection.InvokeAsync(request, cancellationToken).ConfigureAwait(false);
 212                }
 0213                catch (ObjectDisposedException)
 0214                {
 215                    // This can occasionally happen if we find a connection that was just closed and then automatically
 216                    // disposed by this connection cache.
 0217                }
 0218                catch (IceRpcException exception) when (exception.IceRpcError == IceRpcError.InvocationRefused)
 0219                {
 220                    // The connection is refusing new invocations.
 0221                }
 0222                catch (IceRpcException exception) when (exception.IceRpcError == IceRpcError.OperationAborted)
 0223                {
 224                    lock (_mutex)
 0225                    {
 0226                        if (_disposeTask is null)
 0227                        {
 0228                            throw new IceRpcException(
 0229                                IceRpcError.ConnectionAborted,
 0230                                "The underlying connection was disposed while the invocation was in progress.");
 231                        }
 232                        else
 0233                        {
 0234                            throw;
 235                        }
 236                    }
 237                }
 238
 239                // Make sure connection is no longer in _activeConnection before we retry.
 0240                _ = RemoveFromActiveAsync(serverAddressFeature.ServerAddress.Value, connection);
 0241            }
 94242        }
 94243    }
 244
 245    /// <summary>Gracefully shuts down all connections created by this cache.</summary>
 246    /// <param name="cancellationToken">A cancellation token that receives the cancellation requests.</param>
 247    /// <returns>A task that completes successfully once the shutdown of all connections created by this cache has
 248    /// completed. This includes connections that were active when this method is called and connections whose shutdown
 249    /// was initiated prior to this call.</returns>
 250    /// <exception cref="InvalidOperationException">Thrown if this method is called more than once.</exception>
 251    /// <exception cref="ObjectDisposedException">Thrown if the connection cache is disposed.</exception>
 252    /// <remarks><para>The returned task can also complete with one of the following exceptions:</para>
 253    /// <list type="bullet">
 254    /// <item><description><see cref="IceRpcException" /> with error <see cref="IceRpcError.OperationAborted" /> if the
 255    /// connection cache is disposed while being shut down.</description></item>
 256    /// <item><description><see cref="OperationCanceledException" /> if cancellation was requested through the
 257    /// cancellation token.</description></item>
 258    /// <item><description><see cref="TimeoutException" /> if the shutdown timed out.</description></item>
 259    /// </list>
 260    /// </remarks>
 261    public Task ShutdownAsync(CancellationToken cancellationToken = default)
 11262    {
 263        lock (_mutex)
 11264        {
 11265            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 266
 11267            if (_shutdownTask is not null)
 0268            {
 0269                throw new InvalidOperationException("The connection cache is already shut down or shutting down.");
 270            }
 271
 11272            if (_detachedConnectionCount == 0)
 7273            {
 7274                _detachedConnectionsTcs.SetResult();
 7275            }
 276
 11277            _shutdownTask = PerformShutdownAsync();
 11278        }
 279
 11280        return _shutdownTask;
 281
 282        async Task PerformShutdownAsync()
 11283        {
 11284            await Task.Yield(); // exit mutex lock
 285
 11286            using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _disposedCts.Token);
 11287            cts.CancelAfter(_shutdownTimeout);
 288
 289            try
 11290            {
 291                // Since a pending connection is "detached", it's shutdown and disposed via the connectTask, not
 292                // directly by this method.
 293                try
 11294                {
 11295                    await Task.WhenAll(
 8296                        _activeConnections.Values.Select(connection => connection.ShutdownAsync(cts.Token))
 11297                            .Append(_detachedConnectionsTcs.Task.WaitAsync(cts.Token))).ConfigureAwait(false);
 11298                }
 0299                catch (OperationCanceledException)
 0300                {
 0301                    throw;
 302                }
 0303                catch
 0304                {
 305                    // Ignore other connection shutdown failures.
 306
 307                    // Throw OperationCanceledException if this WhenAll exception is hiding an OCE.
 0308                    cts.Token.ThrowIfCancellationRequested();
 0309                }
 11310            }
 0311            catch (OperationCanceledException)
 0312            {
 0313                cancellationToken.ThrowIfCancellationRequested();
 314
 0315                if (_disposedCts.IsCancellationRequested)
 0316                {
 0317                    throw new IceRpcException(
 0318                        IceRpcError.OperationAborted,
 0319                        "The shutdown was aborted because the connection cache was disposed.");
 320                }
 321                else
 0322                {
 0323                    throw new TimeoutException(
 0324                        $"The connection cache shut down timed out after {_shutdownTimeout.TotalSeconds} s.");
 325                }
 326            }
 11327        }
 11328    }
 329
 330    private async Task CreateConnectTask(IProtocolConnection connection, ServerAddress serverAddress)
 20331    {
 20332        await Task.Yield(); // exit mutex lock
 333
 334        // This task "owns" a detachedConnectionCount and as a result _disposedCts can't be disposed.
 20335        using var cts = CancellationTokenSource.CreateLinkedTokenSource(_disposedCts.Token);
 20336        cts.CancelAfter(_connectTimeout);
 337
 338        Task shutdownRequested;
 20339        Task? connectTask = null;
 340
 341        try
 20342        {
 343            try
 20344            {
 20345                (_, shutdownRequested) = await connection.ConnectAsync(cts.Token).ConfigureAwait(false);
 17346            }
 0347            catch (OperationCanceledException)
 0348            {
 0349                if (_disposedCts.IsCancellationRequested)
 0350                {
 0351                    throw new IceRpcException(
 0352                        IceRpcError.OperationAborted,
 0353                        "The connection establishment was aborted because the connection cache was disposed.");
 354                }
 355                else
 0356                {
 0357                    throw new TimeoutException(
 0358                        $"The connection establishment timed out after {_connectTimeout.TotalSeconds} s.");
 359                }
 360            }
 17361        }
 3362        catch
 3363        {
 364            lock (_mutex)
 3365            {
 366                // connectTask is executing this method and about to throw.
 3367                connectTask = _pendingConnections[serverAddress].ConnectTask;
 3368                _pendingConnections.Remove(serverAddress);
 3369            }
 370
 3371            _ = DisposePendingConnectionAsync(connection, connectTask);
 3372            throw;
 373        }
 374
 375        lock (_mutex)
 17376        {
 17377            if (_shutdownTask is null)
 17378            {
 379                // the connection is now "attached" in _activeConnections
 17380                _activeConnections.Add(serverAddress, connection);
 17381                _detachedConnectionCount--;
 17382            }
 383            else
 0384            {
 0385                connectTask = _pendingConnections[serverAddress].ConnectTask;
 0386            }
 17387            bool removed = _pendingConnections.Remove(serverAddress);
 17388            Debug.Assert(removed);
 17389        }
 390
 17391        if (connectTask is null)
 17392        {
 17393            _ = ShutdownWhenRequestedAsync(connection, serverAddress, shutdownRequested);
 17394        }
 395        else
 0396        {
 397            // As soon as this method completes successfully, we shut down then dispose the connection.
 0398            _ = DisposePendingConnectionAsync(connection, connectTask);
 0399        }
 400
 401        async Task DisposePendingConnectionAsync(IProtocolConnection connection, Task connectTask)
 3402        {
 403            try
 3404            {
 3405                await connectTask.ConfigureAwait(false);
 406
 407                // Since we own a detachedConnectionCount, _disposedCts is not disposed.
 0408                using var cts = CancellationTokenSource.CreateLinkedTokenSource(_disposedCts.Token);
 0409                cts.CancelAfter(_shutdownTimeout);
 0410                await connection.ShutdownAsync(cts.Token).ConfigureAwait(false);
 0411            }
 3412            catch
 3413            {
 414                // Observe and ignore exceptions.
 3415            }
 416
 3417            await connection.DisposeAsync().ConfigureAwait(false);
 418
 419            lock (_mutex)
 3420            {
 3421                if (--_detachedConnectionCount == 0 && _shutdownTask is not null)
 0422                {
 0423                    _detachedConnectionsTcs.SetResult();
 0424                }
 3425            }
 3426        }
 427
 428        async Task ShutdownWhenRequestedAsync(
 429            IProtocolConnection connection,
 430            ServerAddress serverAddress,
 431            Task shutdownRequested)
 17432        {
 17433            await shutdownRequested.ConfigureAwait(false);
 13434            await RemoveFromActiveAsync(serverAddress, connection).ConfigureAwait(false);
 13435        }
 17436    }
 437
 438    /// <summary>Gets an active connection, by creating and connecting (if necessary) a new protocol connection.
 439    /// </summary>
 440    /// <param name="serverAddressFeature">The server address feature.</param>
 441    /// <param name="cancellationToken">The cancellation token of the invocation calling this method.</param>
 442    private async Task<IProtocolConnection> GetActiveConnectionAsync(
 443        IServerAddressFeature serverAddressFeature,
 444        CancellationToken cancellationToken)
 17445    {
 17446        Debug.Assert(serverAddressFeature.ServerAddress is not null);
 17447        Exception? connectionException = null;
 448        (IProtocolConnection Connection, Task ConnectTask) pendingConnectionValue;
 17449        var enumerator = new ServerAddressEnumerator(serverAddressFeature);
 20450        while (enumerator.MoveNext())
 20451        {
 20452            ServerAddress serverAddress = enumerator.Current;
 20453            if (enumerator.CurrentIndex > 0)
 3454            {
 455                // Rotate the server addresses before each new connection attempt after the initial attempt
 3456                serverAddressFeature.RotateAddresses();
 3457            }
 458
 459            try
 20460            {
 461                lock (_mutex)
 20462                {
 20463                    if (_disposeTask is not null)
 0464                    {
 0465                        throw new IceRpcException(IceRpcError.OperationAborted, "The connection cache was disposed.");
 466                    }
 20467                    else if (_shutdownTask is not null)
 0468                    {
 0469                        throw new IceRpcException(IceRpcError.InvocationRefused, "The connection cache is shut down.");
 470                    }
 471
 20472                    if (_activeConnections.TryGetValue(serverAddress, out IProtocolConnection? connection))
 0473                    {
 0474                        return connection;
 475                    }
 476
 20477                    if (!_pendingConnections.TryGetValue(serverAddress, out pendingConnectionValue))
 20478                    {
 20479                        connection = _connectionFactory.CreateConnection(serverAddress);
 20480                        _detachedConnectionCount++;
 20481                        pendingConnectionValue = (connection, CreateConnectTask(connection, serverAddress));
 20482                        _pendingConnections.Add(serverAddress, pendingConnectionValue);
 20483                    }
 20484                }
 485                // ConnectTask itself takes care of scheduling its exception observation when it fails.
 20486                await pendingConnectionValue.ConnectTask.WaitAsync(cancellationToken).ConfigureAwait(false);
 17487                return pendingConnectionValue.Connection;
 488            }
 0489            catch (TimeoutException exception)
 0490            {
 0491                connectionException = exception;
 0492            }
 3493            catch (IceRpcException exception) when (exception.IceRpcError is
 3494                IceRpcError.ConnectionAborted or
 3495                IceRpcError.ConnectionRefused or
 3496                IceRpcError.ServerBusy or
 3497                IceRpcError.ServerUnreachable)
 3498            {
 499                // keep going unless the connection cache was disposed or shut down
 3500                connectionException = exception;
 501                lock (_mutex)
 3502                {
 3503                    if (_shutdownTask is not null)
 0504                    {
 0505                        throw;
 506                    }
 3507                }
 3508            }
 3509        }
 510
 0511        Debug.Assert(connectionException is not null);
 0512        ExceptionDispatchInfo.Throw(connectionException);
 0513        Debug.Assert(false);
 0514        throw connectionException;
 17515    }
 516
 517    /// <summary>Removes the connection from _activeConnections, and when successful, shuts down and disposes this
 518    /// connection.</summary>
 519    /// <param name="serverAddress">The server address key in _activeConnections.</param>
 520    /// <param name="connection">The connection to shutdown and dispose after removal.</param>
 521    private Task RemoveFromActiveAsync(ServerAddress serverAddress, IProtocolConnection connection)
 13522    {
 523        lock (_mutex)
 13524        {
 13525            if (_shutdownTask is null && _activeConnections.Remove(serverAddress))
 9526            {
 527                // it's now our connection.
 9528                _detachedConnectionCount++;
 9529            }
 530            else
 4531            {
 532                // Another task owns this connection
 4533                return Task.CompletedTask;
 534            }
 9535        }
 536
 9537        return ShutdownAndDisposeConnectionAsync();
 538
 539        async Task ShutdownAndDisposeConnectionAsync()
 9540        {
 541            // _disposedCts is not disposed since we own a detachedConnectionCount
 9542            using var cts = CancellationTokenSource.CreateLinkedTokenSource(_disposedCts.Token);
 9543            cts.CancelAfter(_shutdownTimeout);
 544
 545            try
 9546            {
 9547                await connection.ShutdownAsync(cts.Token).ConfigureAwait(false);
 9548            }
 0549            catch
 0550            {
 551                // Ignore connection shutdown failures
 0552            }
 553
 9554            await connection.DisposeAsync().ConfigureAwait(false);
 555
 556            lock (_mutex)
 9557            {
 9558                if (--_detachedConnectionCount == 0 && _shutdownTask is not null)
 5559                {
 5560                    _detachedConnectionsTcs.SetResult();
 5561                }
 9562            }
 9563        }
 13564    }
 565
 566    /// <summary>Tries to get an existing connection matching one of the addresses of the server address feature.
 567    /// </summary>
 568    /// <param name="serverAddressFeature">The server address feature.</param>
 569    /// <param name="connection">When this method returns <see langword="true" />, this argument contains an active
 570    /// connection; otherwise, it is set to <see langword="null" />.</param>
 571    /// <returns><see langword="true" /> when an active connection matching any of the addresses of the server address
 572    /// feature is found; otherwise, <see langword="false"/>.</returns>
 573    private bool TryGetActiveConnection(
 574        IServerAddressFeature serverAddressFeature,
 575        [NotNullWhen(true)] out IProtocolConnection? connection)
 92576    {
 577        lock (_mutex)
 92578        {
 92579            connection = null;
 92580            if (_disposeTask is not null)
 0581            {
 0582                throw new IceRpcException(IceRpcError.OperationAborted, "The connection cache was disposed.");
 583            }
 584
 92585            if (_shutdownTask is not null)
 0586            {
 0587                throw new IceRpcException(IceRpcError.InvocationRefused, "The connection cache was shut down.");
 588            }
 589
 92590            var enumerator = new ServerAddressEnumerator(serverAddressFeature);
 112591            while (enumerator.MoveNext())
 97592            {
 97593                ServerAddress serverAddress = enumerator.Current;
 97594                if (_activeConnections.TryGetValue(serverAddress, out connection))
 77595                {
 77596                    if (enumerator.CurrentIndex > 0)
 1597                    {
 598                        // This altServerAddress becomes the main server address, and the existing main
 599                        // server address becomes the first alt server address.
 1600                        serverAddressFeature.AltServerAddresses = serverAddressFeature.AltServerAddresses
 1601                            .RemoveAt(enumerator.CurrentIndex - 1)
 1602                            .Insert(0, serverAddressFeature.ServerAddress!.Value);
 1603                        serverAddressFeature.ServerAddress = serverAddress;
 1604                    }
 77605                    return true;
 606                }
 20607            }
 15608            return false;
 609        }
 92610    }
 611
 612    /// <summary>A helper struct that implements an enumerator that allows iterating the addresses of an
 613    /// <see cref="IServerAddressFeature" /> without allocations.</summary>
 614    private struct ServerAddressEnumerator
 615    {
 616        internal readonly ServerAddress Current
 617        {
 618            get
 117619            {
 117620                Debug.Assert(CurrentIndex >= 0 && CurrentIndex <= _altServerAddresses.Count);
 117621                if (CurrentIndex == 0)
 109622                {
 109623                    Debug.Assert(_mainServerAddress is not null);
 109624                    return _mainServerAddress.Value;
 625                }
 626                else
 8627                {
 8628                    return _altServerAddresses[CurrentIndex - 1];
 629                }
 117630            }
 631        }
 632
 0633        internal int Count { get; }
 634
 955635        internal int CurrentIndex { get; private set; } = -1;
 636
 637        private readonly ServerAddress? _mainServerAddress;
 638        private readonly IList<ServerAddress> _altServerAddresses;
 639
 640        internal bool MoveNext()
 132641        {
 132642            if (CurrentIndex == -1)
 109643            {
 109644                if (_mainServerAddress is not null)
 109645                {
 109646                    CurrentIndex++;
 109647                    return true;
 648                }
 649                else
 0650                {
 0651                    return false;
 652                }
 653            }
 23654            else if (CurrentIndex < _altServerAddresses.Count)
 8655            {
 8656                CurrentIndex++;
 8657                return true;
 658            }
 15659            return false;
 132660        }
 661
 662        internal ServerAddressEnumerator(IServerAddressFeature serverAddressFeature)
 109663        {
 109664            _mainServerAddress = serverAddressFeature.ServerAddress;
 109665            _altServerAddresses = serverAddressFeature.AltServerAddresses;
 109666            if (_mainServerAddress is null)
 0667            {
 0668                Count = 0;
 0669            }
 670            else
 109671            {
 109672                Count = _altServerAddresses.Count + 1;
 109673            }
 109674        }
 675    }
 676}
Generated by: ReportGenerator 5.5.6.0
04/22/2026 - 16:32:34
GitHub | reportgenerator.io

Methods/Properties

.ctor(IceRpc.ConnectionCacheOptions,IceRpc.Transports.IDuplexClientTransport,IceRpc.Transports.IMultiplexedClientTransport,Microsoft.Extensions.Logging.ILogger)
.ctor()
DisposeAsync()
PerformDisposeAsync()
InvokeAsync(IceRpc.OutgoingRequest,System.Threading.CancellationToken)
PerformInvokeAsync()
ShutdownAsync(System.Threading.CancellationToken)
PerformShutdownAsync()
CreateConnectTask()
DisposePendingConnectionAsync()
ShutdownWhenRequestedAsync()
GetActiveConnectionAsync()
RemoveFromActiveAsync(IceRpc.ServerAddress,IceRpc.IProtocolConnection)
ShutdownAndDisposeConnectionAsync()
TryGetActiveConnection(IceRpc.Features.IServerAddressFeature,IceRpc.IProtocolConnection&)
get_Current()
get_Count()
get_CurrentIndex()
MoveNext()
.ctor(IceRpc.Features.IServerAddressFeature)