mono/mcs/class/referencesource/System.ServiceModel/System/ServiceModel/Channels/ReliableInputSessionChannel.cs

//------------------------------------------------------------
// Copyright (c) Microsoft Corporation.  All rights reserved.
//------------------------------------------------------------
namespace System.ServiceModel.Channels
{
    using System.Collections.Generic;
    using System.Runtime;
    using System.ServiceModel;
    using System.ServiceModel.Diagnostics;
    using System.Threading;
    using System.Xml;

    abstract class ReliableInputSessionChannel : InputChannel, IInputSessionChannel
    {
        bool advertisedZero = false;
        IServerReliableChannelBinder binder;
        ReliableInputConnection connection;
        DeliveryStrategy<Message> deliveryStrategy;
        ReliableChannelListenerBase<IInputSessionChannel> listener;
        ServerReliableSession session;
        protected string perfCounterId;
        static Action<object> asyncReceiveComplete = new Action<object>(AsyncReceiveCompleteStatic);
        static AsyncCallback onReceiveCompleted = Fx.ThunkCallback(new AsyncCallback(OnReceiveCompletedStatic));

        protected ReliableInputSessionChannel(
            ReliableChannelListenerBase<IInputSessionChannel> listener,
            IServerReliableChannelBinder binder,
            FaultHelper faultHelper,
            UniqueId inputID)
            : base(listener, binder.LocalAddress)
        {
            this.binder = binder;
            this.listener = listener;
            this.connection = new ReliableInputConnection();
            this.connection.ReliableMessagingVersion = listener.ReliableMessagingVersion;
            this.session = new ServerReliableSession(this, listener, binder, faultHelper, inputID, null);
            this.session.UnblockChannelCloseCallback = this.UnblockClose;

            if (listener.Ordered)
                this.deliveryStrategy = new OrderedDeliveryStrategy<Message>(this, listener.MaxTransferWindowSize, false);
            else
                this.deliveryStrategy = new UnorderedDeliveryStrategy<Message>(this, listener.MaxTransferWindowSize);

            this.binder.Faulted += OnBinderFaulted;
            this.binder.OnException += OnBinderException;
            this.session.Open(TimeSpan.Zero);

            if (PerformanceCounters.PerformanceCountersEnabled)
                this.perfCounterId = this.listener.Uri.ToString().ToUpperInvariant();
        }

        protected bool AdvertisedZero
        {
            get
            {
                return this.advertisedZero;
            }
            set
            {
                this.advertisedZero = value;
            }
        }

        public IServerReliableChannelBinder Binder
        {
            get
            {
                return this.binder;
            }
        }

        protected ReliableInputConnection Connection
        {
            get
            {
                return this.connection;
            }
        }

        protected DeliveryStrategy<Message> DeliveryStrategy
        {
            get
            {
                return this.deliveryStrategy;
            }
        }

        protected ReliableChannelListenerBase<IInputSessionChannel> Listener
        {
            get
            {
                return this.listener;
            }
        }

        protected ChannelReliableSession ReliableSession
        {
            get
            {
                return this.session;
            }
        }

        public IInputSession Session
        {
            get
            {
                return this.session;
            }
        }

        protected virtual void AggregateAsyncCloseOperations(List<OperationWithTimeoutBeginCallback> beginOperations, List<OperationEndCallback> endOperations)
        {
            beginOperations.Add(new OperationWithTimeoutBeginCallback(this.session.BeginClose));
            endOperations.Add(new OperationEndCallback(this.session.EndClose));
        }

        static void AsyncReceiveCompleteStatic(object state)
        {
            IAsyncResult result = (IAsyncResult)state;
            ReliableInputSessionChannel channel = (ReliableInputSessionChannel)(result.AsyncState);

            try
            {
                if (channel.HandleReceiveComplete(result))
                {
                    channel.StartReceiving(true);
                }
            }
#pragma warning suppress 56500 // covered by FxCOP
            catch (Exception e)
            {
                if (Fx.IsFatal(e))
                    throw;

                channel.ReliableSession.OnUnknownException(e);
            }
        }

        static void OnReceiveCompletedStatic(IAsyncResult result)
        {
            if (result.CompletedSynchronously)
                return;

            ReliableInputSessionChannel channel = (ReliableInputSessionChannel)(result.AsyncState);
            try
            {
                if (channel.HandleReceiveComplete(result))
                {
                    channel.StartReceiving(true);
                }
            }
#pragma warning suppress 56500 // covered by FxCOP
            catch (Exception e)
            {
                if (Fx.IsFatal(e))
                    throw;

                channel.ReliableSession.OnUnknownException(e);
            }
        }

        protected abstract bool HandleReceiveComplete(IAsyncResult result);

        protected virtual void AbortGuards()
        {
        }

        protected void AddAcknowledgementHeader(Message message)
        {
            int bufferRemaining = -1;

            if (this.Listener.FlowControlEnabled)
            {
                bufferRemaining = this.Listener.MaxTransferWindowSize - this.deliveryStrategy.EnqueuedCount;
                this.AdvertisedZero = (bufferRemaining == 0);
            }

            WsrmUtilities.AddAcknowledgementHeader(this.listener.ReliableMessagingVersion, message,
                this.session.InputID, this.connection.Ranges, this.connection.IsLastKnown, bufferRemaining);
        }

        IAsyncResult BeginCloseBinder(TimeSpan timeout, AsyncCallback callback, object state)
        {
            return this.binder.BeginClose(timeout, MaskingMode.Handled, callback, state);
        }

        protected virtual IAsyncResult BeginCloseGuards(TimeSpan timeout, AsyncCallback callback, object state)
        {
            return new CompletedAsyncResult(callback, state);
        }

        IAsyncResult BeginUnregisterChannel(TimeSpan timeout, AsyncCallback callback, object state)
        {
            return this.listener.OnReliableChannelBeginClose(this.ReliableSession.InputID, null,
                timeout, callback, state);
        }

        protected override void OnClosed()
        {
            base.OnClosed();
            this.binder.Faulted -= this.OnBinderFaulted;
            this.deliveryStrategy.Dispose();
        }

        protected virtual void CloseGuards(TimeSpan timeout)
        {
        }

        protected Message CreateAcknowledgmentMessage()
        {
            int bufferRemaining = -1;

            if (this.Listener.FlowControlEnabled)
            {
                bufferRemaining = this.Listener.MaxTransferWindowSize - this.deliveryStrategy.EnqueuedCount;
                this.AdvertisedZero = (bufferRemaining == 0);
            }

            Message message = WsrmUtilities.CreateAcknowledgmentMessage(
                this.listener.MessageVersion,
                this.listener.ReliableMessagingVersion,
                this.session.InputID,
                this.connection.Ranges,
                this.connection.IsLastKnown,
                bufferRemaining);

            return message;
        }

        void EndCloseBinder(IAsyncResult result)
        {
            this.binder.EndClose(result);
        }

        protected virtual void EndCloseGuards(IAsyncResult result)
        {
            CompletedAsyncResult.End(result);
        }

        void EndUnregisterChannel(IAsyncResult result)
        {
            this.listener.OnReliableChannelEndClose(result);
        }

        public override T GetProperty<T>()
        {
            if (typeof(T) == typeof(IInputSessionChannel))
            {
                return (T)(object)this;
            }

            T baseProperty = base.GetProperty<T>();
            if (baseProperty != null)
            {
                return baseProperty;
            }

            T innerProperty = this.binder.Channel.GetProperty<T>();
            if ((innerProperty == null) && (typeof(T) == typeof(FaultConverter)))
            {
                return (T)(object)FaultConverter.GetDefaultFaultConverter(this.listener.MessageVersion);
            }
            else
            {
                return innerProperty;
            }
        }

        protected override void OnAbort()
        {
            this.connection.Abort(this);
            this.AbortGuards();
            this.session.Abort();
            this.listener.OnReliableChannelAbort(this.ReliableSession.InputID, null);
            base.OnAbort();
        }

        protected override IAsyncResult OnBeginClose(TimeSpan timeout, AsyncCallback callback, object state)
        {
            this.ThrowIfCloseInvalid();

            OperationWithTimeoutBeginCallback[] beginOperations = new OperationWithTimeoutBeginCallback[] 
                { 
                    this.connection.BeginClose,
                    this.session.BeginClose, 
                    this.BeginCloseGuards, 
                    this.BeginCloseBinder,
                    this.BeginUnregisterChannel, 
                    base.OnBeginClose
                };

            OperationEndCallback[] endOperations = new OperationEndCallback[] 
                { 
                    this.connection.EndClose,
                    this.session.EndClose, 
                    this.EndCloseGuards, 
                    this.EndCloseBinder,
                    this.EndUnregisterChannel, 
                    base.OnEndClose 
                };

            return OperationWithTimeoutComposer.BeginComposeAsyncOperations(timeout,
                beginOperations, endOperations, callback, state);
        }

        void OnBinderException(IReliableChannelBinder sender, Exception exception)
        {
            if (exception is QuotaExceededException)
                this.session.OnLocalFault(exception, SequenceTerminatedFault.CreateQuotaExceededFault(this.session.OutputID), null);
            else
                this.EnqueueAndDispatch(exception, null, false);
        }

        void OnBinderFaulted(IReliableChannelBinder sender, Exception exception)
        {
            this.binder.Abort();
            exception = new CommunicationException(SR.GetString(SR.EarlySecurityFaulted), exception);
            this.session.OnLocalFault(exception, (Message)null, null);
        }

        protected override void OnClose(TimeSpan timeout)
        {
            this.ThrowIfCloseInvalid();

            TimeoutHelper timeoutHelper = new TimeoutHelper(timeout);

            this.connection.Close(timeoutHelper.RemainingTime());
            this.session.Close(timeoutHelper.RemainingTime());
            this.CloseGuards(timeoutHelper.RemainingTime());
            this.binder.Close(timeoutHelper.RemainingTime(), MaskingMode.Handled);
            this.listener.OnReliableChannelClose(this.ReliableSession.InputID, null,
                timeoutHelper.RemainingTime());
            base.OnClose(timeoutHelper.RemainingTime());
        }

        protected override void OnEndClose(IAsyncResult result)
        {
            OperationWithTimeoutComposer.EndComposeAsyncOperations(result);
        }

        protected override void OnFaulted()
        {
            this.session.OnFaulted();
            this.UnblockClose();
            base.OnFaulted();
            if (PerformanceCounters.PerformanceCountersEnabled)
                PerformanceCounters.SessionFaulted(this.perfCounterId);
        }

        protected virtual void OnQuotaAvailable()
        {
        }

        protected void ShutdownCallback(object state)
        {
            this.Shutdown();
        }

        protected void StartReceiving(bool canBlock)
        {
            while (true)
            {
                IAsyncResult result = this.Binder.BeginTryReceive(TimeSpan.MaxValue, onReceiveCompleted, this);
                if (!result.CompletedSynchronously)
                    return;

                if (!canBlock)
                {
                    ActionItem.Schedule(asyncReceiveComplete, result);
                    return;
                }

                if (!this.HandleReceiveComplete(result))
                    break;
            }
        }

        void ThrowIfCloseInvalid()
        {
            bool shouldFault = false;

            if (this.listener.ReliableMessagingVersion == ReliableMessagingVersion.WSReliableMessagingFebruary2005)
            {
                if (this.DeliveryStrategy.EnqueuedCount > 0 || this.Connection.Ranges.Count > 1)
                {
                    shouldFault = true;
                }
            }
            else if (this.listener.ReliableMessagingVersion == ReliableMessagingVersion.WSReliableMessaging11)
            {
                if (this.DeliveryStrategy.EnqueuedCount > 0)
                {
                    shouldFault = true;
                }
            }

            if (shouldFault)
            {
                WsrmFault fault = SequenceTerminatedFault.CreateProtocolFault(this.session.InputID,
                    SR.GetString(SR.SequenceTerminatedSessionClosedBeforeDone), SR.GetString(SR.SessionClosedBeforeDone));
                this.session.OnLocalFault(null, fault, null);
                throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(fault.CreateException());
            }
        }

        void UnblockClose()
        {
            this.connection.Fault(this);
        }
    }

    sealed class ReliableInputSessionChannelOverDuplex : ReliableInputSessionChannel
    {
        TimeSpan acknowledgementInterval;
        bool acknowledgementScheduled = false;
        IOThreadTimer acknowledgementTimer;
        Guard guard = new Guard(Int32.MaxValue);
        int pendingAcknowledgements = 0;

        public ReliableInputSessionChannelOverDuplex(
            ReliableChannelListenerBase<IInputSessionChannel> listener,
            IServerReliableChannelBinder binder, FaultHelper faultHelper,
            UniqueId inputID)
            : base(listener, binder, faultHelper, inputID)
        {
            this.acknowledgementInterval = listener.AcknowledgementInterval;
            this.acknowledgementTimer = new IOThreadTimer(new Action<object>(this.OnAcknowledgementTimeoutElapsed), null, true);
            this.DeliveryStrategy.DequeueCallback = this.OnDeliveryStrategyItemDequeued;

            if (binder.HasSession)
            {
                try
                {
                    this.StartReceiving(false);
                }
#pragma warning suppress 56500 // covered by FxCOP
                catch (Exception e)
                {
                    if (Fx.IsFatal(e))
                    {
                        throw;
                    }

                    this.ReliableSession.OnUnknownException(e);
                }
            }
        }

        protected override void AbortGuards()
        {
            this.guard.Abort();
        }

        protected override IAsyncResult BeginCloseGuards(TimeSpan timeout, AsyncCallback callback, object state)
        {
            return this.guard.BeginClose(timeout, callback, state);
        }

        protected override void CloseGuards(TimeSpan timeout)
        {
            this.guard.Close(timeout);
        }

        protected override void EndCloseGuards(IAsyncResult result)
        {
            this.guard.EndClose(result);
        }

        protected override bool HandleReceiveComplete(IAsyncResult result)
        {
            RequestContext context;
            if (this.Binder.EndTryReceive(result, out context))
            {
                if (context == null)
                {
                    bool terminated = false;

                    lock (this.ThisLock)
                    {
                        terminated = this.Connection.Terminate();
                    }

                    if (!terminated && (this.Binder.State == CommunicationState.Opened))
                    {
                        Exception e = new CommunicationException(SR.GetString(SR.EarlySecurityClose));
                        this.ReliableSession.OnLocalFault(e, (Message)null, null);
                    }
                    return false;
                }

                Message message = context.RequestMessage;
                context.Close();

                WsrmMessageInfo info = WsrmMessageInfo.Get(this.Listener.MessageVersion,
                    this.Listener.ReliableMessagingVersion, this.Binder.Channel, this.Binder.GetInnerSession(),
                    message);

                this.StartReceiving(false);
                this.ProcessMessage(info);
                return false;
            }
            return true;
        }

        void OnAcknowledgementTimeoutElapsed(object state)
        {
            lock (this.ThisLock)
            {
                this.acknowledgementScheduled = false;
                this.pendingAcknowledgements = 0;

                if (this.State == CommunicationState.Closing
                    || this.State == CommunicationState.Closed
                    || this.State == CommunicationState.Faulted)
                    return;
            }

            if (this.guard.Enter())
            {
                try
                {
                    using (Message message = CreateAcknowledgmentMessage())
                    {
                        this.Binder.Send(message, this.DefaultSendTimeout);
                    }
                }

                finally
                {
                    this.guard.Exit();
                }
            }
        }

        void OnDeliveryStrategyItemDequeued()
        {
            if (this.AdvertisedZero)
                this.OnAcknowledgementTimeoutElapsed(null);
        }

        protected override void OnClosing()
        {
            base.OnClosing();
            this.acknowledgementTimer.Cancel();
        }

        protected override void OnQuotaAvailable()
        {
            this.OnAcknowledgementTimeoutElapsed(null);
        }

        public void ProcessDemuxedMessage(WsrmMessageInfo info)
        {
            try
            {
                this.ProcessMessage(info);
            }
#pragma warning suppress 56500 // covered by FxCOP
            catch (Exception e)
            {
                if (Fx.IsFatal(e))
                    throw;

                this.ReliableSession.OnUnknownException(e);
            }
        }

        void ProcessMessage(WsrmMessageInfo info)
        {
            bool closeMessage = true;

            try
            {
                if (!this.ReliableSession.ProcessInfo(info, null))
                {
                    closeMessage = false;
                    return;
                }

                if (!this.ReliableSession.VerifySimplexProtocolElements(info, null))
                {
                    closeMessage = false;
                    return;
                }

                this.ReliableSession.OnRemoteActivity(false);

                if (info.CreateSequenceInfo != null)
                {
                    EndpointAddress acksTo;

                    if (WsrmUtilities.ValidateCreateSequence<IInputSessionChannel>(info, this.Listener, this.Binder.Channel, out acksTo))
                    {
                        Message response = WsrmUtilities.CreateCreateSequenceResponse(this.Listener.MessageVersion,
                            this.Listener.ReliableMessagingVersion, false, info.CreateSequenceInfo,
                            this.Listener.Ordered, this.ReliableSession.InputID, acksTo);

                        using (response)
                        {
                            if (this.Binder.AddressResponse(info.Message, response))
                                this.Binder.Send(response, this.DefaultSendTimeout);
                        }
                    }
                    else
                    {
                        this.ReliableSession.OnLocalFault(info.FaultException, info.FaultReply, null);
                    }

                    return;
                }

                bool needDispatch = false;
                bool scheduleShutdown = false;
                bool tryAckNow = (info.AckRequestedInfo != null);
                bool terminate = false;
                Message message = null;
                WsrmFault fault = null;
                Exception remoteFaultException = null;
                bool wsrmFeb2005 = this.Listener.ReliableMessagingVersion == ReliableMessagingVersion.WSReliableMessagingFebruary2005;
                bool wsrm11 = this.Listener.ReliableMessagingVersion == ReliableMessagingVersion.WSReliableMessaging11;

                if (info.SequencedMessageInfo != null)
                {
                    lock (this.ThisLock)
                    {
                        if (this.Aborted || this.State == CommunicationState.Faulted)
                        {
                            return;
                        }

                        Int64 sequenceNumber = info.SequencedMessageInfo.SequenceNumber;
                        bool isLast = wsrmFeb2005 && info.SequencedMessageInfo.LastMessage;

                        if (!this.Connection.IsValid(sequenceNumber, isLast))
                        {
                            if (wsrmFeb2005)
                            {
                                fault = new LastMessageNumberExceededFault(this.ReliableSession.InputID);
                            }
                            else
                            {
                                message = new SequenceClosedFault(this.ReliableSession.InputID).CreateMessage(
                                    this.Listener.MessageVersion, this.Listener.ReliableMessagingVersion);
                                tryAckNow = true;

                                if (PerformanceCounters.PerformanceCountersEnabled)
                                    PerformanceCounters.MessageDropped(this.perfCounterId);
                            }
                        }
                        else if (this.Connection.Ranges.Contains(sequenceNumber))
                        {
                            if (PerformanceCounters.PerformanceCountersEnabled)
                                PerformanceCounters.MessageDropped(this.perfCounterId);

                            tryAckNow = true;
                        }
                        else if (wsrmFeb2005 && info.Action == WsrmFeb2005Strings.LastMessageAction)
                        {
                            this.Connection.Merge(sequenceNumber, isLast);

                            if (this.Connection.AllAdded)
                            {
                                scheduleShutdown = true;
                                this.ReliableSession.CloseSession();
                            }
                        }
                        else if (this.State == CommunicationState.Closing)
                        {
                            if (wsrmFeb2005)
                            {
                                fault = SequenceTerminatedFault.CreateProtocolFault(this.ReliableSession.InputID,
                                    SR.GetString(SR.SequenceTerminatedSessionClosedBeforeDone),
                                    SR.GetString(SR.SessionClosedBeforeDone));
                            }
                            else
                            {
                                message = new SequenceClosedFault(this.ReliableSession.InputID).CreateMessage(
                                    this.Listener.MessageVersion, this.Listener.ReliableMessagingVersion);
                                tryAckNow = true;

                                if (PerformanceCounters.PerformanceCountersEnabled)
                                    PerformanceCounters.MessageDropped(this.perfCounterId);
                            }
                        }
                        // In the unordered case we accept no more than MaxSequenceRanges ranges to limit the
                        // serialized ack size and the amount of memory taken by the ack ranges. In the
                        // ordered case, the delivery strategy MaxTransferWindowSize quota mitigates this
                        // threat.
                        else if (this.DeliveryStrategy.CanEnqueue(sequenceNumber)
                            && (this.Listener.Ordered || this.Connection.CanMerge(sequenceNumber)))
                        {
                            this.Connection.Merge(sequenceNumber, isLast);
                            needDispatch = this.DeliveryStrategy.Enqueue(info.Message, sequenceNumber);
                            closeMessage = false;

                            this.pendingAcknowledgements++;
                            if (this.pendingAcknowledgements == this.Listener.MaxTransferWindowSize)
                                tryAckNow = true;

                            if (this.Connection.AllAdded)
                            {
                                scheduleShutdown = true;
                                this.ReliableSession.CloseSession();
                            }
                        }
                        else
                        {
                            if (PerformanceCounters.PerformanceCountersEnabled)
                                PerformanceCounters.MessageDropped(this.perfCounterId);
                        }

                        if (this.Connection.IsLastKnown)
                            tryAckNow = true;

                        if (!tryAckNow && this.pendingAcknowledgements > 0 && !this.acknowledgementScheduled && fault == null)
                        {
                            this.acknowledgementScheduled = true;
                            this.acknowledgementTimer.Set(this.acknowledgementInterval);
                        }
                    }
                }
                else if (wsrmFeb2005 && info.TerminateSequenceInfo != null)
                {
                    bool isTerminateEarly;

                    lock (this.ThisLock)
                    {
                        isTerminateEarly = !this.Connection.Terminate();
                    }

                    if (isTerminateEarly)
                    {
                        fault = SequenceTerminatedFault.CreateProtocolFault(this.ReliableSession.InputID,
                            SR.GetString(SR.SequenceTerminatedEarlyTerminateSequence),
                            SR.GetString(SR.EarlyTerminateSequence));
                    }
                }
                else if (wsrm11 && ((info.TerminateSequenceInfo != null) || info.CloseSequenceInfo != null))
                {
                    bool isTerminate = info.TerminateSequenceInfo != null;
                    WsrmRequestInfo requestInfo = isTerminate
                        ? (WsrmRequestInfo)info.TerminateSequenceInfo
                        : (WsrmRequestInfo)info.CloseSequenceInfo;
                    Int64 last = isTerminate ? info.TerminateSequenceInfo.LastMsgNumber : info.CloseSequenceInfo.LastMsgNumber;

                    if (!WsrmUtilities.ValidateWsrmRequest(this.ReliableSession, requestInfo, this.Binder, null))
                    {
                        return;
                    }

                    bool isLastLargeEnough = true;
                    bool isLastConsistent = true;

                    lock (this.ThisLock)
                    {
                        if (!this.Connection.IsLastKnown)
                        {
                            if (isTerminate)
                            {
                                if (this.Connection.SetTerminateSequenceLast(last, out isLastLargeEnough))
                                {
                                    scheduleShutdown = true;
                                }
                                else if (isLastLargeEnough)
                                {
                                    remoteFaultException = new ProtocolException(SR.GetString(SR.EarlyTerminateSequence));
                                }
                            }
                            else
                            {
                                scheduleShutdown = this.Connection.SetCloseSequenceLast(last);
                                isLastLargeEnough = scheduleShutdown;
                            }

                            if (scheduleShutdown)
                            {
                                this.ReliableSession.SetFinalAck(this.Connection.Ranges);
                                this.DeliveryStrategy.Dispose();
                            }
                        }
                        else
                        {
                            isLastConsistent = (last == this.Connection.Last);

                            // Have seen CloseSequence already, TerminateSequence means cleanup.
                            if (isTerminate && isLastConsistent && this.Connection.IsSequenceClosed)
                            {
                                terminate = true;
                            }
                        }
                    }

                    if (!isLastLargeEnough)
                    {
                        fault = SequenceTerminatedFault.CreateProtocolFault(this.ReliableSession.InputID,
                            SR.GetString(SR.SequenceTerminatedSmallLastMsgNumber),
                            SR.GetString(SR.SmallLastMsgNumberExceptionString));
                    }
                    else if (!isLastConsistent)
                    {
                        fault = SequenceTerminatedFault.CreateProtocolFault(this.ReliableSession.InputID,
                            SR.GetString(SR.SequenceTerminatedInconsistentLastMsgNumber),
                            SR.GetString(SR.InconsistentLastMsgNumberExceptionString));
                    }
                    else
                    {
                        message = isTerminate
                            ? WsrmUtilities.CreateTerminateResponseMessage(this.Listener.MessageVersion,
                            requestInfo.MessageId, this.ReliableSession.InputID)
                            : WsrmUtilities.CreateCloseSequenceResponse(this.Listener.MessageVersion,
                            requestInfo.MessageId, this.ReliableSession.InputID);

                        tryAckNow = true;
                    }
                }

                if (fault != null)
                {
                    this.ReliableSession.OnLocalFault(fault.CreateException(), fault, null);
                }
                else
                {
                    if (tryAckNow)
                    {
                        lock (this.ThisLock)
                        {
                            if (this.acknowledgementScheduled)
                            {
                                this.acknowledgementTimer.Cancel();
                                this.acknowledgementScheduled = false;
                            }

                            this.pendingAcknowledgements = 0;
                        }

                        if (message != null)
                        {
                            this.AddAcknowledgementHeader(message);
                        }
                        else
                        {
                            message = this.CreateAcknowledgmentMessage();
                        }
                    }

                    if (message != null)
                    {
                        using (message)
                        {
                            if (this.guard.Enter())
                            {
                                try
                                {
                                    this.Binder.Send(message, this.DefaultSendTimeout);
                                }
                                finally
                                {
                                    this.guard.Exit();
                                }
                            }
                        }
                    }

                    if (terminate)
                    {
                        lock (this.ThisLock)
                        {
                            this.Connection.Terminate();
                        }
                    }

                    if (remoteFaultException != null)
                    {
                        this.ReliableSession.OnRemoteFault(remoteFaultException);
                        return;
                    }

                    if (needDispatch)
                    {
                        this.Dispatch();
                    }

                    if (scheduleShutdown)
                    {
                        ActionItem.Schedule(this.ShutdownCallback, null);
                    }
                }
            }
            finally
            {
                if (closeMessage)
                {
                    info.Message.Close();
                }
            }
        }
    }

    sealed class ReliableInputSessionChannelOverReply : ReliableInputSessionChannel
    {
        public ReliableInputSessionChannelOverReply(
            ReliableChannelListenerBase<IInputSessionChannel> listener,
            IServerReliableChannelBinder binder, FaultHelper faultHelper,
            UniqueId inputID)
            : base(listener, binder, faultHelper, inputID)
        {
            if (binder.HasSession)
            {
                try
                {
                    this.StartReceiving(false);
                }
#pragma warning suppress 56500 // covered by FxCOP
                catch (Exception e)
                {
                    if (Fx.IsFatal(e))
                    {
                        throw;
                    }

                    this.ReliableSession.OnUnknownException(e);
                }
            }
        }

        protected override bool HandleReceiveComplete(IAsyncResult result)
        {
            RequestContext context;
            bool timeoutOkay = this.Binder.EndTryReceive(result, out context);

            if (timeoutOkay)
            {
                if (context == null)
                {
                    bool terminated = false;

                    lock (this.ThisLock)
                    {
                        terminated = this.Connection.Terminate();
                    }

                    if (!terminated && (this.Binder.State == CommunicationState.Opened))
                    {
                        Exception e = new CommunicationException(SR.GetString(SR.EarlySecurityClose));
                        this.ReliableSession.OnLocalFault(e, (Message)null, null);
                    }
                    return false;
                }

                WsrmMessageInfo info = WsrmMessageInfo.Get(this.Listener.MessageVersion,
                    this.Listener.ReliableMessagingVersion, this.Binder.Channel, this.Binder.GetInnerSession(),
                    context.RequestMessage);

                this.StartReceiving(false);
                this.ProcessRequest(context, info);
                return false;
            }
            return true;
        }

        public void ProcessDemuxedRequest(RequestContext context, WsrmMessageInfo info)
        {
            try
            {
                this.ProcessRequest(context, info);
            }
#pragma warning suppress 56500 // covered by FxCOP
            catch (Exception e)
            {
                if (Fx.IsFatal(e))
                    throw;

                this.ReliableSession.OnUnknownException(e);
            }
        }

        void ProcessRequest(RequestContext context, WsrmMessageInfo info)
        {
            bool closeContext = true;
            bool closeMessage = true;

            try
            {
                if (!this.ReliableSession.ProcessInfo(info, context))
                {
                    closeContext = false;
                    closeMessage = false;
                    return;
                }

                if (!this.ReliableSession.VerifySimplexProtocolElements(info, context))
                {
                    closeContext = false;
                    closeMessage = false;
                    return;
                }

                this.ReliableSession.OnRemoteActivity(false);

                if (info.CreateSequenceInfo != null)
                {
                    EndpointAddress acksTo;

                    if (WsrmUtilities.ValidateCreateSequence<IInputSessionChannel>(info, this.Listener, this.Binder.Channel, out acksTo))
                    {
                        Message response = WsrmUtilities.CreateCreateSequenceResponse(this.Listener.MessageVersion,
                            this.Listener.ReliableMessagingVersion, false, info.CreateSequenceInfo,
                            this.Listener.Ordered, this.ReliableSession.InputID, acksTo);

                        using (context)
                        {
                            using (response)
                            {
                                if (this.Binder.AddressResponse(info.Message, response))
                                    context.Reply(response, this.DefaultSendTimeout);
                            }
                        }
                    }
                    else
                    {
                        this.ReliableSession.OnLocalFault(info.FaultException, info.FaultReply, context);
                    }

                    closeContext = false;
                    closeMessage = false;
                    return;
                }

                bool needDispatch = false;
                bool scheduleShutdown = false;
                bool terminate = false;
                WsrmFault fault = null;
                Message message = null;
                Exception remoteFaultException = null;
                bool wsrmFeb2005 = this.Listener.ReliableMessagingVersion == ReliableMessagingVersion.WSReliableMessagingFebruary2005;
                bool wsrm11 = this.Listener.ReliableMessagingVersion == ReliableMessagingVersion.WSReliableMessaging11;
                bool addAck = info.AckRequestedInfo != null;

                if (info.SequencedMessageInfo != null)
                {
                    lock (this.ThisLock)
                    {
                        if (this.Aborted || (this.State == CommunicationState.Faulted))
                        {
                            return;
                        }
                        Int64 sequenceNumber = info.SequencedMessageInfo.SequenceNumber;
                        bool isLast = wsrmFeb2005 && info.SequencedMessageInfo.LastMessage;

                        if (!this.Connection.IsValid(sequenceNumber, isLast))
                        {
                            if (wsrmFeb2005)
                            {
                                fault = new LastMessageNumberExceededFault(this.ReliableSession.InputID);
                            }
                            else
                            {
                                message = new SequenceClosedFault(this.ReliableSession.InputID).CreateMessage(
                                    this.Listener.MessageVersion, this.Listener.ReliableMessagingVersion);

                                if (PerformanceCounters.PerformanceCountersEnabled)
                                    PerformanceCounters.MessageDropped(this.perfCounterId);
                            }
                        }
                        else if (this.Connection.Ranges.Contains(sequenceNumber))
                        {
                            if (PerformanceCounters.PerformanceCountersEnabled)
                                PerformanceCounters.MessageDropped(this.perfCounterId);
                        }
                        else if (wsrmFeb2005 && info.Action == WsrmFeb2005Strings.LastMessageAction)
                        {
                            this.Connection.Merge(sequenceNumber, isLast);
                            scheduleShutdown = this.Connection.AllAdded;
                        }
                        else if (this.State == CommunicationState.Closing)
                        {
                            if (wsrmFeb2005)
                            {
                                fault = SequenceTerminatedFault.CreateProtocolFault(this.ReliableSession.InputID,
                                    SR.GetString(SR.SequenceTerminatedSessionClosedBeforeDone),
                                    SR.GetString(SR.SessionClosedBeforeDone));
                            }
                            else
                            {
                                message = new SequenceClosedFault(this.ReliableSession.InputID).CreateMessage(
                                    this.Listener.MessageVersion, this.Listener.ReliableMessagingVersion);

                                if (PerformanceCounters.PerformanceCountersEnabled)
                                    PerformanceCounters.MessageDropped(this.perfCounterId);
                            }
                        }
                        // In the unordered case we accept no more than MaxSequenceRanges ranges to limit the
                        // serialized ack size and the amount of memory taken by the ack ranges. In the
                        // ordered case, the delivery strategy MaxTransferWindowSize quota mitigates this
                        // threat.
                        else if (this.DeliveryStrategy.CanEnqueue(sequenceNumber)
                            && (this.Listener.Ordered || this.Connection.CanMerge(sequenceNumber)))
                        {
                            this.Connection.Merge(sequenceNumber, isLast);
                            needDispatch = this.DeliveryStrategy.Enqueue(info.Message, sequenceNumber);
                            scheduleShutdown = this.Connection.AllAdded;
                            closeMessage = false;
                        }
                        else
                        {
                            if (PerformanceCounters.PerformanceCountersEnabled)
                                PerformanceCounters.MessageDropped(this.perfCounterId);
                        }
                    }
                }
                else if (wsrmFeb2005 && info.TerminateSequenceInfo != null)
                {
                    bool isTerminateEarly;

                    lock (this.ThisLock)
                    {
                        isTerminateEarly = !this.Connection.Terminate();
                    }

                    if (isTerminateEarly)
                    {
                        fault = SequenceTerminatedFault.CreateProtocolFault(this.ReliableSession.InputID,
                            SR.GetString(SR.SequenceTerminatedEarlyTerminateSequence),
                            SR.GetString(SR.EarlyTerminateSequence));
                    }
                    else
                    {
                        // In the normal case, TerminateSequence is a one-way operation, returning (the finally
                        // block will close the context).
                        return;
                    }
                }
                else if (wsrm11 && ((info.TerminateSequenceInfo != null) || (info.CloseSequenceInfo != null)))
                {
                    bool isTerminate = (info.TerminateSequenceInfo != null);
                    WsrmRequestInfo requestInfo = isTerminate
                        ? (WsrmRequestInfo)info.TerminateSequenceInfo
                        : (WsrmRequestInfo)info.CloseSequenceInfo;
                    Int64 last = isTerminate ? info.TerminateSequenceInfo.LastMsgNumber : info.CloseSequenceInfo.LastMsgNumber;

                    if (!WsrmUtilities.ValidateWsrmRequest(this.ReliableSession, requestInfo, this.Binder, context))
                    {
                        closeMessage = false;
                        closeContext = false;
                        return;
                    }

                    bool isLastLargeEnough = true;
                    bool isLastConsistent = true;

                    lock (this.ThisLock)
                    {
                        if (!this.Connection.IsLastKnown)
                        {
                            if (isTerminate)
                            {
                                if (this.Connection.SetTerminateSequenceLast(last, out isLastLargeEnough))
                                {
                                    scheduleShutdown = true;
                                }
                                else if (isLastLargeEnough)
                                {
                                    remoteFaultException = new ProtocolException(SR.GetString(SR.EarlyTerminateSequence));
                                }
                            }
                            else
                            {
                                scheduleShutdown = this.Connection.SetCloseSequenceLast(last);
                                isLastLargeEnough = scheduleShutdown;
                            }

                            if (scheduleShutdown)
                            {
                                this.ReliableSession.SetFinalAck(this.Connection.Ranges);
                                this.DeliveryStrategy.Dispose();
                            }
                        }
                        else
                        {
                            isLastConsistent = (last == this.Connection.Last);

                            // Have seen CloseSequence already, TerminateSequence means cleanup.
                            if (isTerminate && isLastConsistent && this.Connection.IsSequenceClosed)
                            {
                                terminate = true;
                            }
                        }
                    }

                    if (!isLastLargeEnough)
                    {
                        fault = SequenceTerminatedFault.CreateProtocolFault(this.ReliableSession.InputID,
                            SR.GetString(SR.SequenceTerminatedSmallLastMsgNumber),
                            SR.GetString(SR.SmallLastMsgNumberExceptionString));
                    }
                    else if (!isLastConsistent)
                    {
                        fault = SequenceTerminatedFault.CreateProtocolFault(this.ReliableSession.InputID,
                            SR.GetString(SR.SequenceTerminatedInconsistentLastMsgNumber),
                            SR.GetString(SR.InconsistentLastMsgNumberExceptionString));
                    }
                    else
                    {
                        message = isTerminate
                            ? WsrmUtilities.CreateTerminateResponseMessage(this.Listener.MessageVersion,
                            requestInfo.MessageId, this.ReliableSession.InputID)
                            : WsrmUtilities.CreateCloseSequenceResponse(this.Listener.MessageVersion,
                            requestInfo.MessageId, this.ReliableSession.InputID);
                        addAck = true;
                    }
                }

                if (fault != null)
                {
                    this.ReliableSession.OnLocalFault(fault.CreateException(), fault, context);
                    closeMessage = false;
                    closeContext = false;
                    return;
                }

                if (message != null && addAck)
                {
                    this.AddAcknowledgementHeader(message);
                }
                else if (message == null)
                {
                    message = this.CreateAcknowledgmentMessage();
                }

                using (message)
                {
                    context.Reply(message);
                }

                if (terminate)
                {
                    lock (this.ThisLock)
                    {
                        this.Connection.Terminate();
                    }
                }

                if (remoteFaultException != null)
                {
                    this.ReliableSession.OnRemoteFault(remoteFaultException);
                    return;
                }

                if (needDispatch)
                {
                    this.Dispatch();
                }

                if (scheduleShutdown)
                {
                    ActionItem.Schedule(this.ShutdownCallback, null);
                }
            }
            finally
            {
                if (closeMessage)
                    info.Message.Close();

                if (closeContext)
                    context.Close();
            }
        }
    }
}