< Summary - CoreWCF Coverage — PR #1766

Information
Class: CoreWCF.Channels.NamedPipeConnectionContext
Assembly: CoreWCF.NetNamedPipe
File(s): /home/runner/work/CoreWCF/CoreWCF/src/CoreWCF.NetNamedPipe/src/CoreWCF/Channels/NamedPipeConnectionContext.cs
Line coverage
0%
Covered lines: 0
Uncovered lines: 141
Coverable lines: 141
Total lines: 347
Line coverage: 0%
Branch coverage
0%
Covered branches: 0
Total branches: 42
Branch coverage: 0%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Cyclomatic complexity NPath complexity Sequence coverage
.cctor()100%110%
.ctor(...)100%110%
Start()100%110%
DoReceiveAsync()0%18180%
DoSendAsync()0%880%
WriteAsync()0%880%
Shutdown(...)0%440%
FireConnectionClosed()0%220%
CancelConnectionClosedToken()100%110%
DisposeAsync()100%110%
Abort(...)100%110%
CreateTransport(...)100%110%
CreateApplication(...)100%110%

File(s)

/home/runner/work/CoreWCF/CoreWCF/src/CoreWCF.NetNamedPipe/src/CoreWCF/Channels/NamedPipeConnectionContext.cs

#LineLine coverage
 1// Licensed to the .NET Foundation under one or more agreements.
 2// The .NET Foundation licenses this file to you under the MIT license.
 3
 4using System;
 5using System.Diagnostics;
 6using System.IO.Pipelines;
 7using System.IO.Pipes;
 8using System.Runtime.InteropServices;
 9using System.Threading;
 10using System.Threading.Tasks;
 11using CoreWCF.Runtime;
 12using Microsoft.AspNetCore.Connections;
 13using PipeOptions = System.IO.Pipelines.PipeOptions;
 14
 15namespace CoreWCF.Channels
 16{
 17    // When we eventually move to .NET 6, implement IThreadPoolWorkItem
 18    internal class NamedPipeConnectionContext : DefaultConnectionContext, IAsyncDisposable
 19    {
 020        private static readonly ConnectionAbortedException s_sendGracefullyCompletedException = new ConnectionAbortedExc
 21
 22        private NamedPipeServerStream _pipe;
 23        private string _connectionId;
 24        private IDuplexPipe _originalTransport;
 25        private int _connectionBufferSize;
 26        private byte[] _readBuffer;
 27
 028        private readonly CancellationTokenSource _connectionClosedTokenSource = new CancellationTokenSource();
 29        private bool _connectionShutdown;
 30        private bool _connectionClosed;
 31        private Exception _shutdownReason;
 32        private bool _streamDisconnected;
 033        private readonly object _shutdownLock = new object();
 034        internal Task _receivingTask = Task.CompletedTask;
 035        internal Task _sendingTask = Task.CompletedTask;
 36        private byte[] _writeBuffer;
 37
 038        internal NamedPipeConnectionContext(NamedPipeServerStream pipe, PipeOptions inputOptions, PipeOptions outputOpti
 39        {
 040            _pipe = pipe;
 041            var input = new Pipe(inputOptions);
 042            var output = new Pipe(outputOptions);
 043            Transport = _originalTransport = DuplexPipe.CreateTransport(input, output);
 044            Application = DuplexPipe.CreateApplication(input, output);
 045            _connectionBufferSize = connectionBufferSize;
 046        }
 47
 48        public override string ConnectionId
 49        {
 050            get => _connectionId ??= CorrelationIdGenerator.GetNextId();
 051            set => _connectionId = value;
 52        }
 53
 054        public PipeWriter Input => Application.Output;
 055        public PipeReader Output => Application.Input;
 56
 057        public NetNamedPipeTrace Logger { get; internal set; }
 58
 59        public void Start()
 60        {
 61            try
 62            {
 63                // Spawn send and receive logic
 064                _receivingTask = DoReceiveAsync();
 065                _sendingTask = DoSendAsync();
 066            }
 067            catch (Exception ex)
 68            {
 069                Logger.LogConnectionError(0, ex, $"Unexpected exception in {nameof(NamedPipeConnection)}.{nameof(Start)}
 070            }
 071        }
 72
 73        private async Task DoReceiveAsync()
 74        {
 075            Exception error = null;
 76
 77            try
 78            {
 079                var input = Input;
 080                while (true)
 81                {
 82                    // Ensure we have some reasonable amount of buffer space
 083                    var buffer = input.GetMemory(_connectionBufferSize);
 84                    int bytesReceived;
 085                    if (MemoryMarshal.TryGetArray(buffer, out ArraySegment<byte> byteArray))
 86                    {
 087                        bytesReceived = await _pipe.ReadAsync(byteArray.Array, byteArray.Offset, byteArray.Count);
 88                    }
 89                    else
 90                    {
 091                        _readBuffer ??= Fx.AllocateByteArray(_connectionBufferSize);
 092                        bytesReceived = await _pipe.ReadAsync(_readBuffer, 0, _readBuffer.Length);
 093                        new Memory<byte>(_readBuffer, 0, bytesReceived).CopyTo(buffer);
 94                    }
 95
 096                    if (bytesReceived == 0)
 97                    {
 98                        // Read completed.
 099                        Logger.ConnectionReadEnd(ConnectionId);
 0100                        break;
 101                    }
 102
 0103                    input.Advance(bytesReceived);
 104
 0105                    var flushTask = Input.FlushAsync();
 106
 0107                    var paused = !flushTask.IsCompleted;
 108
 0109                    if (paused)
 110                    {
 0111                        Logger.ConnectionPause(ConnectionId);
 112                    }
 113
 0114                    var result = await flushTask;
 115
 0116                    if (paused)
 117                    {
 0118                        Logger.ConnectionResume(ConnectionId);
 119                    }
 120
 0121                    if (result.IsCompleted || result.IsCanceled)
 122                    {
 123                        // Pipe consumer is shut down, do we stop writing
 124                        break;
 125                    }
 0126                }
 0127            }
 0128            catch (ObjectDisposedException ex)
 129            {
 130                // This exception should always be ignored because _shutdownReason should be set.
 0131                error = ex;
 132
 0133                if (!_connectionShutdown)
 134                {
 135                    // This is unexpected if the socket hasn't been disposed yet.
 0136                    Logger.ConnectionError(ConnectionId, error);
 137                }
 0138            }
 0139            catch (Exception ex)
 140            {
 141                // This is unexpected.
 0142                error = ex;
 0143                Logger.ConnectionError(ConnectionId, error);
 0144            }
 145            finally
 146            {
 147                // If Shutdown() has already been called, assume that was the reason ProcessReceives() exited.
 0148                Input.Complete(_shutdownReason ?? error);
 149
 0150                FireConnectionClosed();
 151            }
 0152        }
 153
 154        private async Task DoSendAsync()
 155        {
 0156            Exception shutdownReason = null;
 0157            Exception unexpectedError = null;
 158
 159            try
 160            {
 0161                while (true)
 162                {
 0163                    var result = await Output.ReadAsync();
 164
 0165                    if (result.IsCanceled)
 166                    {
 167                        break;
 168                    }
 0169                    var buffer = result.Buffer;
 170
 0171                    if (buffer.IsSingleSegment)
 172                    {
 173                        // Fast path when the buffer is a single segment.
 0174                        await WriteAsync(buffer.First);
 175                    }
 176                    else
 177                    {
 0178                        foreach (var segment in buffer)
 179                        {
 0180                            await WriteAsync(segment);
 181                        }
 182                    }
 183
 0184                    Output.AdvanceTo(buffer.End);
 185
 0186                    if (result.IsCompleted)
 187                    {
 188                        break;
 189                    }
 0190                }
 0191            }
 0192            catch (ObjectDisposedException ex)
 193            {
 194                // This should always be ignored since Shutdown() must have already been called by Abort().
 0195                shutdownReason = ex;
 0196            }
 0197            catch (Exception ex)
 198            {
 0199                shutdownReason = ex;
 0200                unexpectedError = ex;
 0201                Logger.ConnectionError(ConnectionId, unexpectedError);
 0202            }
 203            finally
 204            {
 0205                Shutdown(shutdownReason);
 206
 207                // Complete the output after disposing the socket
 0208                Output.Complete(unexpectedError);
 209
 210                // Cancel any pending flushes so that the input loop is un-paused
 0211                Input.CancelPendingFlush();
 212            }
 0213        }
 214
 215        private async ValueTask WriteAsync(ReadOnlyMemory<byte> buffer)
 216        {
 0217            if (MemoryMarshal.TryGetArray(buffer, out ArraySegment<byte> byteArray))
 218            {
 0219                await _pipe.WriteAsync(byteArray.Array, byteArray.Offset, byteArray.Count);
 220            }
 221            else
 222            {
 0223                _writeBuffer ??= Fx.AllocateByteArray(_connectionBufferSize);
 0224                if (buffer.Length <= _connectionBufferSize)
 225                {
 0226                    buffer.CopyTo(_writeBuffer);
 0227                    await _pipe.WriteAsync(_writeBuffer, 0, buffer.Length);
 228                }
 229                else
 230                {
 0231                    while(buffer.Length > _connectionBufferSize)
 232                    {
 0233                        buffer.Slice(0, _connectionBufferSize).CopyTo(_writeBuffer);
 0234                        await _pipe.WriteAsync(_writeBuffer, 0, _connectionBufferSize);
 0235                        buffer = buffer.Slice(_connectionBufferSize);
 236                    }
 0237                    buffer.CopyTo(_writeBuffer);
 0238                    await _pipe.WriteAsync(_writeBuffer, 0, buffer.Length);
 239                }
 240            }
 0241        }
 242
 243        private void Shutdown(Exception shutdownReason)
 244        {
 0245            lock (_shutdownLock)
 246            {
 0247                if (_connectionShutdown)
 248                {
 0249                    return;
 250                }
 251
 252                // Make sure to close the connection only after the _aborted flag is set.
 253                // Without this, the RequestsCanBeAbortedMidRead test will sometimes fail when
 254                // a BadHttpRequestException is thrown instead of a TaskCanceledException.
 0255                _connectionShutdown = true;
 256
 257                // shutdownReason should only be null if the output was completed gracefully, so no one should ever
 258                // ever observe the nondescript ConnectionAbortedException except for connection middleware attempting
 259                // to half close the connection which is currently unsupported.
 0260                _shutdownReason = shutdownReason ?? s_sendGracefullyCompletedException;
 0261                Logger.ConnectionDisconnect(ConnectionId, _shutdownReason.Message);
 262
 263                try
 264                {
 0265                    _pipe.Disconnect();
 0266                    _streamDisconnected = true;
 0267                }
 0268                catch
 269                {
 270                    // Ignore any errors from NamedPipeStream.Disconnect() since we're tearing down the connection anywa
 0271                }
 272            }
 0273        }
 274
 275        private void FireConnectionClosed()
 276        {
 277            // Guard against scheduling this multiple times
 0278            lock (_shutdownLock)
 279            {
 0280                if (_connectionClosed)
 281                {
 0282                    return;
 283                }
 284
 0285                _connectionClosed = true;
 0286            }
 287
 0288            CancelConnectionClosedToken();
 0289        }
 290
 291        private void CancelConnectionClosedToken()
 292        {
 293            try
 294            {
 0295                _connectionClosedTokenSource.Cancel();
 0296            }
 0297            catch (Exception ex)
 298            {
 0299                Logger.LogConnectionError(0, ex, $"Unexpected exception in {nameof(NamedPipeConnectionContext)}.{nameof(
 0300            }
 0301        }
 302
 303        public async ValueTask DisposeAsync()
 304        {
 0305            _originalTransport.Input.Complete();
 0306            _originalTransport.Output.Complete();
 307
 308            try
 309            {
 310                // Now wait for both to complete
 0311                await _receivingTask;
 0312                await _sendingTask;
 0313            }
 0314            catch (Exception ex)
 315            {
 0316                Logger.LogConnectionError(0, ex, $"Unexpected exception in {nameof(NamedPipeConnection)}.{nameof(Start)}
 0317                _pipe.Dispose();
 0318                return;
 319            }
 320
 321            // TODO: Consider pooling NamedPipeServerStream instances
 0322            _pipe.Dispose();
 0323        }
 324
 325        public override void Abort(ConnectionAbortedException abortReason)
 326        {
 327            Debug.Assert(Application != null);
 0328            Application.Input.CancelPendingRead();
 0329        }
 330
 331        private class DuplexPipe : IDuplexPipe
 332        {
 333            public static IDuplexPipe CreateTransport(Pipe input, Pipe output)
 334            {
 0335                return new DuplexPipe { Input = input.Reader, Output = output.Writer };
 336            }
 337
 338            public static IDuplexPipe CreateApplication(Pipe input, Pipe output)
 339            {
 0340                return new DuplexPipe { Input = output.Reader, Output = input.Writer };
 341            }
 342
 0343            public PipeReader Input { get; private set; }
 0344            public PipeWriter Output { get; private set; }
 345        }
 346    }
 347}
Generated by: ReportGenerator 5.5.11.0
08/09/2026 - 10:15:48
GitHub | reportgenerator.io

Methods/Properties

.cctor()
.ctor(System.IO.Pipes.NamedPipeServerStream,System.IO.Pipelines.PipeOptions,System.IO.Pipelines.PipeOptions,System.Int32)
ConnectionId()
ConnectionId(System.String)
Input()
Output()
Logger()
Start()
DoReceiveAsync()
DoSendAsync()
WriteAsync()
Shutdown(System.Exception)
FireConnectionClosed()
CancelConnectionClosedToken()
DisposeAsync()
Abort(Microsoft.AspNetCore.Connections.ConnectionAbortedException)
CreateTransport(System.IO.Pipelines.Pipe,System.IO.Pipelines.Pipe)
CreateApplication(System.IO.Pipelines.Pipe,System.IO.Pipelines.Pipe)
Input()
Output()