< Summary - CoreWCF Coverage — PR #1766

Information
Class: CoreWCF.Channels.TransportDuplexSessionChannel
Assembly: CoreWCF.NetFramingBase
File(s): /home/runner/work/CoreWCF/CoreWCF/src/Common/src/DuplexChannels/CoreWCF/Channels/TransportDuplexSessionChannel.cs
Line coverage
69%
Covered lines: 118
Uncovered lines: 52
Coverable lines: 170
Total lines: 514
Line coverage: 69.4%
Branch coverage
61%
Covered branches: 30
Total branches: 49
Branch coverage: 61.2%
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()66.66%6684.61%
TryReceiveAsync()100%11100%
SetChannelBinding(...)100%110%
CloseOutputSessionAsync()75%4477.27%
SetMessageSource(...)100%11100%
OnAbort()100%11100%
OnFaulted()100%11100%
OnCloseAsync()100%2283.33%
OnClosed()100%11100%
OnReceiveMessage(...)100%22100%
ApplyChannelBinding(...)100%11100%
PrepareMessage(...)100%11100%
OnSendAsync()50%2273.68%
ThrowIfOutputSessionClosed()50%2266.66%
EnsureInputClosedAsync()0%2216.66%
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%11100%

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)
 6431        : base(settings, remoteAddress, via, settings.ManualAddressing, settings.MessageVersion)
 32        {
 6433            LocalAddress = localAddress;
 6434            LocalVia = localVia;
 6435            BufferManager = settings.BufferManager;
 6436            MessageEncoder = settings.MessageEncoderFactory.CreateSessionEncoder();
 6437            Session = new ConnectionDuplexSession(this);
 6438            _inputSessionClosedTcs = new TaskCompletionSource<object>();
 6439        }
 40
 041        public EndpointAddress LocalAddress { get; }
 42
 043        public SecurityMessageProperty RemoteSecurity { get; protected set; }
 44
 13045        public IDuplexSession Session { get; protected set; }
 46
 34447        public SemaphoreSlim SendLock { get; } = new SemaphoreSlim(1);
 48
 49        protected ChannelBinding ChannelBinding
 50        {
 51            get
 52            {
 053                return _channelBindingToken;
 54            }
 55        }
 56
 15257        protected BufferManager BufferManager { get; }
 58
 7659        protected Uri LocalVia { get; }
 60
 14061        protected MessageEncoder MessageEncoder { get; set; }
 62
 20863        protected SynchronizedMessageSource MessageSource { get; private set; }
 64
 65        protected abstract bool IsStreamedOutput { get; }
 66
 67        public async Task StartReceivingAsync()
 68        {
 6469            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
 7375            while (true)
 76            {
 13777                (Message message, bool success) = await TryReceiveAsync(CancellationToken.None);
 13678                if (success)
 79                {
 13180                    await ChannelDispatcher.DispatchAsync(message);
 81                }
 82
 13683                if (message == null || this.DoneReceivingInCurrentState()) // NULL message means client sent FIN byte
 84                {
 6385                    return;
 86                }
 7387            }
 6388        }
 89
 90        public async Task<Message> ReceiveAsync(CancellationToken token)
 91        {
 13992            Message message = null;
 13993            if (this.DoneReceivingInCurrentState())
 94            {
 095                return null;
 96            }
 97
 13998            bool shouldFault = true;
 99            try
 100            {
 139101                message = await MessageSource.ReceiveAsync(token);
 133102                OnReceiveMessage(message);
 133103                shouldFault = false;
 133104                return message;
 105            }
 106            finally
 107            {
 139108                if (shouldFault)
 109                {
 6110                    if (message != null)
 111                    {
 0112                        message.Close();
 113                    }
 114
 6115                    Fault();
 116                }
 117            }
 133118        }
 119
 120        public async Task<(Message message, bool success)> TryReceiveAsync(CancellationToken token)
 121        {
 122            try
 123            {
 139124                return (await ReceiveAsync(token), true);
 125            }
 126            catch (TimeoutException e)
 127            {
 128                //if (TD.ReceiveTimeoutIsEnabled())
 129                //{
 130                //    TD.ReceiveTimeout(e.Message);
 131                //}
 132
 5133                DiagnosticUtility.TraceHandledException(e, TraceEventType.Information);
 134
 5135                return (null, false);
 136            }
 138137        }
 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        {
 64147            ThrowIfNotOpened();
 64148            ThrowIfFaulted();
 149            try
 150            {
 64151                await SendLock.WaitAsync(token);
 64152            }
 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
 64164                ThrowIfFaulted();
 165
 166                // we're synchronized by sendLock here
 64167                if (_isOutputSessionClosed)
 168                {
 2169                    return;
 170                }
 171
 62172                _isOutputSessionClosed = true;
 62173                bool shouldFault = true;
 174                try
 175                {
 62176                    await CloseOutputSessionCoreAsync(token);
 62177                    OnOutputSessionClosed(token);
 62178                    shouldFault = false;
 62179                }
 180                finally
 181                {
 62182                    if (shouldFault)
 183                    {
 0184                        Fault();
 185                    }
 186                }
 62187            }
 188            finally
 189            {
 64190                SendLock.Release();
 191            }
 64192        }
 193
 194        protected void SetMessageSource(IMessageSource messageSource)
 195        {
 64196            MessageSource = new SynchronizedMessageSource(messageSource);
 64197        }
 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        {
 6206            ReturnConnectionIfNecessary(true, CancellationToken.None);
 6207        }
 208
 209        protected override void OnFaulted()
 210        {
 1211            base.OnFaulted();
 1212            ReturnConnectionIfNecessary(true, CancellationToken.None);
 1213        }
 214
 215        protected override async Task OnCloseAsync(CancellationToken token)
 216        {
 62217            await CloseOutputSessionAsync(token);
 218
 219            // close input session if necessary
 62220            if (!_isInputSessionClosed)
 221            {
 5222                await EnsureInputClosedAsync(token);
 0223                OnInputSessionClosed();
 224            }
 225
 57226            await CompleteCloseAsync(token);
 57227        }
 228
 229        protected override void OnClosed()
 230        {
 63231            base.OnClosed();
 232
 233            // clean up the CBT after transitioning to the closed state
 63234            ChannelBindingUtility.Dispose(ref _channelBindingToken);
 63235        }
 236
 237        protected virtual void OnReceiveMessage(Message message)
 238        {
 133239            if (message == null)
 240            {
 57241                OnInputSessionClosed();
 242            }
 243            else
 244            {
 76245                PrepareMessage(message);
 246            }
 76247        }
 248
 249        protected void ApplyChannelBinding(Message message)
 250        {
 152251            ChannelBindingUtility.TryAddToMessage(_channelBindingToken, message, false);
 152252        }
 253
 254        protected virtual void PrepareMessage(Message message)
 255        {
 76256            message.Properties.Via = LocalVia;
 257
 76258            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            //}
 76290        }
 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        {
 76300            ThrowIfDisposedOrNotOpen();
 301
 302            try
 303            {
 76304                await SendLock.WaitAsync(token);
 76305            }
 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
 76317                ThrowIfDisposedOrNotOpen();
 76318                ThrowIfOutputSessionClosed();
 319
 76320                bool success = false;
 321                try
 322                {
 76323                    ApplyChannelBinding(message);
 324
 76325                    await OnSendCoreAsync(message, token);
 76326                    success = true;
 76327                }
 328                finally
 329                {
 76330                    if (!success)
 331                    {
 0332                        Fault();
 333                    }
 334                }
 76335            }
 336            finally
 337            {
 76338                SendLock.Release();
 339            }
 76340        }
 341
 342        // cleanup after the framing handshake has completed
 343        protected abstract Task CompleteCloseAsync(CancellationToken token);
 344
 345        private void ThrowIfOutputSessionClosed()
 346        {
 76347            if (_isOutputSessionClosed)
 348            {
 0349                throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new InvalidOperationException(SR.SendCannotBeC
 350            }
 76351        }
 352
 353        private async Task EnsureInputClosedAsync(CancellationToken token)
 354        {
 5355            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        {
 57368            lock (ThisLock)
 369            {
 57370                if (_isInputSessionClosed)
 371                {
 0372                    return;
 373                }
 374
 57375                _isInputSessionClosed = true;
 57376                _inputSessionClosedTcs.TrySetResult(null);
 57377            }
 57378        }
 379
 380        private void OnOutputSessionClosed(CancellationToken token)
 381        {
 62382            bool releaseConnection = false;
 62383            lock (ThisLock)
 384            {
 62385                if (_isInputSessionClosed)
 386                {
 387                    // we're all done, release the connection
 55388                    releaseConnection = true;
 389                }
 62390            }
 391
 62392            if (releaseConnection)
 393            {
 55394                ReturnConnectionIfNecessary(false, token);
 395            }
 62396        }
 397
 398        internal void ThrowIfFaulted()
 399        {
 128400            ThrowPending();
 401
 128402            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            }
 128425        }
 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)
 128463                : base()
 464            {
 128465                Channel = channel;
 128466            }
 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
 2487            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            {
 2510                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)