< Summary - CoreWCF Coverage — PR #1766

Information
Class: CoreWCF.Channels.Framing.RawStream
Assembly: CoreWCF.NetFramingBase
File(s): /home/runner/work/CoreWCF/CoreWCF/src/CoreWCF.NetFramingBase/src/CoreWCF/Channels/Framing/RawStream.cs
Line coverage
0%
Covered lines: 0
Uncovered lines: 113
Coverable lines: 113
Total lines: 303
Line coverage: 0%
Branch coverage
0%
Covered branches: 0
Total branches: 26
Branch coverage: 0%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Cyclomatic complexity NPath complexity Sequence coverage
.ctor(...)100%110%
Seek(...)100%110%
SetLength(...)100%110%
Read(...)100%110%
ReadAsync(...)100%110%
Write(...)100%110%
WriteAsync()0%220%
Flush()100%110%
FlushAsync(...)100%110%
ReadAsyncInternal()0%10100%
BeginRead(...)0%220%
EndRead(...)100%110%
ReadAsync(...)0%440%
BeginWrite(...)0%220%
EndWrite(...)100%110%
WriteAsync(...)0%440%
StartUnwrapRead()0%220%
FinishUnwrapReadAsync()100%110%

File(s)

/home/runner/work/CoreWCF/CoreWCF/src/CoreWCF.NetFramingBase/src/CoreWCF/Channels/Framing/RawStream.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.IO;
 7using System.IO.Pipelines;
 8using System.Threading;
 9using System.Threading.Tasks;
 10using CoreWCF.Runtime;
 11
 12namespace CoreWCF.Channels.Framing
 13{
 14    public class RawStream : Stream
 15    {
 16#if DEBUG
 17#pragma warning disable IDE0052 // Remove unread private members
 18        private readonly FramingConnection _connection;
 19#pragma warning restore IDE0052 // Remove unread private members
 20#endif
 21        private readonly PipeReader _input;
 22        private readonly PipeWriter _output;
 23        private bool _canRead;
 24        private readonly object _thisLock;
 25        private TaskCompletionSource<object> _unwrapTcs;
 026        private readonly SemaphoreSlim _readSemaphore = new SemaphoreSlim(1, 1);
 27
 028        public RawStream(FramingConnection connection)
 29        {
 30#if DEBUG
 31            _connection = connection;
 32#endif
 033            _input = connection.Input;
 034            _output = connection.Output;
 035            _canRead = true;
 036            _thisLock = new object();
 037        }
 38
 39        public override bool CanRead
 40        {
 41            get
 42            {
 043                lock (_thisLock)
 44                {
 045                    return _canRead;
 46                }
 047            }
 48        }
 49
 050        public override bool CanSeek => false;
 51
 052        public override bool CanWrite => true;
 53
 54        public override long Length
 55        {
 56            get
 57            {
 058                throw new NotSupportedException();
 59            }
 60        }
 61
 62        public override long Position
 63        {
 64            get
 65            {
 066                throw new NotSupportedException();
 67            }
 68            set
 69            {
 070                throw new NotSupportedException();
 71            }
 72        }
 73
 74        public override long Seek(long offset, SeekOrigin origin)
 75        {
 076            throw new NotSupportedException();
 77        }
 78
 79        public override void SetLength(long value)
 80        {
 081            throw new NotSupportedException();
 82        }
 83
 84        public override int Read(byte[] buffer, int offset, int count)
 85        {
 86            // ValueTask uses .GetAwaiter().GetResult() if necessary
 87            // https://github.com/dotnet/corefx/blob/f9da3b4af08214764a51b2331f3595ffaf162abe/src/System.Threading.Tasks
 088            return ReadAsyncInternal(new Memory<byte>(buffer, offset, count)).Result;
 89        }
 90
 91        public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
 92        {
 093            return ReadAsyncInternal(new Memory<byte>(buffer, offset, count)).AsTask();
 94        }
 95
 96        // TODO: Uncomment once moved to netstandard2.1+
 97        // public override ValueTask<int> ReadAsync(Memory<byte> destination, CancellationToken cancellationToken = defa
 98        // {
 99        //     return ReadAsyncInternal(destination);
 100        // }
 101
 102        public override void Write(byte[] buffer, int offset, int count)
 103        {
 0104            WriteAsync(buffer, offset, count).GetAwaiter().GetResult();
 0105        }
 106
 107        public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
 108        {
 0109            if (buffer != null)
 110            {
 0111                _output.Write(new ReadOnlySpan<byte>(buffer, offset, count));
 112            }
 113
 0114            await _output.FlushAsync(cancellationToken);
 0115        }
 116
 117        // TODO: Uncomment once moved to netstandard2.1+
 118        //public override async ValueTask WriteAsync(ReadOnlyMemory<byte> source, CancellationToken cancellationToken = 
 119        //{
 120        //    _output.Write(source.Span);
 121        //    await _output.FlushAsync(cancellationToken);
 122        //}
 123
 124        public override void Flush()
 125        {
 0126            FlushAsync(CancellationToken.None).GetAwaiter().GetResult();
 0127        }
 128
 129        public override Task FlushAsync(CancellationToken cancellationToken)
 130        {
 0131            return WriteAsync(null, 0, 0, cancellationToken);
 132        }
 133
 134        private async ValueTask<int> ReadAsyncInternal(Memory<byte> destination)
 135        {
 0136            await _readSemaphore.WaitAsync();
 137            try
 138            {
 139                while (true)
 140                {
 0141                    if (!CanRead)
 142                    {
 0143                        await _unwrapTcs.Task;
 0144                        return 0;
 145                    }
 146
 0147                    ReadResult result = await _input.ReadAsync();
 0148                    if (!CanRead)
 149                    {
 0150                        _input.AdvanceTo(result.Buffer.Start);
 0151                        await _unwrapTcs.Task;
 0152                        return 0;
 153                    }
 154
 0155                    ReadOnlySequence<byte> readableBuffer = result.Buffer;
 156                    try
 157                    {
 0158                        if (!readableBuffer.IsEmpty)
 159                        {
 160                            // buffer.Count is int
 0161                            int count = (int)Math.Min(readableBuffer.Length, destination.Length);
 0162                            readableBuffer = readableBuffer.Slice(0, count);
 0163                            readableBuffer.CopyTo(destination.Span);
 0164                            return count;
 165                        }
 166
 0167                        if (result.IsCompleted)
 168                        {
 169                            // Treat a closed PipeReader as end-of-stream instead of looping;
 170                            // otherwise an empty + completed result would cause this read to
 171                            // make no progress.
 0172                            return 0;
 173                        }
 0174                    }
 175                    finally
 176                    {
 0177                        _input.AdvanceTo(readableBuffer.End, readableBuffer.End);
 178                    }
 179                }
 180            }
 181            finally
 182            {
 183                Fx.Assert(_readSemaphore.CurrentCount == 0, "_readSemaphore double release");
 0184                _readSemaphore.Release();
 185            }
 0186        }
 187
 188        public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback callback, object stat
 189        {
 0190            Task<int> task = ReadAsync(buffer, offset, count, default, state);
 0191            if (callback != null)
 192            {
 0193                task.ContinueWith(t => callback.Invoke(t));
 194            }
 0195            return task;
 196        }
 197
 198        public override int EndRead(IAsyncResult asyncResult)
 199        {
 0200            return ((Task<int>)asyncResult).GetAwaiter().GetResult();
 201        }
 202
 203        private Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken, object st
 204        {
 0205            var tcs = new TaskCompletionSource<int>(state);
 0206            Task<int> task = ReadAsync(buffer, offset, count, cancellationToken);
 0207            task.ContinueWith((task2, state2) =>
 0208            {
 0209                var tcs2 = (TaskCompletionSource<int>)state2;
 0210                if (task2.IsCanceled)
 0211                {
 0212                    tcs2.SetCanceled();
 0213                }
 0214                else if (task2.IsFaulted)
 0215                {
 0216                    tcs2.SetException(task2.Exception);
 0217                }
 0218                else
 0219                {
 0220                    tcs2.SetResult(task2.Result);
 0221                }
 0222            }, tcs, cancellationToken);
 0223            return tcs.Task;
 224        }
 225
 226        public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback callback, object sta
 227        {
 0228            Task task = WriteAsync(buffer, offset, count, default, state);
 0229            if (callback != null)
 230            {
 0231                task.ContinueWith(t => callback.Invoke(t));
 232            }
 0233            return task;
 234        }
 235
 236        public override void EndWrite(IAsyncResult asyncResult)
 237        {
 0238            ((Task<object>)asyncResult).GetAwaiter().GetResult();
 0239        }
 240
 241        private Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken, object state)
 242        {
 0243            var tcs = new TaskCompletionSource<object>(state);
 0244            Task task = WriteAsync(buffer, offset, count, cancellationToken);
 0245            task.ContinueWith((task2, state2) =>
 0246            {
 0247                var tcs2 = (TaskCompletionSource<object>)state2;
 0248                if (task2.IsCanceled)
 0249                {
 0250                    tcs2.SetCanceled();
 0251                }
 0252                else if (task2.IsFaulted)
 0253                {
 0254                    tcs2.SetException(task2.Exception);
 0255                }
 0256                else
 0257                {
 0258                    tcs2.SetResult(null);
 0259                }
 0260            }, tcs, cancellationToken);
 0261            return tcs.Task;
 262        }
 263
 264        public void StartUnwrapRead()
 265        {
 266            // Upon sending the session end byte, the client can start another session immediately.
 267            // We need to stop those bytes from being consumed by the upgrade stream (e.g. NegotiateStream),
 268            // but we can't return a zero byte response to the pending read until after we've sent the session
 269            // end byte otherwise it will close the upgrade stream and prevent the session end byte from being
 270            // sent. Calling StartUnwrapRead prevents any reads from completing until FinisheUnwrapRead has
 271            // been called. This ensures any client bytes from the next session are not consumed and still
 272            // allows a write to be sent through the wrapping stream.
 273
 0274            bool acquired = _readSemaphore.Wait(0);
 275            try
 276            {
 0277                lock (_thisLock)
 278                {
 0279                    _unwrapTcs = new TaskCompletionSource<object>();
 0280                    _canRead = false;
 0281                }
 282
 0283                _input.CancelPendingRead();
 0284            }
 285            finally
 286            {
 0287                if (acquired)
 288                {
 0289                    _readSemaphore.Release();
 290                }
 0291            }
 0292        }
 293
 294        public async Task FinishUnwrapReadAsync()
 295        {
 296            Fx.Assert(_unwrapTcs != null, "StartUnwrapRead must be called first");
 0297            _unwrapTcs.TrySetResult(null);
 298            // Ensure any reads have completed before continuing on to connection reuse
 0299            await _readSemaphore.WaitAsync();
 0300            _readSemaphore.Release();
 0301        }
 302    }
 303}
Generated by: ReportGenerator 5.5.11.0
08/09/2026 - 10:15:48
GitHub | reportgenerator.io

Methods/Properties

.ctor(CoreWCF.Channels.Framing.FramingConnection)
CanRead()
CanSeek()
CanWrite()
Length()
Position()
Position(System.Int64)
Seek(System.Int64,System.IO.SeekOrigin)
SetLength(System.Int64)
Read(System.Byte[],System.Int32,System.Int32)
ReadAsync(System.Byte[],System.Int32,System.Int32,System.Threading.CancellationToken)
Write(System.Byte[],System.Int32,System.Int32)
WriteAsync()
Flush()
FlushAsync(System.Threading.CancellationToken)
ReadAsyncInternal()
BeginRead(System.Byte[],System.Int32,System.Int32,System.AsyncCallback,System.Object)
EndRead(System.IAsyncResult)
ReadAsync(System.Byte[],System.Int32,System.Int32,System.Threading.CancellationToken,System.Object)
BeginWrite(System.Byte[],System.Int32,System.Int32,System.AsyncCallback,System.Object)
EndWrite(System.IAsyncResult)
WriteAsync(System.Byte[],System.Int32,System.Int32,System.Threading.CancellationToken,System.Object)
StartUnwrapRead()
FinishUnwrapReadAsync()