< Summary

Information
Class: IceRpc.ConnectionCache
Assembly: IceRpc
File(s): /home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc/ConnectionCache.cs
Tag: 275_13775359185
Line coverage
71%
Covered lines: 273
Uncovered lines: 108
Coverable lines: 381
Total lines: 677
Line coverage: 71.6%
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%13.45856%
PerformInvokeAsync()66.66%12.87642.42%
ShutdownAsync(...)75%4.04486.66%
PerformShutdownAsync()50%7.46440%
CreateConnectTask()50%10.1868%
DisposePendingConnectionAsync()75%4.8463.15%
ShutdownWhenRequestedAsync()100%11100%
GetActiveConnectionAsync()64.28%18.341471.92%
RemoveFromActiveAsync(...)50%4.07483.33%
ShutdownAndDisposeConnectionAsync()75%4.07483.33%
TryGetActiveConnection(...)80%10.331085.18%
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.
 1820    private readonly Dictionary<ServerAddress, IProtocolConnection> _activeConnections =
 1821        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
 1832    private readonly TaskCompletionSource _detachedConnectionsTcs = new();
 33
 34    // A cancellation token source that is canceled when DisposeAsync is called.
 1835    private readonly CancellationTokenSource _disposedCts = new();
 36
 37    private Task? _disposeTask;
 38
 1839    private readonly object _mutex = new();
 40
 41    // New connections in the process of connecting.
 1842    private readonly Dictionary<ServerAddress, (IProtocolConnection Connection, Task ConnectTask)> _pendingConnections =
 1843        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>
 1859    public ConnectionCache(
 1860        ConnectionCacheOptions options,
 1861        IDuplexClientTransport? duplexClientTransport = null,
 1862        IMultiplexedClientTransport? multiplexedClientTransport = null,
 1863        ILogger? logger = null)
 1864    {
 1865        _connectionFactory = new ClientProtocolConnectionFactory(
 1866            options.ConnectionOptions,
 1867            options.ClientAuthenticationOptions,
 1868            duplexClientTransport,
 1869            multiplexedClientTransport,
 1870            logger);
 71
 1872        _connectTimeout = options.ConnectTimeout;
 1873        _shutdownTimeout = options.ShutdownTimeout;
 74
 1875        _preferExistingConnection = options.PreferExistingConnection;
 1876    }
 77
 78    /// <summary>Constructs a connection cache using the default options.</summary>
 79    public ConnectionCache()
 080        : this(new ConnectionCacheOptions())
 081    {
 082    }
 83
 84    /// <summary>Releases all resources allocated by the cache. The cache disposes all the connections it
 85    /// created.</summary>
 86    /// <returns>A value task that completes when the disposal of all connections created by this cache has completed.
 87    /// This includes connections that were active when this method is called and connections whose disposal was
 88    /// initiated prior to this call.</returns>
 89    /// <remarks>The disposal of an underlying connection of the cache  aborts invocations, cancels dispatches and
 90    /// disposes the underlying transport connection without waiting for the peer. To wait for invocations and
 91    /// dispatches to complete, call <see cref="ShutdownAsync" /> first. If the configured dispatcher does not complete
 92    /// promptly when its cancellation token is canceled, the disposal can hang.</remarks>
 93    public ValueTask DisposeAsync()
 2094    {
 2095        lock (_mutex)
 2096        {
 2097            if (_disposeTask is null)
 1898            {
 1899                _shutdownTask ??= Task.CompletedTask;
 18100                if (_detachedConnectionCount == 0)
 16101                {
 16102                    _ = _detachedConnectionsTcs.TrySetResult();
 16103                }
 104
 18105                _disposeTask = PerformDisposeAsync();
 18106            }
 20107            return new(_disposeTask);
 108        }
 109
 110        async Task PerformDisposeAsync()
 18111        {
 18112            await Task.Yield(); // exit mutex lock
 113
 18114            _disposedCts.Cancel();
 115
 116            // Wait for shutdown before disposing connections.
 117            try
 18118            {
 18119                await _shutdownTask.ConfigureAwait(false);
 18120            }
 0121            catch
 0122            {
 123                // Ignore exceptions.
 0124            }
 125
 126            // Since a pending connection is "detached", it's disposed via the connectTask, not directly by this method.
 18127            await Task.WhenAll(
 8128                _activeConnections.Values.Select(connection => connection.DisposeAsync().AsTask())
 18129                    .Append(_detachedConnectionsTcs.Task)).ConfigureAwait(false);
 130
 18131            _disposedCts.Dispose();
 18132        }
 20133    }
 134
 135    /// <summary>Sends an outgoing request and returns the corresponding incoming response.</summary>
 136    /// <param name="request">The outgoing request being sent.</param>
 137    /// <param name="cancellationToken">A cancellation token that receives the cancellation requests.</param>
 138    /// <returns>The corresponding <see cref="IncomingResponse" />.</returns>
 139    /// <exception cref="InvalidOperationException">Thrown if no <see cref="IServerAddressFeature" /> feature is set and
 140    /// the request's service address has no server addresses.</exception>
 141    /// <exception cref="IceRpcException">Thrown with one of the following error:
 142    /// <list type="bullet"><item><term><see cref="IceRpcError.InvocationRefused" /></term><description>This error
 143    /// indicates that the connection cache is shutdown.</description></item>
 144    /// <item><term><see cref="IceRpcError.NoConnection" /></term><description>This error indicates that the request
 145    /// <see cref="IServerAddressFeature" /> feature has no server addresses.</description></item></list>.
 146    /// </exception>
 147    /// <exception cref="ObjectDisposedException">Thrown if this connection cache is disposed.</exception>
 148    /// <remarks><para>If the request <see cref="IServerAddressFeature" /> feature is not set, the cache sets it from
 149    /// the server addresses of the target service.</para>
 150    /// <para>It then looks for an active connection. The <see cref="ConnectionCacheOptions.PreferExistingConnection" />
 151    /// property influences how the cache selects this active connection. If no active connection can be found, the
 152    /// cache creates a new connection to one of the server addresses from the <see cref="IServerAddressFeature" />
 153    /// feature.</para>
 154    /// <para>If the connection establishment to <see cref="IServerAddressFeature.ServerAddress" /> fails, <see
 155    /// cref="IServerAddressFeature.ServerAddress" /> is appended at the end of <see
 156    /// cref="IServerAddressFeature.AltServerAddresses" /> and the first address from <see
 157    /// cref="IServerAddressFeature.AltServerAddresses" /> replaces <see cref="IServerAddressFeature.ServerAddress" />.
 158    /// The cache tries again to find or establish a connection to <see cref="IServerAddressFeature.ServerAddress" />.
 159    /// If unsuccessful, the cache repeats this process until success or until it tried all the addresses. If all the
 160    /// attempts fail, this method throws the exception from the last attempt.</para></remarks>
 161    public Task<IncomingResponse> InvokeAsync(OutgoingRequest request, CancellationToken cancellationToken)
 102162    {
 102163        if (request.Features.Get<IServerAddressFeature>() is IServerAddressFeature serverAddressFeature)
 0164        {
 0165            if (serverAddressFeature.ServerAddress is null)
 0166            {
 0167                throw new IceRpcException(
 0168                    IceRpcError.NoConnection,
 0169                    $"Could not invoke '{request.Operation}' on '{request.ServiceAddress}': tried all server addresses w
 170            }
 0171        }
 172        else
 102173        {
 102174            if (request.ServiceAddress.ServerAddress is null)
 0175            {
 0176                throw new InvalidOperationException("Cannot send a request to a service without a server address.");
 177            }
 178
 102179            serverAddressFeature = new ServerAddressFeature(request.ServiceAddress);
 102180            request.Features = request.Features.With(serverAddressFeature);
 102181        }
 182
 102183        lock (_mutex)
 102184        {
 102185            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 186
 102187            if (_shutdownTask is not null)
 0188            {
 0189                throw new IceRpcException(IceRpcError.InvocationRefused, "The connection cache was shut down.");
 190            }
 102191        }
 192
 102193        return PerformInvokeAsync();
 194
 195        async Task<IncomingResponse> PerformInvokeAsync()
 102196        {
 102197            Debug.Assert(serverAddressFeature.ServerAddress is not null);
 198
 199            // When InvokeAsync (or ConnectAsync) throws an IceRpcException(InvocationRefused) we retry unless the
 200            // cache is being shutdown.
 102201            while (true)
 102202            {
 102203                IProtocolConnection? connection = null;
 102204                if (_preferExistingConnection)
 98205                {
 98206                    _ = TryGetActiveConnection(serverAddressFeature, out connection);
 98207                }
 102208                connection ??= await GetActiveConnectionAsync(serverAddressFeature, cancellationToken)
 102209                    .ConfigureAwait(false);
 210
 211                try
 102212                {
 102213                    return await connection.InvokeAsync(request, cancellationToken).ConfigureAwait(false);
 214                }
 0215                catch (ObjectDisposedException)
 0216                {
 217                    // This can occasionally happen if we find a connection that was just closed and then automatically
 218                    // disposed by this connection cache.
 0219                }
 0220                catch (IceRpcException exception) when (exception.IceRpcError == IceRpcError.InvocationRefused)
 0221                {
 222                    // The connection is refusing new invocations.
 0223                }
 0224                catch (IceRpcException exception) when (exception.IceRpcError == IceRpcError.OperationAborted)
 0225                {
 0226                    lock (_mutex)
 0227                    {
 0228                        if (_disposeTask is null)
 0229                        {
 0230                            throw new IceRpcException(
 0231                                IceRpcError.ConnectionAborted,
 0232                                "The underlying connection was disposed while the invocation was in progress.");
 233                        }
 234                        else
 0235                        {
 0236                            throw;
 237                        }
 238                    }
 239                }
 240
 241                // Make sure connection is no longer in _activeConnection before we retry.
 0242                _ = RemoveFromActiveAsync(serverAddressFeature.ServerAddress.Value, connection);
 0243            }
 102244        }
 102245    }
 246
 247    /// <summary>Gracefully shuts down all connections created by this cache.</summary>
 248    /// <param name="cancellationToken">A cancellation token that receives the cancellation requests.</param>
 249    /// <returns>A task that completes successfully once the shutdown of all connections created by this cache has
 250    /// completed. This includes connections that were active when this method is called and connections whose shutdown
 251    /// was initiated prior to this call. This task can also complete with one of the following exceptions:
 252    /// <list type="bullet">
 253    /// <item><description><see cref="IceRpcException" /> with error <see cref="IceRpcError.OperationAborted" /> if the
 254    /// connection cache is disposed while being shut down.</description></item>
 255    /// <item><description><see cref="OperationCanceledException" /> if cancellation was requested through the
 256    /// cancellation token.</description></item>
 257    /// <item><description><see cref="TimeoutException" /> if the shutdown timed out.</description></item>
 258    /// </list>
 259    /// </returns>
 260    /// <exception cref="InvalidOperationException">Thrown if this method is called more than once.</exception>
 261    /// <exception cref="ObjectDisposedException">Thrown if the connection cache is disposed.</exception>
 262    public Task ShutdownAsync(CancellationToken cancellationToken = default)
 16263    {
 16264        lock (_mutex)
 16265        {
 16266            ObjectDisposedException.ThrowIf(_disposeTask is not null, this);
 267
 16268            if (_shutdownTask is not null)
 0269            {
 0270                throw new InvalidOperationException("The connection cache is already shut down or shutting down.");
 271            }
 272
 16273            if (_detachedConnectionCount == 0)
 7274            {
 7275                _detachedConnectionsTcs.SetResult();
 7276            }
 277
 16278            _shutdownTask = PerformShutdownAsync();
 16279        }
 280
 16281        return _shutdownTask;
 282
 283        async Task PerformShutdownAsync()
 16284        {
 16285            await Task.Yield(); // exit mutex lock
 286
 16287            using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _disposedCts.Token);
 16288            cts.CancelAfter(_shutdownTimeout);
 289
 290            try
 16291            {
 292                // Since a pending connection is "detached", it's shutdown and disposed via the connectTask, not
 293                // directly by this method.
 294                try
 16295                {
 16296                    await Task.WhenAll(
 8297                        _activeConnections.Values.Select(connection => connection.ShutdownAsync(cts.Token))
 16298                            .Append(_detachedConnectionsTcs.Task.WaitAsync(cts.Token))).ConfigureAwait(false);
 16299                }
 0300                catch (OperationCanceledException)
 0301                {
 0302                    throw;
 303                }
 0304                catch
 0305                {
 306                    // Ignore other connection shutdown failures.
 307
 308                    // Throw OperationCanceledException if this WhenAll exception is hiding an OCE.
 0309                    cts.Token.ThrowIfCancellationRequested();
 0310                }
 16311            }
 0312            catch (OperationCanceledException)
 0313            {
 0314                cancellationToken.ThrowIfCancellationRequested();
 315
 0316                if (_disposedCts.IsCancellationRequested)
 0317                {
 0318                    throw new IceRpcException(
 0319                        IceRpcError.OperationAborted,
 0320                        "The shutdown was aborted because the connection cache was disposed.");
 321                }
 322                else
 0323                {
 0324                    throw new TimeoutException(
 0325                        $"The connection cache shut down timed out after {_shutdownTimeout.TotalSeconds} s.");
 326                }
 327            }
 16328        }
 16329    }
 330
 331    private async Task CreateConnectTask(IProtocolConnection connection, ServerAddress serverAddress)
 30332    {
 30333        await Task.Yield(); // exit mutex lock
 334
 335        // This task "owns" a detachedConnectionCount and as a result _disposedCts can't be disposed.
 30336        using var cts = CancellationTokenSource.CreateLinkedTokenSource(_disposedCts.Token);
 30337        cts.CancelAfter(_connectTimeout);
 338
 339        Task shutdownRequested;
 30340        Task? connectTask = null;
 341
 342        try
 30343        {
 344            try
 30345            {
 30346                (_, shutdownRequested) = await connection.ConnectAsync(cts.Token).ConfigureAwait(false);
 24347            }
 0348            catch (OperationCanceledException)
 0349            {
 0350                if (_disposedCts.IsCancellationRequested)
 0351                {
 0352                    throw new IceRpcException(
 0353                        IceRpcError.OperationAborted,
 0354                        "The connection establishment was aborted because the connection cache was disposed.");
 355                }
 356                else
 0357                {
 0358                    throw new TimeoutException(
 0359                        $"The connection establishment timed out after {_connectTimeout.TotalSeconds} s.");
 360                }
 361            }
 24362        }
 6363        catch
 6364        {
 6365            lock (_mutex)
 6366            {
 367                // connectTask is executing this method and about to throw.
 6368                connectTask = _pendingConnections[serverAddress].ConnectTask;
 6369                _pendingConnections.Remove(serverAddress);
 6370            }
 371
 6372            _ = DisposePendingConnectionAsync(connection, connectTask);
 6373            throw;
 374        }
 375
 24376        lock (_mutex)
 24377        {
 24378            if (_shutdownTask is null)
 24379            {
 380                // the connection is now "attached" in _activeConnections
 24381                _activeConnections.Add(serverAddress, connection);
 24382                _detachedConnectionCount--;
 24383            }
 384            else
 0385            {
 0386                connectTask = _pendingConnections[serverAddress].ConnectTask;
 0387            }
 24388            bool removed = _pendingConnections.Remove(serverAddress);
 24389            Debug.Assert(removed);
 24390        }
 391
 24392        if (connectTask is null)
 24393        {
 24394            _ = ShutdownWhenRequestedAsync(connection, serverAddress, shutdownRequested);
 24395        }
 396        else
 0397        {
 398            // As soon as this method completes successfully, we shut down then dispose the connection.
 0399            _ = DisposePendingConnectionAsync(connection, connectTask);
 0400        }
 401
 402        async Task DisposePendingConnectionAsync(IProtocolConnection connection, Task connectTask)
 6403        {
 404            try
 6405            {
 6406                await connectTask.ConfigureAwait(false);
 407
 408                // Since we own a detachedConnectionCount, _disposedCts is not disposed.
 0409                using var cts = CancellationTokenSource.CreateLinkedTokenSource(_disposedCts.Token);
 0410                cts.CancelAfter(_shutdownTimeout);
 0411                await connection.ShutdownAsync(cts.Token).ConfigureAwait(false);
 0412            }
 6413            catch
 6414            {
 415                // Observe and ignore exceptions.
 6416            }
 417
 6418            await connection.DisposeAsync().ConfigureAwait(false);
 419
 6420            lock (_mutex)
 6421            {
 6422                if (--_detachedConnectionCount == 0 && _shutdownTask is not null)
 0423                {
 0424                    _detachedConnectionsTcs.SetResult();
 0425                }
 6426            }
 6427        }
 428
 429        async Task ShutdownWhenRequestedAsync(
 430            IProtocolConnection connection,
 431            ServerAddress serverAddress,
 432            Task shutdownRequested)
 24433        {
 24434            await shutdownRequested.ConfigureAwait(false);
 20435            await RemoveFromActiveAsync(serverAddress, connection).ConfigureAwait(false);
 20436        }
 24437    }
 438
 439    /// <summary>Gets an active connection, by creating and connecting (if necessary) a new protocol connection.
 440    /// </summary>
 441    /// <param name="serverAddressFeature">The server address feature.</param>
 442    /// <param name="cancellationToken">The cancellation token of the invocation calling this method.</param>
 443    private async Task<IProtocolConnection> GetActiveConnectionAsync(
 444        IServerAddressFeature serverAddressFeature,
 445        CancellationToken cancellationToken)
 24446    {
 24447        Debug.Assert(serverAddressFeature.ServerAddress is not null);
 24448        Exception? connectionException = null;
 449        (IProtocolConnection Connection, Task ConnectTask) pendingConnectionValue;
 24450        var enumerator = new ServerAddressEnumerator(serverAddressFeature);
 30451        while (enumerator.MoveNext())
 30452        {
 30453            ServerAddress serverAddress = enumerator.Current;
 30454            if (enumerator.CurrentIndex > 0)
 6455            {
 456                // Rotate the server addresses before each new connection attempt after the initial attempt
 6457                serverAddressFeature.RotateAddresses();
 6458            }
 459
 460            try
 30461            {
 30462                lock (_mutex)
 30463                {
 30464                    if (_disposeTask is not null)
 0465                    {
 0466                        throw new IceRpcException(IceRpcError.OperationAborted, "The connection cache was disposed.");
 467                    }
 30468                    else if (_shutdownTask is not null)
 0469                    {
 0470                        throw new IceRpcException(IceRpcError.InvocationRefused, "The connection cache is shut down.");
 471                    }
 472
 30473                    if (_activeConnections.TryGetValue(serverAddress, out IProtocolConnection? connection))
 0474                    {
 0475                        return connection;
 476                    }
 477
 30478                    if (!_pendingConnections.TryGetValue(serverAddress, out pendingConnectionValue))
 30479                    {
 30480                        connection = _connectionFactory.CreateConnection(serverAddress);
 30481                        _detachedConnectionCount++;
 30482                        pendingConnectionValue = (connection, CreateConnectTask(connection, serverAddress));
 30483                        _pendingConnections.Add(serverAddress, pendingConnectionValue);
 30484                    }
 30485                }
 486                // ConnectTask itself takes care of scheduling its exception observation when it fails.
 30487                await pendingConnectionValue.ConnectTask.WaitAsync(cancellationToken).ConfigureAwait(false);
 24488                return pendingConnectionValue.Connection;
 489            }
 0490            catch (TimeoutException exception)
 0491            {
 0492                connectionException = exception;
 0493            }
 6494            catch (IceRpcException exception) when (exception.IceRpcError is
 6495                IceRpcError.ConnectionAborted or
 6496                IceRpcError.ConnectionRefused or
 6497                IceRpcError.ServerBusy or
 6498                IceRpcError.ServerUnreachable)
 6499            {
 500                // keep going unless the connection cache was disposed or shut down
 6501                connectionException = exception;
 6502                lock (_mutex)
 6503                {
 6504                    if (_shutdownTask is not null)
 0505                    {
 0506                        throw;
 507                    }
 6508                }
 6509            }
 6510        }
 511
 0512        Debug.Assert(connectionException is not null);
 0513        ExceptionDispatchInfo.Throw(connectionException);
 0514        Debug.Assert(false);
 0515        throw connectionException;
 24516    }
 517
 518    /// <summary>Removes the connection from _activeConnections, and when successful, shuts down and disposes this
 519    /// connection.</summary>
 520    /// <param name="serverAddress">The server address key in _activeConnections.</param>
 521    /// <param name="connection">The connection to shutdown and dispose after removal.</param>
 522    private Task RemoveFromActiveAsync(ServerAddress serverAddress, IProtocolConnection connection)
 20523    {
 20524        lock (_mutex)
 20525        {
 20526            if (_shutdownTask is null && _activeConnections.Remove(serverAddress))
 16527            {
 528                // it's now our connection.
 16529                _detachedConnectionCount++;
 16530            }
 531            else
 4532            {
 533                // Another task owns this connection
 4534                return Task.CompletedTask;
 535            }
 16536        }
 537
 16538        return ShutdownAndDisposeConnectionAsync();
 539
 540        async Task ShutdownAndDisposeConnectionAsync()
 16541        {
 542            // _disposedCts is not disposed since we own a detachedConnectionCount
 16543            using var cts = CancellationTokenSource.CreateLinkedTokenSource(_disposedCts.Token);
 16544            cts.CancelAfter(_shutdownTimeout);
 545
 546            try
 16547            {
 16548                await connection.ShutdownAsync(cts.Token).ConfigureAwait(false);
 16549            }
 0550            catch
 0551            {
 552                // Ignore connection shutdown failures
 0553            }
 554
 16555            await connection.DisposeAsync().ConfigureAwait(false);
 556
 16557            lock (_mutex)
 16558            {
 16559                if (--_detachedConnectionCount == 0 && _shutdownTask is not null)
 11560                {
 11561                    _detachedConnectionsTcs.SetResult();
 11562                }
 16563            }
 16564        }
 20565    }
 566
 567    /// <summary>Tries to get an existing connection matching one of the addresses of the server address feature.
 568    /// </summary>
 569    /// <param name="serverAddressFeature">The server address feature.</param>
 570    /// <param name="connection">When this method returns <see langword="true" />, this argument contains an active
 571    /// connection; otherwise, it is set to <see langword="null" />.</param>
 572    /// <returns><see langword="true" /> when an active connection matching any of the addresses of the server address
 573    /// feature is found; otherwise, <see langword="false"/>.</returns>
 574    private bool TryGetActiveConnection(
 575        IServerAddressFeature serverAddressFeature,
 576        [NotNullWhen(true)] out IProtocolConnection? connection)
 98577    {
 98578        lock (_mutex)
 98579        {
 98580            connection = null;
 98581            if (_disposeTask is not null)
 0582            {
 0583                throw new IceRpcException(IceRpcError.OperationAborted, "The connection cache was disposed.");
 584            }
 585
 98586            if (_shutdownTask is not null)
 0587            {
 0588                throw new IceRpcException(IceRpcError.InvocationRefused, "The connection cache was shut down.");
 589            }
 590
 98591            var enumerator = new ServerAddressEnumerator(serverAddressFeature);
 128592            while (enumerator.MoveNext())
 108593            {
 108594                ServerAddress serverAddress = enumerator.Current;
 108595                if (_activeConnections.TryGetValue(serverAddress, out connection))
 78596                {
 78597                    if (enumerator.CurrentIndex > 0)
 2598                    {
 599                        // This altServerAddress becomes the main server address, and the existing main
 600                        // server address becomes the first alt server address.
 2601                        serverAddressFeature.AltServerAddresses = serverAddressFeature.AltServerAddresses
 2602                            .RemoveAt(enumerator.CurrentIndex - 1)
 2603                            .Insert(0, serverAddressFeature.ServerAddress!.Value);
 2604                        serverAddressFeature.ServerAddress = serverAddress;
 2605                    }
 78606                    return true;
 607                }
 30608            }
 20609            return false;
 610        }
 98611    }
 612
 613    /// <summary>A helper struct that implements an enumerator that allows iterating the addresses of an
 614    /// <see cref="IServerAddressFeature" /> without allocations.</summary>
 615    private struct ServerAddressEnumerator
 616    {
 617        internal readonly ServerAddress Current
 618        {
 619            get
 138620            {
 138621                Debug.Assert(CurrentIndex >= 0 && CurrentIndex <= _altServerAddresses.Count);
 138622                if (CurrentIndex == 0)
 122623                {
 122624                    Debug.Assert(_mainServerAddress is not null);
 122625                    return _mainServerAddress.Value;
 626                }
 627                else
 16628                {
 16629                    return _altServerAddresses[CurrentIndex - 1];
 630                }
 138631            }
 632        }
 633
 0634        internal int Count { get; }
 635
 1132636        internal int CurrentIndex { get; private set; } = -1;
 637
 638        private readonly ServerAddress? _mainServerAddress;
 639        private readonly IList<ServerAddress> _altServerAddresses;
 640
 641        internal bool MoveNext()
 158642        {
 158643            if (CurrentIndex == -1)
 122644            {
 122645                if (_mainServerAddress is not null)
 122646                {
 122647                    CurrentIndex++;
 122648                    return true;
 649                }
 650                else
 0651                {
 0652                    return false;
 653                }
 654            }
 36655            else if (CurrentIndex < _altServerAddresses.Count)
 16656            {
 16657                CurrentIndex++;
 16658                return true;
 659            }
 20660            return false;
 158661        }
 662
 663        internal ServerAddressEnumerator(IServerAddressFeature serverAddressFeature)
 122664        {
 122665            _mainServerAddress = serverAddressFeature.ServerAddress;
 122666            _altServerAddresses = serverAddressFeature.AltServerAddresses;
 122667            if (_mainServerAddress is null)
 0668            {
 0669                Count = 0;
 0670            }
 671            else
 122672            {
 122673                Count = _altServerAddresses.Count + 1;
 122674            }
 122675        }
 676    }
 677}
Generated by: ReportGenerator 5.3.8.0
03/10/2025 - 21:46:44
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)