< Summary - CoreWCF Coverage — PR #1766

Information
Class: CoreWCF.Channels.TransportDuplexSessionChannel
Assembly: CoreWCF.Http
File(s): /home/runner/work/CoreWCF/CoreWCF/src/Common/src/DuplexChannels/CoreWCF/Channels/TransportDuplexSessionChannel.cs
Line coverage
62%
Covered lines: 106
Uncovered lines: 64
Coverable lines: 170
Total lines: 514
Line coverage: 62.3%
Branch coverage
53%
Covered branches: 26
Total branches: 49
Branch coverage: 53%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Cyclomatic complexity NPath complexity Sequence coverage
.ctor(...)100%11100%
StartReceivingAsync()87.5%8890%
ReceiveAsync()33.33%6669.23%
TryReceiveAsync()100%1150%
SetChannelBinding(...)100%110%
CloseOutputSessionAsync()50%4472.72%
SetMessageSource(...)100%11100%
OnAbort()100%110%
OnFaulted()100%110%
OnCloseAsync()50%2266.66%
OnClosed()100%11100%
OnReceiveMessage(...)100%22100%
ApplyChannelBinding(...)100%11100%
PrepareMessage(...)100%11100%
OnSendAsync()50%2273.68%
ThrowIfOutputSessionClosed()50%2266.66%
EnsureInputClosedAsync()0%220%
OnInputSessionClosed()50%2285.71%
OnOutputSessionClosed(...)100%44100%
ThrowIfFaulted()71.42%7760%
CreateFaultedException()100%110%
ReceiveShutdownReturnedNonNull(...)0%220%
.ctor(...)100%11100%
CloseOutputSessionAsync()100%110%
CloseOutputSessionAsync(...)100%110%

File(s)

/home/runner/work/CoreWCF/CoreWCF/src/Common/src/DuplexChannels/CoreWCF/Channels/TransportDuplexSessionChannel.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.Globalization;
 7using System.Security.Authentication.ExtendedProtection;
 8using System.Threading;
 9using System.Threading.Tasks;
 10using CoreWCF.Configuration;
 11using CoreWCF.Diagnostics;
 12using CoreWCF.Dispatcher;
 13using CoreWCF.Runtime;
 14using CoreWCF.Security;
 15
 16namespace CoreWCF.Channels
 17{
 18    internal abstract class TransportDuplexSessionChannel : TransportOutputChannel, IDuplexSessionChannel
 19    {
 20        private bool _isInputSessionClosed;
 21        private bool _isOutputSessionClosed;
 22        private ChannelBinding _channelBindingToken;
 23        private TaskCompletionSource<object> _inputSessionClosedTcs;
 24
 25        protected TransportDuplexSessionChannel(
 26          ITransportFactorySettings settings,
 27          EndpointAddress localAddress,
 28          Uri localVia,
 29          EndpointAddress remoteAddress,
 30          Uri via)
 1131        : base(settings, remoteAddress, via, settings.ManualAddressing, settings.MessageVersion)
 32        {
 1133            LocalAddress = localAddress;
 1134            LocalVia = localVia;
 1135            BufferManager = settings.BufferManager;
 1136            MessageEncoder = settings.MessageEncoderFactory.CreateSessionEncoder();
 1137            Session = new ConnectionDuplexSession(this);
 1138            _inputSessionClosedTcs = new TaskCompletionSource<object>();
 1139        }
 40
 1241        public EndpointAddress LocalAddress { get; }
 42
 1143        public SecurityMessageProperty RemoteSecurity { get; protected set; }
 44
 1145        public IDuplexSession Session { get; protected set; }
 46
 6347        public SemaphoreSlim SendLock { get; } = new SemaphoreSlim(1);
 48
 49        protected ChannelBinding ChannelBinding
 50        {
 51            get
 52            {
 053                return _channelBindingToken;
 54            }
 55        }
 56
 2957        protected BufferManager BufferManager { get; }
 58
 1559        protected Uri LocalVia { get; }
 60
 3761        protected MessageEncoder MessageEncoder { get; set; }
 62
 3763        protected SynchronizedMessageSource MessageSource { get; private set; }
 64
 65        protected abstract bool IsStreamedOutput { get; }
 66
 67        public async Task StartReceivingAsync()
 68        {
 1169            if (ChannelDispatcher == null)
 70            {
 71                // TODO: Cleanup exception message, find a SR to use and add Fx error handling
 072                throw new InvalidOperationException("ChannelDispatcher isn't set");
 73            }
 74
 1575            while (true)
 76            {
 2677                (Message message, bool success) = await TryReceiveAsync(CancellationToken.None);
 2678                if (success)
 79                {
 2680                    await ChannelDispatcher.DispatchAsync(message);
 81                }
 82
 2683                if (message == null || this.DoneReceivingInCurrentState()) // NULL message means client sent FIN byte
 84                {
 1185                    return;
 86                }
 1587            }
 1188        }
 89
 90        public async Task<Message> ReceiveAsync(CancellationToken token)
 91        {
 2692            Message message = null;
 2693            if (this.DoneReceivingInCurrentState())
 94            {
 095                return null;
 96            }
 97
 2698            bool shouldFault = true;
 99            try
 100            {
 26101                message = await MessageSource.ReceiveAsync(token);
 26102                OnReceiveMessage(message);
 26103                shouldFault = false;
 26104                return message;
 105            }
 106            finally
 107            {
 26108                if (shouldFault)
 109                {
 0110                    if (message != null)
 111                    {
 0112                        message.Close();
 113                    }
 114
 0115                    Fault();
 116                }
 117            }
 26118        }
 119
 120        public async Task<(Message message, bool success)> TryReceiveAsync(CancellationToken token)
 121        {
 122            try
 123            {
 26124                return (await ReceiveAsync(token), true);
 125            }
 126            catch (TimeoutException e)
 127            {
 128                //if (TD.ReceiveTimeoutIsEnabled())
 129                //{
 130                //    TD.ReceiveTimeout(e.Message);
 131                //}
 132
 0133                DiagnosticUtility.TraceHandledException(e, TraceEventType.Information);
 134
 0135                return (null, false);
 136            }
 26137        }
 138
 139        protected void SetChannelBinding(ChannelBinding channelBinding)
 140        {
 141            Fx.Assert(_channelBindingToken == null, "ChannelBinding token can only be set once.");
 0142            _channelBindingToken = channelBinding;
 0143        }
 144
 145        protected async Task CloseOutputSessionAsync(CancellationToken token)
 146        {
 11147            ThrowIfNotOpened();
 11148            ThrowIfFaulted();
 149            try
 150            {
 11151                await SendLock.WaitAsync(token);
 11152            }
 0153            catch (OperationCanceledException)
 154            {
 155                // TODO: Fix the timeout value reported
 0156                throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new TimeoutException(
 0157                                                SR.Format(SR.CloseTimedOut, TimeSpan.Zero),
 0158                                                TimeoutHelper.CreateEnterTimedOutException(TimeSpan.Zero)));
 159            }
 160
 161            try
 162            {
 163                // check again in case the previous send faulted while we were waiting for the lock
 11164                ThrowIfFaulted();
 165
 166                // we're synchronized by sendLock here
 11167                if (_isOutputSessionClosed)
 168                {
 0169                    return;
 170                }
 171
 11172                _isOutputSessionClosed = true;
 11173                bool shouldFault = true;
 174                try
 175                {
 11176                    await CloseOutputSessionCoreAsync(token);
 11177                    OnOutputSessionClosed(token);
 11178                    shouldFault = false;
 11179                }
 180                finally
 181                {
 11182                    if (shouldFault)
 183                    {
 0184                        Fault();
 185                    }
 186                }
 11187            }
 188            finally
 189            {
 11190                SendLock.Release();
 191            }
 11192        }
 193
 194        protected void SetMessageSource(IMessageSource messageSource)
 195        {
 11196            MessageSource = new SynchronizedMessageSource(messageSource);
 11197        }
 198
 199        protected abstract Task CloseOutputSessionCoreAsync(CancellationToken token);
 200
 201        // used to return cached connection to the pool/reader pool
 202        protected abstract void ReturnConnectionIfNecessary(bool abort, CancellationToken token);
 203
 204        protected override void OnAbort()
 205        {
 0206            ReturnConnectionIfNecessary(true, CancellationToken.None);
 0207        }
 208
 209        protected override void OnFaulted()
 210        {
 0211            base.OnFaulted();
 0212            ReturnConnectionIfNecessary(true, CancellationToken.None);
 0213        }
 214
 215        protected override async Task OnCloseAsync(CancellationToken token)
 216        {
 11217            await CloseOutputSessionAsync(token);
 218
 219            // close input session if necessary
 11220            if (!_isInputSessionClosed)
 221            {
 0222                await EnsureInputClosedAsync(token);
 0223                OnInputSessionClosed();
 224            }
 225
 11226            await CompleteCloseAsync(token);
 11227        }
 228
 229        protected override void OnClosed()
 230        {
 11231            base.OnClosed();
 232
 233            // clean up the CBT after transitioning to the closed state
 11234            ChannelBindingUtility.Dispose(ref _channelBindingToken);
 11235        }
 236
 237        protected virtual void OnReceiveMessage(Message message)
 238        {
 26239            if (message == null)
 240            {
 11241                OnInputSessionClosed();
 242            }
 243            else
 244            {
 15245                PrepareMessage(message);
 246            }
 15247        }
 248
 249        protected void ApplyChannelBinding(Message message)
 250        {
 30251            ChannelBindingUtility.TryAddToMessage(_channelBindingToken, message, false);
 30252        }
 253
 254        protected virtual void PrepareMessage(Message message)
 255        {
 15256            message.Properties.Via = LocalVia;
 257
 15258            ApplyChannelBinding(message);
 259
 260            //if (FxTrace.Trace.IsEnd2EndActivityTracingEnabled)
 261            //{
 262            //    EventTraceActivity eventTraceActivity = EventTraceActivityHelper.TryExtractActivity(message);
 263            //    Guid relatedActivityId = EventTraceActivity.GetActivityIdFromThread();
 264            //    if (eventTraceActivity == null)
 265            //    {
 266            //        eventTraceActivity = EventTraceActivity.GetFromThreadOrCreate();
 267            //        EventTraceActivityHelper.TryAttachActivity(message, eventTraceActivity);
 268            //    }
 269
 270            //    if (TD.MessageReceivedByTransportIsEnabled())
 271            //    {
 272            //        TD.MessageReceivedByTransport(
 273            //            eventTraceActivity,
 274            //            this.LocalAddress != null && this.LocalAddress.Uri != null ? this.LocalAddress.Uri.AbsoluteUri
 275            //            relatedActivityId);
 276            //    }
 277            //}
 278
 279            //if (DiagnosticUtility.ShouldTraceInformation)
 280            //{
 281            //    TraceUtility.TraceEvent(
 282            //                 TraceEventType.Information,
 283            //                 TraceCode.MessageReceived,
 284            //                 SR.GetString(SR.TraceCodeMessageReceived),
 285            //                 MessageTransmitTraceRecord.CreateReceiveTraceRecord(message, this.LocalAddress),
 286            //                 this,
 287            //                 null,
 288            //                 message);
 289            //}
 15290        }
 291
 292        protected abstract Task CloseOutputAsync(CancellationToken token);
 293
 294        protected abstract ArraySegment<byte> EncodeMessage(Message message);
 295
 296        protected abstract Task OnSendCoreAsync(Message message, CancellationToken token);
 297
 298        protected override async Task OnSendAsync(Message message, CancellationToken token)
 299        {
 15300            ThrowIfDisposedOrNotOpen();
 301
 302            try
 303            {
 15304                await SendLock.WaitAsync(token);
 15305            }
 0306            catch (OperationCanceledException)
 307            {
 308                // TODO: Fix the timeout value reported
 0309                throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new TimeoutException(
 0310                                                SR.Format(SR.SendToViaTimedOut, Via, TimeSpan.Zero),
 0311                                                TimeoutHelper.CreateEnterTimedOutException(TimeSpan.Zero)));
 312            }
 313
 314            try
 315            {
 316                // check again in case the previous send faulted while we were waiting for the lock
 15317                ThrowIfDisposedOrNotOpen();
 15318                ThrowIfOutputSessionClosed();
 319
 15320                bool success = false;
 321                try
 322                {
 15323                    ApplyChannelBinding(message);
 324
 15325                    await OnSendCoreAsync(message, token);
 15326                    success = true;
 15327                }
 328                finally
 329                {
 15330                    if (!success)
 331                    {
 0332                        Fault();
 333                    }
 334                }
 15335            }
 336            finally
 337            {
 15338                SendLock.Release();
 339            }
 15340        }
 341
 342        // cleanup after the framing handshake has completed
 343        protected abstract Task CompleteCloseAsync(CancellationToken token);
 344
 345        private void ThrowIfOutputSessionClosed()
 346        {
 15347            if (_isOutputSessionClosed)
 348            {
 0349                throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new InvalidOperationException(SR.SendCannotBeC
 350            }
 15351        }
 352
 353        private async Task EnsureInputClosedAsync(CancellationToken token)
 354        {
 0355            Message message = await MessageSource.ReceiveAsync(token);
 0356            if (message != null)
 357            {
 0358                using (message)
 359                {
 0360                    ProtocolException error = ReceiveShutdownReturnedNonNull(message);
 0361                    throw TraceUtility.ThrowHelperError(error, message);
 362                }
 363            }
 0364        }
 365
 366        private void OnInputSessionClosed()
 367        {
 11368            lock (ThisLock)
 369            {
 11370                if (_isInputSessionClosed)
 371                {
 0372                    return;
 373                }
 374
 11375                _isInputSessionClosed = true;
 11376                _inputSessionClosedTcs.TrySetResult(null);
 11377            }
 11378        }
 379
 380        private void OnOutputSessionClosed(CancellationToken token)
 381        {
 11382            bool releaseConnection = false;
 11383            lock (ThisLock)
 384            {
 11385                if (_isInputSessionClosed)
 386                {
 387                    // we're all done, release the connection
 11388                    releaseConnection = true;
 389                }
 11390            }
 391
 11392            if (releaseConnection)
 393            {
 11394                ReturnConnectionIfNecessary(false, token);
 395            }
 11396        }
 397
 398        internal void ThrowIfFaulted()
 399        {
 22400            ThrowPending();
 401
 22402            switch (State)
 403            {
 404                case CommunicationState.Created:
 405                    break;
 406
 407                case CommunicationState.Opening:
 408                    break;
 409
 410                case CommunicationState.Opened:
 411                    break;
 412
 413                case CommunicationState.Closing:
 414                    break;
 415
 416                case CommunicationState.Closed:
 417                    break;
 418
 419                case CommunicationState.Faulted:
 0420                    throw TraceUtility.ThrowHelperError(CreateFaultedException(), Guid.Empty, this);
 421
 422                default:
 0423                    throw Fx.AssertAndThrow("ThrowIfFaulted: Unknown CommunicationObject.state");
 424            }
 22425        }
 426
 427        internal Exception CreateFaultedException()
 428        {
 0429            string message = SR.Format(SR.CommunicationObjectFaulted1, GetCommunicationObjectType().ToString());
 0430            return new CommunicationObjectFaultedException(message);
 431        }
 432
 433        internal ProtocolException ReceiveShutdownReturnedNonNull(Message message)
 434        {
 0435            if (message.IsFault)
 436            {
 437                try
 438                {
 0439                    MessageFault fault = MessageFault.CreateFault(message, 64 * 1024);
 0440                    FaultReasonText reason = fault.Reason.GetMatchingTranslation(CultureInfo.CurrentCulture);
 0441                    string text = SR.Format(SR.ReceiveShutdownReturnedFault, reason.Text);
 0442                    return new ProtocolException(text);
 443                }
 0444                catch (QuotaExceededException)
 445                {
 0446                    string text = SR.Format(SR.ReceiveShutdownReturnedLargeFault, message.Headers.Action);
 0447                    return new ProtocolException(text);
 448                }
 449            }
 450            else
 451            {
 0452                string text = SR.Format(SR.ReceiveShutdownReturnedMessage, message.Headers.Action);
 0453                return new ProtocolException(text);
 454            }
 0455        }
 456
 457        internal class ConnectionDuplexSession : IDuplexSession
 458        {
 459            private static UriGenerator s_uriGenerator;
 460            private string _id;
 461
 462            public ConnectionDuplexSession(TransportDuplexSessionChannel channel)
 11463                : base()
 464            {
 11465                Channel = channel;
 11466            }
 467
 468            public string Id
 469            {
 470                get
 471                {
 0472                    if (_id == null)
 473                    {
 0474                        lock (Channel)
 475                        {
 0476                            if (_id == null)
 477                            {
 0478                                _id = UriGenerator.Next();
 479                            }
 0480                        }
 481                    }
 482
 0483                    return _id;
 484                }
 485            }
 486
 0487            public TransportDuplexSessionChannel Channel { get; }
 488
 489            private static UriGenerator UriGenerator
 490            {
 491                get
 492                {
 0493                    if (s_uriGenerator == null)
 494                    {
 0495                        s_uriGenerator = new UriGenerator();
 496                    }
 497
 0498                    return s_uriGenerator;
 499                }
 500            }
 501
 502            public Task CloseOutputSessionAsync()
 503            {
 0504                var timeoutHelper = new TimeoutHelper(Channel.DefaultCloseTimeout);
 0505                return CloseOutputSessionAsync(timeoutHelper.GetCancellationToken());
 506            }
 507
 508            public Task CloseOutputSessionAsync(CancellationToken token)
 509            {
 0510                return Channel.CloseOutputSessionAsync(token);
 511            }
 512        }
 513    }
 514}
Generated by: ReportGenerator 5.5.11.0
08/09/2026 - 10:15:48
GitHub | reportgenerator.io

Methods/Properties

.ctor(CoreWCF.Channels.ITransportFactorySettings,CoreWCF.EndpointAddress,System.Uri,CoreWCF.EndpointAddress,System.Uri)
LocalAddress()
RemoteSecurity()
Session()
SendLock()
ChannelBinding()
BufferManager()
LocalVia()
MessageEncoder()
MessageSource()
StartReceivingAsync()
ReceiveAsync()
TryReceiveAsync()
SetChannelBinding(System.Security.Authentication.ExtendedProtection.ChannelBinding)
CloseOutputSessionAsync()
SetMessageSource(CoreWCF.Channels.IMessageSource)
OnAbort()
OnFaulted()
OnCloseAsync()
OnClosed()
OnReceiveMessage(CoreWCF.Channels.Message)
ApplyChannelBinding(CoreWCF.Channels.Message)
PrepareMessage(CoreWCF.Channels.Message)
OnSendAsync()
ThrowIfOutputSessionClosed()
EnsureInputClosedAsync()
OnInputSessionClosed()
OnOutputSessionClosed(System.Threading.CancellationToken)
ThrowIfFaulted()
CreateFaultedException()
ReceiveShutdownReturnedNonNull(CoreWCF.Channels.Message)
.ctor(CoreWCF.Channels.TransportDuplexSessionChannel)
Id()
Channel()
UriGenerator()
CloseOutputSessionAsync()
CloseOutputSessionAsync(System.Threading.CancellationToken)