< Summary - CoreWCF Coverage — PR #1766

Information
Class: CoreWCF.Channels.Framing.StreamingConnectionHelper
Assembly: CoreWCF.NetFramingBase
File(s): /home/runner/work/CoreWCF/CoreWCF/src/CoreWCF.NetFramingBase/src/CoreWCF/Channels/Framing/StreamedFramingRequestContext.cs
Line coverage
86%
Covered lines: 26
Uncovered lines: 4
Coverable lines: 30
Total lines: 258
Line coverage: 86.6%
Branch coverage
70%
Covered branches: 7
Total branches: 10
Branch coverage: 70%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Cyclomatic complexity NPath complexity Sequence coverage
WriteMessageAsync()70%101086.66%

File(s)

/home/runner/work/CoreWCF/CoreWCF/src/CoreWCF.NetFramingBase/src/CoreWCF/Channels/Framing/StreamedFramingRequestContext.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.IO;
 6using System.Threading;
 7using System.Threading.Tasks;
 8using CoreWCF.Runtime;
 9
 10namespace CoreWCF.Channels.Framing
 11{
 12    internal class StreamedFramingRequestContext : RequestContextBase
 13    {
 14        private readonly FramingConnection _connection;
 15        private readonly Message _requestMessage;
 16        private readonly Stream _inputStream;
 17        private bool _isClosed;
 18        private readonly TaskCompletionSource<object> _tcs;
 19
 20        public StreamedFramingRequestContext(FramingConnection connection, Message requestMessage, Stream inputStream)
 21            : base(requestMessage, connection.ServiceDispatcher.Binding.CloseTimeout, connection.ServiceDispatcher.Bindi
 22        {
 23            _connection = connection;
 24            _requestMessage = requestMessage;
 25            _inputStream = inputStream;
 26            _tcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
 27        }
 28
 29        protected override void OnAbort()
 30        {
 31            _tcs.TrySetResult(null);
 32            _connection.Abort();
 33        }
 34
 35        protected override async Task OnCloseAsync(CancellationToken token)
 36        {
 37            lock (ThisLock)
 38            {
 39                if (_isClosed)
 40                {
 41                    return;
 42                }
 43
 44                _isClosed = true;
 45            }
 46
 47            bool success = false;
 48            try
 49            {
 50                // first drain our stream if necessary
 51                if (_inputStream != null)
 52                {
 53                    byte[] dummy = Fx.AllocateByteArray(_connection.ConnectionBufferSize);
 54                    while (!_connection.EOF)
 55                    {
 56                        int bytesRead = await _inputStream.ReadAsync(dummy, 0, dummy.Length, token);
 57                        if (bytesRead == 0)
 58                        {
 59                            _connection.EOF = true;
 60                        }
 61                    }
 62                }
 63
 64                // send back EOF and then recycle the connection
 65                try
 66                {
 67                    await _connection.Output.WriteAsync(SingletonEncoder.EndBytes, token);
 68                    await _connection.Output.FlushAsync(token);
 69                }
 70                finally
 71                {
 72                    if (_connection.RawStream != null)
 73                    {
 74                        _connection.Logger.UnwrappingRawStream();
 75                        _connection.RawStream = null;
 76                        await _connection.Output.CompleteAsync();
 77                        await _connection.Input.CompleteAsync();
 78                    }
 79                }
 80
 81                // TODO: ChannelBinding
 82                //ChannelBindingUtility.Dispose(ref this.channelBindingToken);
 83
 84                success = true;
 85            }
 86            finally
 87            {
 88                _tcs.TrySetResult(null);
 89
 90                if (!success)
 91                {
 92                    Abort();
 93                }
 94            }
 95        }
 96
 97        protected override async Task OnReplyAsync(Message message, CancellationToken token)
 98        {
 99            if (_connection.MessageEncoderFactory.Encoder is ICompressedMessageEncoder compressedMessageEncoder && compr
 100            {
 101                compressedMessageEncoder.AddCompressedMessageProperties(message, _connection.FramingDecoder.ContentType)
 102            }
 103
 104            await StreamingConnectionHelper.WriteMessageAsync(message, _connection, false, _connection.ServiceDispatcher
 105        }
 106
 107        public Task ReplySent => _tcs.Task;
 108    }
 109
 110    internal static class StreamingConnectionHelper
 111    {
 112        public static async Task WriteMessageAsync(Message message, FramingConnection connection, bool isRequest,
 113            IDefaultCommunicationTimeouts settings, CancellationToken token)
 114        {
 50115            byte[] endBytes = null;
 50116            if (message != null)
 117            {
 50118                MessageEncoder messageEncoder = connection.MessageEncoderFactory.Encoder;
 50119                byte[] envelopeStartBytes = SingletonEncoder.EnvelopeStartBytes;
 120
 121                bool writeStreamed;
 50122                if (isRequest)
 123                {
 0124                    endBytes = SingletonEncoder.EnvelopeEndFramingEndBytes;
 0125                    writeStreamed = TransferModeHelper.IsRequestStreamed(connection.TransferMode);
 126                }
 127                else
 128                {
 50129                    endBytes = SingletonEncoder.EnvelopeEndBytes;
 50130                    writeStreamed = TransferModeHelper.IsResponseStreamed(connection.TransferMode);
 131                }
 132
 50133                if (writeStreamed)
 134                {
 43135                    await connection.Output.WriteAsync(envelopeStartBytes, token);
 43136                    Stream connectionStream = new StreamingOutputConnectionStream(connection, settings);
 137                    // TODO: Determine if timeout stream is needed as StreamingOutputConnectionStream implements some ti
 138                    //Stream writeTimeoutStream = new TimeoutStream(connectionStream, ref timeoutHelper);
 43139                    await messageEncoder.WriteMessageAsync(message, connectionStream);
 43140                    await connection.Output.FlushAsync();
 141                }
 142                else
 143                {
 7144                    ArraySegment<byte> messageData = messageEncoder.WriteMessage(message,
 7145                        int.MaxValue, connection.BufferManager, envelopeStartBytes.Length + IntEncoder.MaxEncodedSize);
 7146                    messageData = SingletonEncoder.EncodeMessageFrame(messageData);
 7147                    Buffer.BlockCopy(envelopeStartBytes, 0, messageData.Array, messageData.Offset - envelopeStartBytes.L
 7148                        envelopeStartBytes.Length);
 7149                    await connection.Output.WriteAsync(new ArraySegment<byte>(messageData.Array, messageData.Offset - en
 7150                        messageData.Count + envelopeStartBytes.Length), token);
 7151                    await connection.Output.FlushAsync();
 7152                    connection.BufferManager.ReturnBuffer(messageData.Array);
 153                }
 50154            }
 0155            else if (isRequest) // context handles response end bytes
 156            {
 0157                endBytes = SingletonEncoder.EndBytes;
 158            }
 159
 50160            if (endBytes != null)
 161            {
 50162                await connection.Output.WriteAsync(endBytes, token);
 50163                await connection.Output.FlushAsync();
 164            }
 50165        }
 166    }
 167
 168    // overrides Stream to add a Framing int at the beginning of each record
 169    internal class StreamingOutputConnectionStream : Stream
 170    {
 171        private readonly byte[] _encodedSize;
 172        private readonly FramingConnection _connection;
 173        private readonly IDefaultCommunicationTimeouts _timeouts;
 174
 175        public override bool CanRead => false;
 176
 177        public override bool CanSeek => false;
 178
 179        public override bool CanWrite => true;
 180
 181        public override long Length => throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new NotSupportedExcepti
 182
 183        public override long Position
 184        {
 185            get => throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new NotSupportedException(SR.SeekNotSupport
 186            set => throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new NotSupportedException(SR.SeekNotSupport
 187        }
 188
 189        public StreamingOutputConnectionStream(FramingConnection connection, IDefaultCommunicationTimeouts timeouts)
 190        {
 191            _encodedSize = new byte[IntEncoder.MaxEncodedSize];
 192            _connection = connection;
 193            _timeouts = timeouts;
 194        }
 195
 196        private async Task WriteChunkSizeAsync(int size, CancellationToken token)
 197        {
 198            if (size > 0)
 199            {
 200                int bytesEncoded = IntEncoder.Encode(size, _encodedSize, 0);
 201                await _connection.Output.WriteAsync(new ArraySegment<byte>(_encodedSize, 0, bytesEncoded), token);
 202            }
 203        }
 204
 205        public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback callback, object sta
 206        {
 207            return WriteAsync(buffer, offset, count, CancellationToken.None).ToApm(callback, state);
 208        }
 209
 210        public override void EndWrite(IAsyncResult asyncResult)
 211        {
 212            asyncResult.ToApmEnd();
 213        }
 214
 215        public override void WriteByte(byte value)
 216        {
 217            var timeoutHelper = new TimeoutHelper(_timeouts.SendTimeout);
 218            CancellationToken ct = timeoutHelper.GetCancellationToken();
 219            WriteChunkSizeAsync(1, ct).GetAwaiter().GetResult();
 220            _connection.Output.WriteAsync(new byte[] { value }, ct).GetAwaiter().GetResult();
 221            _connection.Output.FlushAsync();
 222        }
 223
 224        public override void Write(byte[] buffer, int offset, int count)
 225        {
 226            WriteAsync(buffer, offset, count, CancellationToken.None).GetAwaiter().GetResult();
 227        }
 228
 229        public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
 230        {
 231            var timeoutHelper = new TimeoutHelper(_timeouts.SendTimeout);
 232            CancellationToken ct = timeoutHelper.GetCancellationToken();
 233            await WriteChunkSizeAsync(count, ct);
 234            await _connection.Output.WriteAsync(new ArraySegment<byte>(buffer, offset, count), ct);
 235            await _connection.Output.FlushAsync();
 236        }
 237
 238        public override void Flush()
 239        {
 240            _connection.Output.FlushAsync().GetAwaiter().GetResult();
 241        }
 242
 243        public override int Read(byte[] buffer, int offset, int count)
 244        {
 245            throw new NotImplementedException();
 246        }
 247
 248        public override long Seek(long offset, SeekOrigin origin)
 249        {
 250            throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new NotSupportedException(SR.SeekNotSupported));
 251        }
 252
 253        public override void SetLength(long value)
 254        {
 255            throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new NotSupportedException(SR.SeekNotSupported));
 256        }
 257    }
 258}
Generated by: ReportGenerator 5.5.11.0
08/09/2026 - 10:15:48
GitHub | reportgenerator.io

Methods/Properties

WriteMessageAsync()