< Summary

Information
Class: IceRpc.Transports.Tcp.Internal.TcpConnection
Assembly: IceRpc
File(s): /home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc/Transports/Tcp/Internal/TcpConnection.cs
Tag: 2300_35243572715
Line coverage
73%
Covered lines: 125
Uncovered lines: 44
Coverable lines: 169
Total lines: 442
Line coverage: 73.9%
Branch coverage
80%
Covered branches: 33
Total branches: 41
Branch coverage: 80.4%
Method coverage
90%
Covered methods: 9
Fully covered methods: 4
Total methods: 10
Method coverage: 90%
Full method coverage: 40%

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%22100%
ConnectAsync(...)100%11100%
Dispose()100%66100%
ReadAsync(...)100%22100%
PerformReadAsync()100%22100%
ShutdownWriteAsync(...)100%11100%
PerformShutdownAsync()100%2255.55%
WriteAsync(...)100%22100%
PerformWriteAsync()68%422569.66%
AbortAndObserveAsync()100%210%

File(s)

/home/runner/work/icerpc-csharp/icerpc-csharp/src/IceRpc/Transports/Tcp/Internal/TcpConnection.cs

#LineLine coverage
 1// Copyright (c) ZeroC, Inc.
 2
 3using System.Buffers;
 4using System.Diagnostics;
 5using System.Net;
 6using System.Net.Security;
 7using System.Net.Sockets;
 8using System.Runtime.InteropServices;
 9
 10namespace IceRpc.Transports.Tcp.Internal;
 11
 12/// <summary>Implements <see cref="IDuplexConnection" /> for tcp with or without TLS.</summary>
 13/// <remarks>Unlike Coloc, the Tcp transport is not a "checked" transport, which means it does not need to detect
 14/// violations of the duplex transport contract or report such violations. It assumes its clients are sufficiently well
 15/// tested to never violate this contract. As a result, this implementation does not throw
 16/// <see cref="InvalidOperationException" />.</remarks>
 17internal abstract class TcpConnection : IDuplexConnection
 18{
 19    internal abstract Socket Socket { get; }
 20
 21    internal abstract SslStream? SslStream { get; }
 22
 23    private protected volatile bool _isDisposed;
 24
 25    // The MaxDataSize of the SSL implementation.
 26    private const int MaxSslDataSize = 16 * 1024;
 27
 28    private bool _isShutdown;
 29    private readonly int _maxSslBufferSize;
 23530    private readonly List<ArraySegment<byte>> _segments = new();
 31    private readonly IMemoryOwner<byte>? _writeBufferOwner;
 32
 33    public Task<TransportConnectionInformation> ConnectAsync(CancellationToken cancellationToken)
 22234    {
 22235        ObjectDisposedException.ThrowIf(_isDisposed, this);
 22236        return ConnectAsyncCore(cancellationToken);
 22237    }
 38
 39    public void Dispose()
 25640    {
 25641        _isDisposed = true;
 42
 25643        if (SslStream is SslStream sslStream)
 6644        {
 6645            sslStream.Dispose();
 6646        }
 47
 48        // If shutdown was called, we can just dispose the connection to complete the graceful TCP closure. Otherwise,
 49        // we abort the TCP connection to ensure the connection doesn't end up in the TIME_WAIT state.
 25650        if (_isShutdown)
 2751        {
 2752            Socket.Dispose();
 2753        }
 54        else
 22955        {
 22956            Socket.Close(0);
 22957        }
 25658        _writeBufferOwner?.Dispose();
 25659    }
 60
 61    public ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken)
 935562    {
 935563        ObjectDisposedException.ThrowIf(_isDisposed, this);
 64
 935565        return buffer.Length > 0 ? PerformReadAsync() :
 935566            throw new ArgumentException($"The {nameof(buffer)} cannot be empty.", nameof(buffer));
 67
 68        async ValueTask<int> PerformReadAsync()
 935269        {
 70            try
 935271            {
 935272                return SslStream is SslStream sslStream ?
 935273                    await SslStream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false) :
 935274                    await Socket.ReceiveAsync(buffer, SocketFlags.None, cancellationToken).ConfigureAwait(false);
 75            }
 676            catch (IOException exception)
 677            {
 678                throw exception.ToIceRpcException();
 79            }
 2080            catch (SocketException exception)
 2081            {
 2082                throw exception.ToIceRpcException();
 83            }
 930784        }
 935285    }
 86
 87    public Task ShutdownWriteAsync(CancellationToken cancellationToken)
 2788    {
 2789        ObjectDisposedException.ThrowIf(_isDisposed, this);
 90
 2791        return PerformShutdownAsync();
 92
 93        async Task PerformShutdownAsync()
 2794        {
 95            try
 2796            {
 2797                if (SslStream is SslStream sslStream)
 998                {
 999                    Task shutdownTask = sslStream.ShutdownAsync();
 100
 101                    try
 9102                    {
 9103                        await shutdownTask.WaitAsync(cancellationToken).ConfigureAwait(false);
 9104                    }
 0105                    catch (OperationCanceledException)
 0106                    {
 0107                        await AbortAndObserveAsync(shutdownTask).ConfigureAwait(false);
 0108                        throw;
 109                    }
 110                }
 111
 112                // Shutdown the socket send side to send a TCP FIN packet. We don't close the read side because we want
 113                // to be notified when the peer shuts down it's side of the socket (through the ReceiveAsync call).
 27114                Socket.Shutdown(SocketShutdown.Send);
 115
 116                // If shutdown is successful mark the connection as shutdown to ensure Dispose won't reset the TCP
 117                // connection.
 27118                _isShutdown = true;
 119            }
 0120            catch (IOException exception)
 121            {
 0122                throw exception.ToIceRpcException();
 123            }
 0124            catch (SocketException exception)
 125            {
 0126                throw exception.ToIceRpcException();
 127            }
 128        }
 54129    }
 130
 131    public ValueTask WriteAsync(ReadOnlySequence<byte> buffer, CancellationToken cancellationToken)
 111132    {
 111133        ObjectDisposedException.ThrowIf(_isDisposed, this);
 134
 111135        if (buffer.IsEmpty)
 3136        {
 3137            throw new ArgumentException($"The {nameof(buffer)} cannot be empty.", nameof(buffer));
 138        }
 139
 108140        return PerformWriteAsync();
 141
 142        async ValueTask PerformWriteAsync()
 108143        {
 144            try
 108145            {
 108146                if (SslStream is SslStream sslStream)
 21147                {
 21148                    if (buffer.IsSingleSegment)
 19149                    {
 19150                        await sslStream.WriteAsync(buffer.First, cancellationToken).ConfigureAwait(false);
 15151                    }
 152                    else
 2153                    {
 154                        // Coalesce leading segments up to _maxSslBufferSize. We don't coalesce trailing segments as we
 155                        // assume these segments are large enough.
 2156                        int leadingSize = 0;
 2157                        int leadingSegmentCount = 0;
 304158                        foreach (ReadOnlyMemory<byte> memory in buffer)
 149159                        {
 149160                            if (leadingSize + memory.Length <= _maxSslBufferSize)
 149161                            {
 149162                                leadingSize += memory.Length;
 149163                                leadingSegmentCount++;
 149164                            }
 165                            else
 0166                            {
 0167                                break;
 168                            }
 149169                        }
 170
 2171                        if (leadingSegmentCount > 1)
 2172                        {
 2173                            ReadOnlySequence<byte> leading = buffer.Slice(0, leadingSize);
 2174                            buffer = buffer.Slice(leadingSize); // buffer can become empty
 175
 2176                            Debug.Assert(_writeBufferOwner is not null);
 2177                            Memory<byte> writeBuffer = _writeBufferOwner.Memory[0..leadingSize];
 2178                            leading.CopyTo(writeBuffer.Span);
 179
 180                            // Send the "coalesced" leading segments
 2181                            await sslStream.WriteAsync(writeBuffer, cancellationToken).ConfigureAwait(false);
 2182                        }
 183                        // else no need to coalesce (copy) a single segment
 184
 185                        // Send the remaining segments one by one
 2186                        if (buffer.IsEmpty)
 2187                        {
 188                            // done
 2189                        }
 0190                        else if (buffer.IsSingleSegment)
 0191                        {
 0192                            await sslStream.WriteAsync(buffer.First, cancellationToken).ConfigureAwait(false);
 0193                        }
 194                        else
 0195                        {
 0196                            foreach (ReadOnlyMemory<byte> memory in buffer)
 0197                            {
 0198                                await sslStream.WriteAsync(memory, cancellationToken).ConfigureAwait(false);
 0199                            }
 0200                        }
 2201                    }
 17202                }
 203                else
 87204                {
 87205                    if (buffer.IsSingleSegment)
 83206                    {
 83207                        int bytesSent = await Socket.SendAsync(
 83208                            buffer.First,
 83209                            SocketFlags.None,
 83210                            cancellationToken).ConfigureAwait(false);
 211
 75212                        if (bytesSent != buffer.First.Length)
 0213                        {
 214                            // This should never happen.
 0215                            throw new IceRpcException(
 0216                                IceRpcError.IceRpcError,
 0217                                $"Short write on TCP socket: expected {buffer.First.Length} bytes but sent {bytesSent}."
 218                        }
 75219                    }
 220                    else
 4221                    {
 4222                        _segments.Clear();
 4223                        long totalBytes = buffer.Length;
 608224                        foreach (ReadOnlyMemory<byte> memory in buffer)
 298225                        {
 298226                            if (MemoryMarshal.TryGetArray(memory, out ArraySegment<byte> segment))
 298227                            {
 298228                                _segments.Add(segment);
 298229                            }
 230                            else
 0231                            {
 0232                                throw new ArgumentException(
 0233                                    $"The {nameof(buffer)} must be backed by arrays.",
 0234                                    nameof(buffer));
 235                            }
 298236                        }
 237
 4238                        Task<int> sendTask = Socket.SendAsync(_segments, SocketFlags.None);
 239
 240                        int bytesSent;
 241                        try
 4242                        {
 4243                            bytesSent = await sendTask.WaitAsync(cancellationToken).ConfigureAwait(false);
 4244                        }
 0245                        catch (OperationCanceledException)
 0246                        {
 0247                            await AbortAndObserveAsync(sendTask).ConfigureAwait(false);
 0248                            throw;
 249                        }
 250
 4251                        if (bytesSent != totalBytes)
 252                        {
 253                            // This should never happen.
 0254                            throw new IceRpcException(
 0255                                IceRpcError.IceRpcError,
 0256                                $"Short write on TCP socket: expected {totalBytes} bytes but sent {bytesSent}.");
 257                        }
 4258                    }
 259                }
 260            }
 1261            catch (IOException exception)
 262            {
 1263                throw exception.ToIceRpcException();
 264            }
 4265            catch (SocketException exception)
 266            {
 4267                throw exception.ToIceRpcException();
 268            }
 269        }
 204270    }
 271
 235272    private protected TcpConnection(IMemoryOwner<byte>? memoryOwner)
 235273    {
 235274        _writeBufferOwner = memoryOwner;
 275        // When coalescing leading buffers in WriteAsync (SSL only), the upper size limit is the lesser of the size of
 276        // the buffer we rented from the memory pool (typically 4K) and MaxSslDataSize (16K).
 235277        _maxSslBufferSize = Math.Min(memoryOwner?.Memory.Length ?? 0, MaxSslDataSize);
 235278    }
 279
 280    private protected abstract Task<TransportConnectionInformation> ConnectAsyncCore(
 281        CancellationToken cancellationToken);
 282
 283    /// <summary>Aborts the connection and then observes the exception of the provided task.</summary>
 284    private async Task AbortAndObserveAsync(Task task)
 0285    {
 0286        Socket.Close(0);
 287        try
 0288        {
 0289            await task.ConfigureAwait(false);
 0290        }
 0291        catch
 0292        {
 293            // observe exception
 0294        }
 0295    }
 296}
 297
 298internal class TcpClientConnection : TcpConnection
 299{
 300    internal override Socket Socket { get; }
 301
 302    internal override SslStream? SslStream => _sslStream;
 303
 304    private readonly EndPoint _address;
 305    private readonly SslClientAuthenticationOptions? _authenticationOptions;
 306
 307    private SslStream? _sslStream;
 308
 309    internal TcpClientConnection(
 310        TransportAddress transportAddress,
 311        SslClientAuthenticationOptions? authenticationOptions,
 312        MemoryPool<byte> pool,
 313        int minimumSegmentSize,
 314        TcpClientTransportOptions options)
 315        : base(authenticationOptions is not null ? pool.Rent(minimumSegmentSize) : null)
 316    {
 317        _address = IPAddress.TryParse(transportAddress.Host, out IPAddress? ipAddress) ?
 318            new IPEndPoint(ipAddress, transportAddress.Port) :
 319            new DnsEndPoint(transportAddress.Host, transportAddress.Port);
 320
 321        _authenticationOptions = authenticationOptions;
 322
 323        // When using IPv6 address family we use the socket constructor without AddressFamily parameter to ensure
 324        // dual-mode socket are used in platforms that support them.
 325        Socket = ipAddress?.AddressFamily == AddressFamily.InterNetwork ?
 326            new Socket(ipAddress.AddressFamily, SocketType.Stream, ProtocolType.Tcp) :
 327            new Socket(SocketType.Stream, ProtocolType.Tcp);
 328
 329        try
 330        {
 331            if (options.LocalNetworkAddress is IPEndPoint localNetworkAddress)
 332            {
 333                Socket.Bind(localNetworkAddress);
 334            }
 335
 336            Socket.Configure(options);
 337        }
 338        catch (SocketException exception)
 339        {
 340            Dispose();
 341            throw exception.ToIceRpcException();
 342        }
 343        catch
 344        {
 345            Dispose();
 346            throw;
 347        }
 348    }
 349
 350    private protected override async Task<TransportConnectionInformation> ConnectAsyncCore(
 351        CancellationToken cancellationToken)
 352    {
 353        bool isConnected = false;
 354        try
 355        {
 356            Debug.Assert(Socket is not null);
 357
 358            // Connect to the peer.
 359            await Socket.ConnectAsync(_address, cancellationToken).ConfigureAwait(false);
 360            isConnected = true;
 361
 362            if (_authenticationOptions is not null)
 363            {
 364                _sslStream = new SslStream(new NetworkStream(Socket, false), false);
 365
 366                await _sslStream.AuthenticateAsClientAsync(
 367                    _authenticationOptions,
 368                    cancellationToken).ConfigureAwait(false);
 369            }
 370
 371            return new TransportConnectionInformation(
 372                localNetworkAddress: Socket.LocalEndPoint!,
 373                remoteNetworkAddress: Socket.RemoteEndPoint!,
 374                _sslStream?.RemoteCertificate);
 375        }
 376        catch (IOException exception)
 377        {
 378            throw exception.ToIceRpcException();
 379        }
 380        catch (SocketException exception) when (isConnected)
 381        {
 382            // This can happen if the peer closes the connection immediately after accepting it, which can cause the
 383            // endpoint information to be unavailable. Any SocketException at this point means the connection is no
 384            // longer usable.
 385            throw new IceRpcException(IceRpcError.ConnectionAborted, exception);
 386        }
 387        catch (SocketException exception)
 388        {
 389            throw exception.ToIceRpcException();
 390        }
 391    }
 392}
 393
 394internal class TcpServerConnection : TcpConnection
 395{
 396    internal override Socket Socket { get; }
 397
 398    internal override SslStream? SslStream => _sslStream;
 399
 400    private readonly SslServerAuthenticationOptions? _authenticationOptions;
 401    private SslStream? _sslStream;
 402
 403    internal TcpServerConnection(
 404        Socket socket,
 405        SslServerAuthenticationOptions? authenticationOptions,
 406        MemoryPool<byte> pool,
 407        int minimumSegmentSize)
 408        : base(authenticationOptions is not null ? pool.Rent(minimumSegmentSize) : null)
 409    {
 410        Socket = socket;
 411        _authenticationOptions = authenticationOptions;
 412    }
 413
 414    private protected override async Task<TransportConnectionInformation> ConnectAsyncCore(
 415        CancellationToken cancellationToken)
 416    {
 417        try
 418        {
 419            if (_authenticationOptions is not null)
 420            {
 421                // This can only be created with a connected socket.
 422                _sslStream = new SslStream(new NetworkStream(Socket, false), false);
 423                await _sslStream.AuthenticateAsServerAsync(
 424                    _authenticationOptions,
 425                    cancellationToken).ConfigureAwait(false);
 426            }
 427
 428            return new TransportConnectionInformation(
 429                localNetworkAddress: Socket.LocalEndPoint!,
 430                remoteNetworkAddress: Socket.RemoteEndPoint!,
 431                _sslStream?.RemoteCertificate);
 432        }
 433        catch (IOException exception)
 434        {
 435            throw exception.ToIceRpcException();
 436        }
 437        catch (SocketException exception)
 438        {
 439            throw exception.ToIceRpcException();
 440        }
 441    }
 442}
Generated by: ReportGenerator 5.5.11.0
09/17/2026 - 16:01:28
GitHub | reportgenerator.io

Methods/Properties

.ctor(System.Buffers.IMemoryOwner`1<System.Byte>)
ConnectAsync(System.Threading.CancellationToken)
Dispose()
ReadAsync(System.Memory`1<System.Byte>,System.Threading.CancellationToken)
PerformReadAsync()
ShutdownWriteAsync(System.Threading.CancellationToken)
PerformShutdownAsync()
WriteAsync(System.Buffers.ReadOnlySequence`1<System.Byte>,System.Threading.CancellationToken)
PerformWriteAsync()
AbortAndObserveAsync()