< Summary

Information
Class: IceRpc.Transports.Coloc.Internal.ColocListener
Assembly: IceRpc.Transports.Coloc
File(s): /home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc.Transports.Coloc/Internal/ColocListener.cs
Tag: 2300_35243572715
Line coverage
94%
Covered lines: 79
Uncovered lines: 5
Coverable lines: 84
Total lines: 172
Line coverage: 94%
Branch coverage
87%
Covered branches: 7
Total branches: 8
Branch coverage: 87.5%
Method coverage
100%
Covered methods: 6
Fully covered methods: 4
Total methods: 6
Method coverage: 100%
Full method coverage: 66.6%

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
get_TransportAddress()100%11100%
.ctor(...)100%11100%
AcceptAsync()50%2283.33%
Dispose()100%44100%
DisposeAsync()100%11100%
TryQueueConnect(...)100%22100%

File(s)

/home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc.Transports.Coloc/Internal/ColocListener.cs

#LineLine coverage
 1// Copyright (c) ZeroC, Inc.
 2
 3using System.Diagnostics;
 4using System.Diagnostics.CodeAnalysis;
 5using System.IO.Pipelines;
 6using System.Net;
 7using System.Threading.Channels;
 8
 9using ConnectRequest = (
 10    System.Threading.Tasks.TaskCompletionSource<System.IO.Pipelines.PipeReader> Tcs,
 11    System.IO.Pipelines.PipeReader ClientPipeReader,
 12    System.Threading.CancellationTokenRegistration Registration);
 13
 14namespace IceRpc.Transports.Coloc.Internal;
 15
 16/// <summary>The listener implementation for the colocated transport.</summary>
 17internal class ColocListener : IListener<IDuplexConnection>, IDisposable
 18{
 108019    public TransportAddress TransportAddress { get; }
 20
 21    [SuppressMessage(
 22        "Usage",
 23        "CA2213:Disposable fields should be disposed",
 24        Justification = "Disposing this CTS races with AcceptAsync creating a linked token source from its Token; a CTS 
 57525    private readonly CancellationTokenSource _disposeCts = new();
 26    private bool _disposed;
 27    private readonly Action<ColocListener> _onDispose;
 57528    private readonly Lock _mutex = new();
 29    private readonly EndPoint _networkAddress;
 30    private readonly PipeOptions _pipeOptions;
 31
 32    // The channel used by the client connection ConnectAsync method to queue a connection establishment request. A
 33    // client connection establishment request is represented by:
 34    // - a TaskCompletionSource which is completed by AcceptAsync when the connection is accepted. The server connection
 35    //   pipe reader is set as the result. ClientColocConnection.ConnectAsync waits on the task completion source task.
 36    // - the client connection pipe reader provided to the server connection when the server connection is created by
 37    //   AcceptAsync.
 38    // - the cancellation token registration that cancels the TaskCompletionSource; it's disposed when the request is
 39    //   dequeued.
 40    private readonly Channel<ConnectRequest> _channel;
 41
 42    public async Task<(IDuplexConnection, EndPoint)> AcceptAsync(CancellationToken cancellationToken)
 61043    {
 44        CancellationTokenSource cts;
 45        lock (_mutex)
 61046        {
 61047            ObjectDisposedException.ThrowIf(_disposed, this);
 60848            cts = CancellationTokenSource.CreateLinkedTokenSource(_disposeCts.Token, cancellationToken);
 60849        }
 60850        using var _ = cts;
 51        try
 60852        {
 60853            while (true)
 60854            {
 60855                ConnectRequest request = await _channel.Reader.ReadAsync(cts.Token).ConfigureAwait(false);
 52756                request.Registration.Dispose();
 57
 52758                var serverPipe = new Pipe(_pipeOptions);
 52759                if (request.Tcs.TrySetResult(serverPipe.Reader))
 52760                {
 52761                    var serverConnection = new ServerColocConnection(
 52762                        TransportAddress,
 52763                        serverPipe.Writer,
 52764                        request.ClientPipeReader);
 52765                    return (serverConnection, _networkAddress);
 66                }
 67                else
 068                {
 69                    // The client connection establishment was canceled.
 070                    serverPipe.Writer.Complete();
 071                    serverPipe.Reader.Complete();
 072                }
 073            }
 74        }
 8175        catch (OperationCanceledException)
 8176        {
 8177            cancellationToken.ThrowIfCancellationRequested();
 78            // The accept operation was canceled because the listener was disposed.
 1179            Debug.Assert(_disposeCts.IsCancellationRequested);
 1180            throw new ObjectDisposedException($"{typeof(ColocListener)}");
 81        }
 52782    }
 83
 84    public void Dispose()
 102285    {
 86        lock (_mutex)
 102287        {
 102288            if (_disposed)
 44789            {
 44790                return;
 91            }
 57592            _disposed = true;
 93
 94            // Notify the owner (e.g. the server transport) so it can release its reference to this listener.
 57595            _onDispose(this);
 96
 97            // Cancel pending AcceptAsync.
 57598            _disposeCts.Cancel();
 99
 100            // Ensure no more client connection establishment request is queued.
 575101            _channel.Writer.Complete();
 102
 103            // Complete all the queued client connection establishment requests with IceRpcError.ConnectionRefused.
 104            // Use TrySetException in case the task has been already canceled.
 594105            while (_channel.Reader.TryRead(out ConnectRequest item))
 19106            {
 19107                item.Registration.Dispose();
 19108                item.Tcs.TrySetException(new IceRpcException(IceRpcError.ConnectionRefused));
 19109            }
 575110        }
 1022111    }
 112
 113    public ValueTask DisposeAsync()
 1020114    {
 1020115        Dispose();
 1020116        return default;
 1020117    }
 118
 575119    internal ColocListener(
 575120        TransportAddress transportAddress,
 575121        Action<ColocListener> onDispose,
 575122        ColocTransportOptions colocTransportOptions,
 575123        DuplexConnectionOptions duplexConnectionOptions)
 575124    {
 575125        TransportAddress = transportAddress;
 126
 575127        _onDispose = onDispose;
 575128        _networkAddress = new ColocEndPoint(transportAddress);
 575129        _pipeOptions = new PipeOptions(
 575130            pool: duplexConnectionOptions.Pool,
 575131            minimumSegmentSize: duplexConnectionOptions.MinSegmentSize,
 575132            pauseWriterThreshold: colocTransportOptions.PauseWriterThreshold,
 575133            resumeWriterThreshold: colocTransportOptions.ResumeWriterThreshold,
 575134            useSynchronizationContext: false);
 135
 136        // Create a bounded channel with a capacity that matches the listen backlog, and with
 137        // the default concurrency settings that allow multiple reader and writers.
 575138        _channel = Channel.CreateBounded<ConnectRequest>(
 575139            new BoundedChannelOptions(colocTransportOptions.ListenBacklog));
 575140    }
 141
 142    /// <summary>Queue client connection establishment requests from the client.</summary>
 143    /// <param name="clientPipeReader">A <see cref="PipeReader"/> for reading from the client connection.</param>
 144    /// <param name="cancellationToken">>A cancellation token that receives the cancellation requests.</param>
 145    /// <param name="serverPipeReaderTask">A task that returns a <see cref="PipeReader"/> for reading from the server
 146    /// connection.</param>
 147    /// <returns>Returns true if the connection establishment request has been queue otherwise, false.</returns>
 148    internal bool TryQueueConnect(
 149        PipeReader clientPipeReader,
 150        CancellationToken cancellationToken,
 151        [NotNullWhen(true)] out Task<PipeReader>? serverPipeReaderTask)
 549152    {
 153        // Create a tcs that is completed by AcceptAsync when accepts the corresponding connection, at which point
 154        // the client side connect operation will complete.
 155        // We use RunContinuationsAsynchronously to avoid the ConnectAsync continuation end up running in the AcceptAsyn
 156        // loop that completes this tcs.
 549157        var tcs = new TaskCompletionSource<PipeReader>(TaskCreationOptions.RunContinuationsAsynchronously);
 549158        CancellationTokenRegistration registration =
 556159            cancellationToken.Register(() => tcs.TrySetCanceled(cancellationToken));
 549160        if (_channel.Writer.TryWrite((tcs, clientPipeReader, registration)))
 546161        {
 546162            serverPipeReaderTask = tcs.Task;
 546163            return true;
 164        }
 165        else
 3166        {
 3167            registration.Dispose();
 3168            serverPipeReaderTask = null;
 3169            return false;
 170        }
 549171    }
 172}
Generated by: ReportGenerator 5.5.11.0
09/17/2026 - 16:01:28
GitHub | reportgenerator.io

Methods/Properties

get_TransportAddress()
.ctor(IceRpc.Transports.TransportAddress,System.Action`1<IceRpc.Transports.Coloc.Internal.ColocListener>,IceRpc.Transports.Coloc.ColocTransportOptions,IceRpc.Transports.DuplexConnectionOptions)
AcceptAsync()
Dispose()
DisposeAsync()
TryQueueConnect(System.IO.Pipelines.PipeReader,System.Threading.CancellationToken,System.Threading.Tasks.Task`1<System.IO.Pipelines.PipeReader>&)