< Summary - CoreWCF Coverage — PR #1766

Information
Class: CoreWCF.Channels.Framing.ServerSingletonConnectionReaderMiddleware
Assembly: CoreWCF.NetFramingBase
File(s): /home/runner/work/CoreWCF/CoreWCF/src/CoreWCF.NetFramingBase/src/CoreWCF/Channels/Framing/ServerSingletonConnectionReaderMiddleware.cs
Line coverage
68%
Covered lines: 126
Uncovered lines: 59
Coverable lines: 185
Total lines: 486
Line coverage: 68.1%
Branch coverage
66%
Covered branches: 48
Total branches: 72
Branch coverage: 66.6%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Cyclomatic complexity NPath complexity Sequence coverage
.ctor(...)100%11100%
OnConnectedAsync()75%4496.15%
ReceiveRequestAsync()100%11100%
ReceiveAsync()42.85%141464.28%
PrepareMessage(...)100%44100%
DecodeBytes(...)70%101070%
.ctor(...)100%11100%
AbortReader()100%110%
Close()100%11100%
DecodeData(...)100%22100%
DecodeSize(...)87.5%8890%
EnsureBuffer(...)100%11100%
EnsureBufferAsync()87.5%8875%
Flush()100%110%
Read(...)100%8892%
ReadAsync()0%880%
BeginRead(...)100%110%
EndRead(...)100%110%
Seek(...)100%110%
SetLength(...)100%110%
ProcessEof()66.66%6680%
Write(...)100%110%

File(s)

/home/runner/work/CoreWCF/CoreWCF/src/CoreWCF.NetFramingBase/src/CoreWCF/Channels/Framing/ServerSingletonConnectionReaderMiddleware.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.Buffers;
 6using System.Collections;
 7using System.IO;
 8using System.IO.Pipelines;
 9using System.Net;
 10using System.Threading;
 11using System.Threading.Tasks;
 12using System.Xml;
 13using CoreWCF.Configuration;
 14using CoreWCF.Runtime;
 15using CoreWCF.Security;
 16using Microsoft.Extensions.DependencyInjection;
 17
 18namespace CoreWCF.Channels.Framing
 19{
 20    internal class ServerSingletonConnectionReaderMiddleware
 21    {
 22        private readonly HandshakeDelegate _next;
 8323        private readonly Hashtable _serviceChannelDispatcherCache = new Hashtable();
 24        private readonly IServiceScopeFactory _servicesScopeFactory;
 8325        private readonly AsyncLock _lock = new AsyncLock();
 26
 8327        public ServerSingletonConnectionReaderMiddleware(HandshakeDelegate next, IServiceScopeFactory servicesScopeFacto
 28        {
 8329            _next = next;
 8330            _servicesScopeFactory = servicesScopeFactory;
 8331        }
 32
 33        public async Task OnConnectedAsync(FramingConnection connection)
 34        {
 35            IServiceChannelDispatcher channelDispatcher;
 5036            if (_serviceChannelDispatcherCache.ContainsKey(connection.ServiceDispatcher))
 37            {
 2238                channelDispatcher = (IServiceChannelDispatcher)_serviceChannelDispatcherCache[connection.ServiceDispatch
 39            }
 40            else
 41            {
 2842                await using (await _lock.TakeLockAsync())
 43                {
 2844                    if (_serviceChannelDispatcherCache.ContainsKey(connection.ServiceDispatcher))
 45                    {
 046                        channelDispatcher = (IServiceChannelDispatcher)_serviceChannelDispatcherCache[connection.Service
 47                    }
 48                    else
 49                    {
 2850                        BindingElementCollection be = connection.ServiceDispatcher.Binding.CreateBindingElements();
 2851                        TransportBindingElement tbe = be.Find<TransportBindingElement>();
 2852                        ITransportFactorySettings settings = new NetFramingTransportSettings
 2853                        {
 2854                            CloseTimeout = connection.ServiceDispatcher.Binding.CloseTimeout,
 2855                            OpenTimeout = connection.ServiceDispatcher.Binding.OpenTimeout,
 2856                            ReceiveTimeout = connection.ServiceDispatcher.Binding.ReceiveTimeout,
 2857                            SendTimeout = connection.ServiceDispatcher.Binding.SendTimeout,
 2858                            ManualAddressing = tbe.ManualAddressing,
 2859                            BufferManager = connection.BufferManager,
 2860                            MaxReceivedMessageSize = tbe.MaxReceivedMessageSize,
 2861                            MessageEncoderFactory = connection.MessageEncoderFactory
 2862                        };
 63                        // Even though channel is reused for multiple connections, there are some scoped dependencies us
 2864                        var replyChannel = new ConnectionOrientedTransportReplyChannel(settings, null, _servicesScopeFac
 2865                        channelDispatcher = await connection.ServiceDispatcher.CreateServiceChannelDispatcherAsync(reply
 2866                        _serviceChannelDispatcherCache[connection.ServiceDispatcher] = channelDispatcher;
 67                    }
 68                }
 69            }
 70
 71            // TODO: I think that the receive timeout starts counting at the start of the preamble on .NET Framework. Th
 72            // after the preamble has completed. This probably needs to be addressed otherwise worse case you could end 
 73            // I believe the preamble should really use the OpenTimeout but that's not how this is implemented on .NET F
 5074            var timeoutHelper = new TimeoutHelper(connection.ServiceDispatcher.Binding.ReceiveTimeout);
 5075            StreamedFramingRequestContext requestContext = await ReceiveRequestAsync(connection, timeoutHelper.Remaining
 5076            await channelDispatcher.DispatchAsync(requestContext);
 5077            await requestContext.ReplySent;
 5078        }
 79
 80        public async Task<StreamedFramingRequestContext> ReceiveRequestAsync(FramingConnection connection, TimeSpan time
 81        {
 5082            (Message requestMessage, Stream inputStream) = await ReceiveAsync(connection, timeout);
 5083            return new StreamedFramingRequestContext(connection, requestMessage, inputStream);
 5084        }
 85
 86        public async Task<(Message, Stream)> ReceiveAsync(FramingConnection connection, TimeSpan timeout)
 87        {
 5088            TimeoutHelper timeoutHelper = new TimeoutHelper(timeout);
 5089            ReadOnlySequence<byte> buffer = ReadOnlySequence<byte>.Empty;
 90            for (; ; )
 91            {
 5092                ReadResult readResult = await connection.Input.ReadAsync();
 5093                await Task.Yield();
 5094                if (readResult.IsCompleted || readResult.Buffer.Length == 0)
 95                {
 096                    if (!readResult.IsCompleted)
 97                    {
 098                        connection.Input.AdvanceTo(readResult.Buffer.Start);
 99                    }
 100                    //EnsureDecoderAtEof(connection);
 0101                    connection.EOF = true;
 102                }
 103
 50104                if (connection.EOF)
 105                {
 0106                    return (null, null);
 107                }
 108
 50109                buffer = readResult.Buffer;
 50110                bool atEnvelopeStart = DecodeBytes(connection, ref buffer);
 50111                connection.Input.AdvanceTo(buffer.Start);
 50112                if (atEnvelopeStart)
 113                {
 114                    break;
 115                }
 116
 117
 0118                if (connection.EOF)
 119                {
 0120                    return (null, null);
 121                }
 0122            }
 123
 124            // we're ready to read a message
 50125            Stream connectionStream = new SingletonInputConnectionStream(connection, connection.ServiceDispatcher.Bindin
 50126            Stream inputStream = new MaxMessageSizeStream(connectionStream, connection.MaxReceivedMessageSize);
 127            //using (ServiceModelActivity activity = DiagnosticUtility.ShouldUseActivity ? ServiceModelActivity.CreateBo
 128            //{
 129            //    if (DiagnosticUtility.ShouldUseActivity)
 130            //    {
 131            //        ServiceModelActivity.Start(activity, SR.GetString(SR.ActivityProcessingMessage, TraceUtility.Retri
 132            //    }
 133
 134            Message message;
 135            try
 136            {
 50137                message = await connection.MessageEncoderFactory.Encoder.ReadMessageAsync(
 50138                    inputStream, connection.MaxBufferSize, connection.FramingDecoder.ContentType);
 50139            }
 0140            catch (XmlException xmlException)
 141            {
 0142                throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(
 0143                    new ProtocolException(SR.MessageXmlProtocolError, xmlException));
 144            }
 145
 146            //if (DiagnosticUtility.ShouldUseActivity)
 147            //{
 148            //    TraceUtility.TransferFromTransport(message);
 149            //}
 150
 50151            PrepareMessage(connection, message);
 152
 50153            return (message, inputStream);
 154            //}
 50155        }
 156
 157        private void PrepareMessage(FramingConnection connection, Message message)
 158        {
 50159            message.Properties.Via = connection.Via;
 50160            message.Properties.Security = (connection.SecurityMessageProperty != null) ? (SecurityMessageProperty)connec
 161
 50162            IPEndPoint remoteEndPoint = connection.RemoteEndpoint;
 163
 164            // pipes will return null
 50165            if (remoteEndPoint != null)
 166            {
 50167                var remoteEndpointProperty = new RemoteEndpointMessageProperty(remoteEndPoint);
 50168                message.Properties.Add(RemoteEndpointMessageProperty.Name, remoteEndpointProperty);
 169            }
 170
 171            // TODO: ChannelBindingToken
 172            //if (this.channelBindingToken != null)
 173            //{
 174            //    ChannelBindingMessageProperty property = new ChannelBindingMessageProperty(this.channelBindingToken, f
 175            //    property.AddTo(message);
 176            //    property.Dispose(); //message.Properties.Add() creates a copy...
 177            //}
 50178        }
 179
 180        private bool DecodeBytes(FramingConnection connection, ref ReadOnlySequence<byte> buffer)
 181        {
 50182            var decoder = connection.FramingDecoder as ServerSingletonDecoder;
 183            Fx.Assert(decoder != null, "FramingDecoder must be a non-null ServerSingletonDecoder");
 100184            while (!connection.EOF && buffer.Length > 0)
 185            {
 100186                int bytesRead = decoder.Decode(buffer);
 100187                if (bytesRead > 0)
 188                {
 50189                    buffer = buffer.Slice(bytesRead);
 190                }
 191
 100192                switch (decoder.CurrentState)
 193                {
 194                    case ServerSingletonDecoder.State.EnvelopeStart:
 195                        // we're at the envelope
 50196                        return true;
 197
 198                    case ServerSingletonDecoder.State.End:
 0199                        connection.EOF = true;
 0200                        return false;
 201                }
 202            }
 203
 0204            return false;
 205        }
 206
 207        // ensures that the reader is notified at end-of-stream, and takes care of the framing chunk headers
 208        private class SingletonInputConnectionStream : Stream
 209        {
 210            private readonly FramingConnection _connection;
 211            private readonly IDefaultCommunicationTimeouts _timeouts;
 212            private readonly SingletonMessageDecoder _decoder;
 50213            private ReadOnlySequence<byte> _buffer = ReadOnlySequence<byte>.Empty;
 214            private bool _atEof;
 215            private int _chunkBytesRemaining;
 216            private TimeoutHelper _timeoutHelper;
 217
 50218            public SingletonInputConnectionStream(FramingConnection connection,
 50219                IDefaultCommunicationTimeouts defaultTimeouts)
 220            {
 50221                _connection = connection;
 50222                _timeouts = defaultTimeouts;
 50223                _decoder = new SingletonMessageDecoder(connection.Logger);
 50224                _chunkBytesRemaining = 0;
 50225                _timeoutHelper = new TimeoutHelper(_timeouts.ReceiveTimeout);
 50226            }
 227
 12228            public override bool CanRead => true;
 229
 0230            public override bool CanSeek => false;
 231
 0232            public override bool CanWrite => false;
 233
 0234            public override long Length => throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new NotSupportedExc
 235
 236            public override long Position
 237            {
 238                get
 239                {
 0240                    throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new NotSupportedException(SR.SeekNotSuppor
 241                }
 242                set
 243                {
 0244                    throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new NotSupportedException(SR.SeekNotSuppor
 245                }
 246            }
 247
 248            private void AbortReader()
 249            {
 0250                _connection.Abort();
 0251            }
 252
 253            public override void Close()
 254            {
 50255                _connection.EOF = _atEof;
 50256            }
 257
 258            // run chunk data through the decoder
 259            private void DecodeData(ReadOnlySequence<byte> buffer)
 260            {
 5278261                while (buffer.Length > 0)
 262                {
 2832263                    int bytesRead = _decoder.Decode(buffer);
 2832264                    buffer = buffer.Slice(bytesRead);
 265                    Fx.Assert(_decoder.CurrentState == SingletonMessageDecoder.State.ReadingEnvelopeBytes || _decoder.Cu
 266                }
 2446267            }
 268
 269            // run the current data through the decoder to get valid message bytes
 270            private void DecodeSize(ref ReadOnlySequence<byte> buffer)
 271            {
 1043272                while (buffer.Length > 0)
 273                {
 1043274                    int bytesRead = _decoder.Decode(buffer);
 275
 1043276                    if (bytesRead > 0)
 277                    {
 507278                        buffer = buffer.Slice(bytesRead);
 279                    }
 280
 1043281                    switch (_decoder.CurrentState)
 282                    {
 283                        case SingletonMessageDecoder.State.ChunkStart:
 386284                            _chunkBytesRemaining = _decoder.ChunkSize;
 386285                            return;
 286                        case SingletonMessageDecoder.State.End:
 50287                            ProcessEof();
 50288                            return;
 289                    }
 290                }
 0291            }
 292
 293            private void EnsureBuffer(CancellationToken token)
 294            {
 2982295                EnsureBufferAsync(token).GetAwaiter().GetResult();
 2982296            }
 297
 298            private async Task EnsureBufferAsync(CancellationToken token)
 299            {
 2982300                if (_buffer.Length == 0 && !_atEof)
 301                {
 92302                    if (!_connection.Input.TryRead(out ReadResult readResult))
 303                    {
 43304                        readResult = await _connection.Input.ReadAsync(token).ConfigureAwait(false);
 305                    }
 306
 92307                    if (readResult.IsCompleted)
 308                    {
 0309                        _atEof = true;
 0310                        return;
 311                    }
 312
 92313                    _buffer = readResult.Buffer;
 314                }
 2982315            }
 316
 0317            public override void Flush() { /* NOP */ }
 318
 319            public override int Read(byte[] buffer, int offset, int count)
 320            {
 321                // TODO: Create a ReadByte override which is optimized for that single case
 2210322                CancellationToken ct = _timeoutHelper.GetCancellationToken();
 2210323                int result = 0;
 92324                while (true)
 325                {
 5092326                    if (count == 0)
 327                    {
 2110328                        return result;
 329                    }
 330
 331                    try
 332                    {
 2982333                        EnsureBuffer(ct);
 2982334                    }
 0335                    catch (OperationCanceledException oce)
 336                    {
 0337                        throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new TimeoutException(SR.Format(SR.Rece
 338                    }
 339
 2982340                    if (_atEof)
 341                    {
 100342                        return result;
 343                    }
 344
 2882345                    if (_chunkBytesRemaining > 0) // We're in the middle of a chunk.
 346                    {
 347                        // How many bytes to copy into the buffer passed to this method. The read from the input pipe mi
 348                        // from the next chunk and we're not ready to consume them yet. Also we can't copy more bytes th
 2446349                        int bytesToCopy = Math.Min((int)Math.Min((int)_buffer.Length, _chunkBytesRemaining), count);
 350
 351                        // When copying a ReadOnlySequence to a Span, they must be the same size so create a temporary
 352                        // ReadOnlySequence which has the same number of bytes as we wish to copy.
 2446353                        ReadOnlySequence<byte> _fromBuffer = _buffer.Slice(_buffer.Start, bytesToCopy);
 354
 355                        // keep decoder up to date
 2446356                        DecodeData(_fromBuffer);
 357
 358                        // Consume those bytes from our buffer
 2446359                        _buffer = _buffer.Slice(bytesToCopy);
 360
 361                        // TODO: Possible perf improvement would be to call ReadAsync and save the Task<ReadResult> with
 362                        // likely have been completed before the next call to avoid blocking waiting for the Task to com
 363
 364                        // Create Span of the right size to copy the bytes to.
 2446365                        var _toBuffer = new Span<byte>(buffer, offset, bytesToCopy);
 2446366                        _fromBuffer.CopyTo(_toBuffer);
 367                        // Fix up counts
 2446368                        result += bytesToCopy;
 2446369                        offset += bytesToCopy;
 2446370                        count -= bytesToCopy;
 2446371                        _chunkBytesRemaining -= bytesToCopy;
 372                    }
 373                    else
 374                    {
 375                        // We are starting a new chunk. Read the size, and loop around again
 436376                        DecodeSize(ref _buffer);
 377                    }
 378
 379                    // If the buffer has been exhausted, advance the input pipe to consume them and release the buffer
 2882380                    if (_buffer.Length == 0)
 381                    {
 92382                        _connection.Input.AdvanceTo(_buffer.End);
 383                    }
 384
 385                    //if (atEof)
 386                    //{
 387                    //    _connection.Input.Complete();
 388                    //}
 389                }
 390            }
 391
 392            public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellati
 393            {
 0394                CancellationToken ct = new TimeoutHelper(_timeouts.ReceiveTimeout).GetCancellationToken();
 0395                int result = 0;
 0396                while (true)
 397                {
 0398                    if (count == 0)
 399                    {
 0400                        return result;
 401                    }
 402
 403                    try
 404                    {
 0405                        await EnsureBufferAsync(ct);
 0406                    }
 0407                    catch (OperationCanceledException oce)
 408                    {
 0409                        throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new TimeoutException(SR.Format(SR.Rece
 410                    }
 411
 0412                    if (_atEof)
 413                    {
 0414                        return result;
 415                    }
 416
 0417                    if (_chunkBytesRemaining > 0) // We're in the middle of a chunk.
 418                    {
 419                        // How many bytes to copy into the buffer passed to this method. The read from the input pipe mi
 420                        // from the next chunk and we're not ready to consume them yet. Also we can't copy more bytes th
 0421                        int bytesToCopy = Math.Min((int)Math.Min((int)_buffer.Length, _chunkBytesRemaining), count);
 422
 423                        // When copying a ReadOnlySequence to a Span, they must be the same size so create a temporary
 424                        // ReadOnlySequence which has the same number of bytes as we wish to copy.
 0425                        ReadOnlySequence<byte> _fromBuffer = _buffer.Slice(_buffer.Start, bytesToCopy);
 426
 427                        // keep decoder up to date
 0428                        DecodeData(_fromBuffer);
 429
 430                        // Consume those bytes from our buffer
 0431                        _buffer = _buffer.Slice(bytesToCopy);
 432
 433                        // Create an ArraySegment of the right size to copy the bytes to. The synchronous Read method us
 434                        // you can't instantiate a Span<T> in an async method but there's an implicit case of ArraySegme
 0435                        var _toBuffer = new ArraySegment<byte>(buffer, offset, bytesToCopy);
 0436                        _fromBuffer.CopyTo(_toBuffer);
 437                        // Fix up counts
 0438                        result += bytesToCopy;
 0439                        offset += bytesToCopy;
 0440                        count -= bytesToCopy;
 0441                        _chunkBytesRemaining -= bytesToCopy;
 442                    }
 443                    else
 444                    {
 445                        // We are starting a new chunk. Read the size, and loop around again
 0446                        DecodeSize(ref _buffer);
 447                    }
 448
 449                    // If the buffer has been exhausted, advance the input pipe to consume them and release the buffer
 0450                    if (_buffer.Length == 0)
 451                    {
 0452                        _connection.Input.AdvanceTo(_buffer.End);
 453                    }
 454                }
 0455            }
 456
 457            public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback callback, object 
 458            {
 0459                return ReadAsync(buffer, offset, count).ToApm(callback, state);
 460            }
 461
 462            public override int EndRead(IAsyncResult result)
 463            {
 0464                return result.ToApmEnd<int>();
 465            }
 466
 0467            public override long Seek(long offset, SeekOrigin origin) => throw DiagnosticUtility.ExceptionUtility.ThrowH
 468
 0469            public override void SetLength(long value) => throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new 
 470
 471            private void ProcessEof()
 472            {
 50473                if (!_atEof)
 474                {
 50475                    _atEof = true;
 50476                    if (_chunkBytesRemaining > 0 || _decoder.CurrentState != SingletonMessageDecoder.State.End)
 477                    {
 0478                        throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(_decoder.CreatePrematureEOFException()
 479                    }
 480                }
 50481            }
 482
 0483            public override void Write(byte[] buffer, int offset, int count) => throw new NotImplementedException();
 484        }
 485    }
 486}
Generated by: ReportGenerator 5.5.11.0
08/09/2026 - 10:15:48
GitHub | reportgenerator.io

Methods/Properties

.ctor(CoreWCF.Configuration.HandshakeDelegate,Microsoft.Extensions.DependencyInjection.IServiceScopeFactory)
OnConnectedAsync()
ReceiveRequestAsync()
ReceiveAsync()
PrepareMessage(CoreWCF.Channels.Framing.FramingConnection,CoreWCF.Channels.Message)
DecodeBytes(CoreWCF.Channels.Framing.FramingConnection,System.Buffers.ReadOnlySequence`1<System.Byte>&)
.ctor(CoreWCF.Channels.Framing.FramingConnection,CoreWCF.IDefaultCommunicationTimeouts)
CanRead()
CanSeek()
CanWrite()
Length()
Position()
Position(System.Int64)
AbortReader()
Close()
DecodeData(System.Buffers.ReadOnlySequence`1<System.Byte>)
DecodeSize(System.Buffers.ReadOnlySequence`1<System.Byte>&)
EnsureBuffer(System.Threading.CancellationToken)
EnsureBufferAsync()
Flush()
Read(System.Byte[],System.Int32,System.Int32)
ReadAsync()
BeginRead(System.Byte[],System.Int32,System.Int32,System.AsyncCallback,System.Object)
EndRead(System.IAsyncResult)
Seek(System.Int64,System.IO.SeekOrigin)
SetLength(System.Int64)
ProcessEof()
Write(System.Byte[],System.Int32,System.Int32)