< Summary

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

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
.ctor()100%210%
DisposeAsync()100%66100%
PerformDisposeAsync()100%2.04278.57%
InvokeAsync(...)37.5%14.16854.16%
PerformInvokeAsync()66.66%12.41643.75%
ShutdownAsync(...)75%4.05485.71%
PerformShutdownAsync()50%7.46440%
CreateConnectTask()50%10.37866.66%
DisposePendingConnectionAsync()75%4.94461.11%
ShutdownWhenRequestedAsync()100%11100%
GetActiveConnectionAsync()64.28%18.831470.9%
RemoveFromActiveAsync(...)50%4.1481.81%
ShutdownAndDisposeConnectionAsync()75%4.09482.35%
TryGetActiveConnection(...)80%10.361084.61%
get_Current()75%44100%
get_Count()100%210%
get_CurrentIndex()100%11100%
MoveNext()83.33%6.09686.66%
.ctor(...)50%2.08272.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 one of the following error:
 143    /// <list type="bullet"><item><term><see cref="IceRpcError.InvocationRefused" /></term><description>This error
 144    /// indicates that the connection cache is shutdown.</description></item>
 145    /// <item><term><see cref="IceRpcError.NoConnection" /></term><description>This error indicates that the request
 146    /// <see cref="IServerAddressFeature" /> feature has no server addresses.</description></item></list>.
 147    /// </exception>
 148    /// <exception cref="ObjectDisposedException">Thrown if this connection cache is disposed.</exception>
 149    /// <remarks><para>If the request <see cref="IServerAddressFeature" /> feature is not set, the cache sets it from
 150    /// the server addresses of the target service.</para>
 151    /// <para>It then looks for an active connection. The <see cref="ConnectionCacheOptions.PreferExistingConnection" />
 152    /// property influences how the cache selects this active connection. If no active connection can be found, the
 153    /// cache creates a new connection to one of the server addresses from the <see cref="IServerAddressFeature" />
 154    /// feature.</para>
 155    /// <para>If the connection establishment to <see cref="IServerAddressFeature.ServerAddress" /> fails, <see
 156    /// cref="IServerAddressFeature.ServerAddress" /> is appended at the end of <see
 157    /// cref="IServerAddressFeature.AltServerAddresses" /> and the first address from <see
 158    /// cref="IServerAddressFeature.AltServerAddresses" /> replaces <see cref="IServerAddressFeature.ServerAddress" />.
 159    /// The cache tries again to find or establish a connection to <see cref="IServerAddressFeature.ServerAddress" />.
 160    /// If unsuccessful, the cache repeats this process until success or until it tried all the addresses. If all the
 161    /// attempts fail, this method throws the exception from the last attempt.</para></remarks>
 162    public Task<IncomingResponse> InvokeAsync(OutgoingRequest request, CancellationToken cancellationToken)
 94163    {
 94164        if (request.Features.Get<IServerAddressFeature>() is IServerAddressFeature serverAddressFeature)
 0165        {
 0166            if (serverAddressFeature.ServerAddress is null)
 0167            {
 0168                throw new IceRpcException(
 0169                    IceRpcError.NoConnection,
 0170                    $"Could not invoke '{request.Operation}' on '{request.ServiceAddress}': tried all server addresses w
 171            }
 0172        }
 173        else
 94174        {
 94175            if (request.ServiceAddress.ServerAddress is null)
 0176            {
 0177                throw new InvalidOperationException("Cannot send a request to a service without a server address.");
 178            }
 179
 94180            serverAddressFeature = new ServerAddressFeature(request.ServiceAddress);
 94181            request.Features = request.Features.With(serverAddressFeature);
 94182        }
 183
 184        lock (_mutex)
 94185        {
 94186            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 187
 94188            if (_shutdownTask is not null)
 0189            {
 0190                throw new IceRpcException(IceRpcError.InvocationRefused, "The connection cache was shut down.");
 191            }
 94192        }
 193
 94194        return PerformInvokeAsync();
 195
 196        async Task<IncomingResponse> PerformInvokeAsync()
 94197        {
 94198            Debug.Assert(serverAddressFeature.ServerAddress is not null);
 199
 200            // When InvokeAsync (or ConnectAsync) throws an IceRpcException(InvocationRefused) we retry unless the
 201            // cache is being shutdown.
 94202            while (true)
 94203            {
 94204                IProtocolConnection? connection = null;
 94205                if (_preferExistingConnection)
 92206                {
 92207                    _ = TryGetActiveConnection(serverAddressFeature, out connection);
 92208                }
 94209                connection ??= await GetActiveConnectionAsync(serverAddressFeature, cancellationToken)
 94210                    .ConfigureAwait(false);
 211
 212                try
 94213                {
 94214                    return await connection.InvokeAsync(request, cancellationToken).ConfigureAwait(false);
 215                }
 0216                catch (ObjectDisposedException)
 0217                {
 218                    // This can occasionally happen if we find a connection that was just closed and then automatically
 219                    // disposed by this connection cache.
 0220                }
 0221                catch (IceRpcException exception) when (exception.IceRpcError == IceRpcError.InvocationRefused)
 0222                {
 223                    // The connection is refusing new invocations.
 0224                }
 0225                catch (IceRpcException exception) when (exception.IceRpcError == IceRpcError.OperationAborted)
 0226                {
 227                    lock (_mutex)
 0228                    {
 0229                        if (_disposeTask is null)
 0230                        {
 0231                            throw new IceRpcException(
 0232                                IceRpcError.ConnectionAborted,
 0233                                "The underlying connection was disposed while the invocation was in progress.");
 234                        }
 235                        else
 0236                        {
 0237                            throw;
 238                        }
 239                    }
 240                }
 241
 242                // Make sure connection is no longer in _activeConnection before we retry.
 0243                _ = RemoveFromActiveAsync(serverAddressFeature.ServerAddress.Value, connection);
 0244            }
 94245        }
 94246    }
 247
 248    /// <summary>Gracefully shuts down all connections created by this cache.</summary>
 249    /// <param name="cancellationToken">A cancellation token that receives the cancellation requests.</param>
 250    /// <returns>A task that completes successfully once the shutdown of all connections created by this cache has
 251    /// completed. This includes connections that were active when this method is called and connections whose shutdown
 252    /// was initiated prior to this call. This task can also complete with one of the following exceptions:
 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    /// </returns>
 261    /// <exception cref="InvalidOperationException">Thrown if this method is called more than once.</exception>
 262    /// <exception cref="ObjectDisposedException">Thrown if the connection cache is disposed.</exception>
 263    public Task ShutdownAsync(CancellationToken cancellationToken = default)
 11264    {
 265        lock (_mutex)
 11266        {
 11267            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 268
 11269            if (_shutdownTask is not null)
 0270            {
 0271                throw new InvalidOperationException("The connection cache is already shut down or shutting down.");
 272            }
 273
 11274            if (_detachedConnectionCount == 0)
 6275            {
 6276                _detachedConnectionsTcs.SetResult();
 6277            }
 278
 11279            _shutdownTask = PerformShutdownAsync();
 11280        }
 281
 11282        return _shutdownTask;
 283
 284        async Task PerformShutdownAsync()
 11285        {
 11286            await Task.Yield(); // exit mutex lock
 287
 11288            using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _disposedCts.Token);
 11289            cts.CancelAfter(_shutdownTimeout);
 290
 291            try
 11292            {
 293                // Since a pending connection is "detached", it's shutdown and disposed via the connectTask, not
 294                // directly by this method.
 295                try
 11296                {
 11297                    await Task.WhenAll(
 8298                        _activeConnections.Values.Select(connection => connection.ShutdownAsync(cts.Token))
 11299                            .Append(_detachedConnectionsTcs.Task.WaitAsync(cts.Token))).ConfigureAwait(false);
 11300                }
 0301                catch (OperationCanceledException)
 0302                {
 0303                    throw;
 304                }
 0305                catch
 0306                {
 307                    // Ignore other connection shutdown failures.
 308
 309                    // Throw OperationCanceledException if this WhenAll exception is hiding an OCE.
 0310                    cts.Token.ThrowIfCancellationRequested();
 0311                }
 11312            }
 0313            catch (OperationCanceledException)
 0314            {
 0315                cancellationToken.ThrowIfCancellationRequested();
 316
 0317                if (_disposedCts.IsCancellationRequested)
 0318                {
 0319                    throw new IceRpcException(
 0320                        IceRpcError.OperationAborted,
 0321                        "The shutdown was aborted because the connection cache was disposed.");
 322                }
 323                else
 0324                {
 0325                    throw new TimeoutException(
 0326                        $"The connection cache shut down timed out after {_shutdownTimeout.TotalSeconds} s.");
 327                }
 328            }
 11329        }
 11330    }
 331
 332    private async Task CreateConnectTask(IProtocolConnection connection, ServerAddress serverAddress)
 20333    {
 20334        await Task.Yield(); // exit mutex lock
 335
 336        // This task "owns" a detachedConnectionCount and as a result _disposedCts can't be disposed.
 20337        using var cts = CancellationTokenSource.CreateLinkedTokenSource(_disposedCts.Token);
 20338        cts.CancelAfter(_connectTimeout);
 339
 340        Task shutdownRequested;
 20341        Task? connectTask = null;
 342
 343        try
 20344        {
 345            try
 20346            {
 20347                (_, shutdownRequested) = await connection.ConnectAsync(cts.Token).ConfigureAwait(false);
 17348            }
 0349            catch (OperationCanceledException)
 0350            {
 0351                if (_disposedCts.IsCancellationRequested)
 0352                {
 0353                    throw new IceRpcException(
 0354                        IceRpcError.OperationAborted,
 0355                        "The connection establishment was aborted because the connection cache was disposed.");
 356                }
 357                else
 0358                {
 0359                    throw new TimeoutException(
 0360                        $"The connection establishment timed out after {_connectTimeout.TotalSeconds} s.");
 361                }
 362            }
 17363        }
 3364        catch
 3365        {
 366            lock (_mutex)
 3367            {
 368                // connectTask is executing this method and about to throw.
 3369                connectTask = _pendingConnections[serverAddress].ConnectTask;
 3370                _pendingConnections.Remove(serverAddress);
 3371            }
 372
 3373            _ = DisposePendingConnectionAsync(connection, connectTask);
 3374            throw;
 375        }
 376
 377        lock (_mutex)
 17378        {
 17379            if (_shutdownTask is null)
 17380            {
 381                // the connection is now "attached" in _activeConnections
 17382                _activeConnections.Add(serverAddress, connection);
 17383                _detachedConnectionCount--;
 17384            }
 385            else
 0386            {
 0387                connectTask = _pendingConnections[serverAddress].ConnectTask;
 0388            }
 17389            bool removed = _pendingConnections.Remove(serverAddress);
 17390            Debug.Assert(removed);
 17391        }
 392
 17393        if (connectTask is null)
 17394        {
 17395            _ = ShutdownWhenRequestedAsync(connection, serverAddress, shutdownRequested);
 17396        }
 397        else
 0398        {
 399            // As soon as this method completes successfully, we shut down then dispose the connection.
 0400            _ = DisposePendingConnectionAsync(connection, connectTask);
 0401        }
 402
 403        async Task DisposePendingConnectionAsync(IProtocolConnection connection, Task connectTask)
 3404        {
 405            try
 3406            {
 3407                await connectTask.ConfigureAwait(false);
 408
 409                // Since we own a detachedConnectionCount, _disposedCts is not disposed.
 0410                using var cts = CancellationTokenSource.CreateLinkedTokenSource(_disposedCts.Token);
 0411                cts.CancelAfter(_shutdownTimeout);
 0412                await connection.ShutdownAsync(cts.Token).ConfigureAwait(false);
 0413            }
 3414            catch
 3415            {
 416                // Observe and ignore exceptions.
 3417            }
 418
 3419            await connection.DisposeAsync().ConfigureAwait(false);
 420
 421            lock (_mutex)
 3422            {
 3423                if (--_detachedConnectionCount == 0 && _shutdownTask is not null)
 0424                {
 0425                    _detachedConnectionsTcs.SetResult();
 0426                }
 3427            }
 3428        }
 429
 430        async Task ShutdownWhenRequestedAsync(
 431            IProtocolConnection connection,
 432            ServerAddress serverAddress,
 433            Task shutdownRequested)
 17434        {
 17435            await shutdownRequested.ConfigureAwait(false);
 13436            await RemoveFromActiveAsync(serverAddress, connection).ConfigureAwait(false);
 13437        }
 17438    }
 439
 440    /// <summary>Gets an active connection, by creating and connecting (if necessary) a new protocol connection.
 441    /// </summary>
 442    /// <param name="serverAddressFeature">The server address feature.</param>
 443    /// <param name="cancellationToken">The cancellation token of the invocation calling this method.</param>
 444    private async Task<IProtocolConnection> GetActiveConnectionAsync(
 445        IServerAddressFeature serverAddressFeature,
 446        CancellationToken cancellationToken)
 17447    {
 17448        Debug.Assert(serverAddressFeature.ServerAddress is not null);
 17449        Exception? connectionException = null;
 450        (IProtocolConnection Connection, Task ConnectTask) pendingConnectionValue;
 17451        var enumerator = new ServerAddressEnumerator(serverAddressFeature);
 20452        while (enumerator.MoveNext())
 20453        {
 20454            ServerAddress serverAddress = enumerator.Current;
 20455            if (enumerator.CurrentIndex > 0)
 3456            {
 457                // Rotate the server addresses before each new connection attempt after the initial attempt
 3458                serverAddressFeature.RotateAddresses();
 3459            }
 460
 461            try
 20462            {
 463                lock (_mutex)
 20464                {
 20465                    if (_disposeTask is not null)
 0466                    {
 0467                        throw new IceRpcException(IceRpcError.OperationAborted, "The connection cache was disposed.");
 468                    }
 20469                    else if (_shutdownTask is not null)
 0470                    {
 0471                        throw new IceRpcException(IceRpcError.InvocationRefused, "The connection cache is shut down.");
 472                    }
 473
 20474                    if (_activeConnections.TryGetValue(serverAddress, out IProtocolConnection? connection))
 0475                    {
 0476                        return connection;
 477                    }
 478
 20479                    if (!_pendingConnections.TryGetValue(serverAddress, out pendingConnectionValue))
 20480                    {
 20481                        connection = _connectionFactory.CreateConnection(serverAddress);
 20482                        _detachedConnectionCount++;
 20483                        pendingConnectionValue = (connection, CreateConnectTask(connection, serverAddress));
 20484                        _pendingConnections.Add(serverAddress, pendingConnectionValue);
 20485                    }
 20486                }
 487                // ConnectTask itself takes care of scheduling its exception observation when it fails.
 20488                await pendingConnectionValue.ConnectTask.WaitAsync(cancellationToken).ConfigureAwait(false);
 17489                return pendingConnectionValue.Connection;
 490            }
 0491            catch (TimeoutException exception)
 0492            {
 0493                connectionException = exception;
 0494            }
 3495            catch (IceRpcException exception) when (exception.IceRpcError is
 3496                IceRpcError.ConnectionAborted or
 3497                IceRpcError.ConnectionRefused or
 3498                IceRpcError.ServerBusy or
 3499                IceRpcError.ServerUnreachable)
 3500            {
 501                // keep going unless the connection cache was disposed or shut down
 3502                connectionException = exception;
 503                lock (_mutex)
 3504                {
 3505                    if (_shutdownTask is not null)
 0506                    {
 0507                        throw;
 508                    }
 3509                }
 3510            }
 3511        }
 512
 0513        Debug.Assert(connectionException is not null);
 0514        ExceptionDispatchInfo.Throw(connectionException);
 0515        Debug.Assert(false);
 0516        throw connectionException;
 17517    }
 518
 519    /// <summary>Removes the connection from _activeConnections, and when successful, shuts down and disposes this
 520    /// connection.</summary>
 521    /// <param name="serverAddress">The server address key in _activeConnections.</param>
 522    /// <param name="connection">The connection to shutdown and dispose after removal.</param>
 523    private Task RemoveFromActiveAsync(ServerAddress serverAddress, IProtocolConnection connection)
 13524    {
 525        lock (_mutex)
 13526        {
 13527            if (_shutdownTask is null && _activeConnections.Remove(serverAddress))
 9528            {
 529                // it's now our connection.
 9530                _detachedConnectionCount++;
 9531            }
 532            else
 4533            {
 534                // Another task owns this connection
 4535                return Task.CompletedTask;
 536            }
 9537        }
 538
 9539        return ShutdownAndDisposeConnectionAsync();
 540
 541        async Task ShutdownAndDisposeConnectionAsync()
 9542        {
 543            // _disposedCts is not disposed since we own a detachedConnectionCount
 9544            using var cts = CancellationTokenSource.CreateLinkedTokenSource(_disposedCts.Token);
 9545            cts.CancelAfter(_shutdownTimeout);
 546
 547            try
 9548            {
 9549                await connection.ShutdownAsync(cts.Token).ConfigureAwait(false);
 9550            }
 0551            catch
 0552            {
 553                // Ignore connection shutdown failures
 0554            }
 555
 9556            await connection.DisposeAsync().ConfigureAwait(false);
 557
 558            lock (_mutex)
 9559            {
 9560                if (--_detachedConnectionCount == 0 && _shutdownTask is not null)
 6561                {
 6562                    _detachedConnectionsTcs.SetResult();
 6563                }
 9564            }
 9565        }
 13566    }
 567
 568    /// <summary>Tries to get an existing connection matching one of the addresses of the server address feature.
 569    /// </summary>
 570    /// <param name="serverAddressFeature">The server address feature.</param>
 571    /// <param name="connection">When this method returns <see langword="true" />, this argument contains an active
 572    /// connection; otherwise, it is set to <see langword="null" />.</param>
 573    /// <returns><see langword="true" /> when an active connection matching any of the addresses of the server address
 574    /// feature is found; otherwise, <see langword="false"/>.</returns>
 575    private bool TryGetActiveConnection(
 576        IServerAddressFeature serverAddressFeature,
 577        [NotNullWhen(true)] out IProtocolConnection? connection)
 92578    {
 579        lock (_mutex)
 92580        {
 92581            connection = null;
 92582            if (_disposeTask is not null)
 0583            {
 0584                throw new IceRpcException(IceRpcError.OperationAborted, "The connection cache was disposed.");
 585            }
 586
 92587            if (_shutdownTask is not null)
 0588            {
 0589                throw new IceRpcException(IceRpcError.InvocationRefused, "The connection cache was shut down.");
 590            }
 591
 92592            var enumerator = new ServerAddressEnumerator(serverAddressFeature);
 112593            while (enumerator.MoveNext())
 97594            {
 97595                ServerAddress serverAddress = enumerator.Current;
 97596                if (_activeConnections.TryGetValue(serverAddress, out connection))
 77597                {
 77598                    if (enumerator.CurrentIndex > 0)
 1599                    {
 600                        // This altServerAddress becomes the main server address, and the existing main
 601                        // server address becomes the first alt server address.
 1602                        serverAddressFeature.AltServerAddresses = serverAddressFeature.AltServerAddresses
 1603                            .RemoveAt(enumerator.CurrentIndex - 1)
 1604                            .Insert(0, serverAddressFeature.ServerAddress!.Value);
 1605                        serverAddressFeature.ServerAddress = serverAddress;
 1606                    }
 77607                    return true;
 608                }
 20609            }
 15610            return false;
 611        }
 92612    }
 613
 614    /// <summary>A helper struct that implements an enumerator that allows iterating the addresses of an
 615    /// <see cref="IServerAddressFeature" /> without allocations.</summary>
 616    private struct ServerAddressEnumerator
 617    {
 618        internal readonly ServerAddress Current
 619        {
 620            get
 117621            {
 117622                Debug.Assert(CurrentIndex >= 0 && CurrentIndex <= _altServerAddresses.Count);
 117623                if (CurrentIndex == 0)
 109624                {
 109625                    Debug.Assert(_mainServerAddress is not null);
 109626                    return _mainServerAddress.Value;
 627                }
 628                else
 8629                {
 8630                    return _altServerAddresses[CurrentIndex - 1];
 631                }
 117632            }
 633        }
 634
 0635        internal int Count { get; }
 636
 955637        internal int CurrentIndex { get; private set; } = -1;
 638
 639        private readonly ServerAddress? _mainServerAddress;
 640        private readonly IList<ServerAddress> _altServerAddresses;
 641
 642        internal bool MoveNext()
 132643        {
 132644            if (CurrentIndex == -1)
 109645            {
 109646                if (_mainServerAddress is not null)
 109647                {
 109648                    CurrentIndex++;
 109649                    return true;
 650                }
 651                else
 0652                {
 0653                    return false;
 654                }
 655            }
 23656            else if (CurrentIndex < _altServerAddresses.Count)
 8657            {
 8658                CurrentIndex++;
 8659                return true;
 660            }
 15661            return false;
 132662        }
 663
 664        internal ServerAddressEnumerator(IServerAddressFeature serverAddressFeature)
 109665        {
 109666            _mainServerAddress = serverAddressFeature.ServerAddress;
 109667            _altServerAddresses = serverAddressFeature.AltServerAddresses;
 109668            if (_mainServerAddress is null)
 0669            {
 0670                Count = 0;
 0671            }
 672            else
 109673            {
 109674                Count = _altServerAddresses.Count + 1;
 109675            }
 109676        }
 677    }
 678}
Generated by: ReportGenerator 5.3.8.0
01/09/2026 - 15:07:32
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)