linux-packaging-mono/mcs/class/referencesource/System.ServiceModel.Channels/System/ServiceModel/Channels/UdpOutputChannel.cs

// <copyright>
// Copyright (c) Microsoft Corporation.  All rights reserved.
// </copyright> 

namespace System.ServiceModel.Channels
{
    using System;
    using System.Collections.Generic;
    using System.Diagnostics.CodeAnalysis;
    using System.Globalization;
    using System.Net;
    using System.Net.Sockets;
    using System.Runtime;
    using System.Runtime.Diagnostics;
    using System.ServiceModel.Diagnostics;
    using System.Threading;
    using System.Xml;

    internal abstract class UdpOutputChannel : OutputChannel, IOutputChannel
    {
        private bool cleanedUp;
        private volatile AsyncWaitHandle retransmissionDoneWaitHandle;
        private UdpRetransmissionSettings retransmitSettings;
        private volatile Dictionary<UniqueId, IUdpRetransmitter> retransmitList;
        private SynchronizedRandom randomNumberGenerator;
        private Uri via;

        public UdpOutputChannel(
            ChannelManagerBase factory,
            MessageEncoder encoder,
            BufferManager bufferManager,
            UdpSocket[] sendSockets,
            UdpRetransmissionSettings retransmissionSettings,
            Uri via,
            bool isMulticast)
            : base(factory)
        {
            Fx.Assert(encoder != null, "encoder shouldn't be null");
            Fx.Assert(bufferManager != null, "buffer manager shouldn't be null");
            Fx.Assert(sendSockets != null, "sendSockets can't be null");
            Fx.Assert(sendSockets.Length > 0, "sendSockets can't be empty");
            Fx.Assert(retransmissionSettings != null, "retransmissionSettings can't be null");
            Fx.Assert(via != null, "via can't be null");

            this.BufferManager = bufferManager;
            this.IsMulticast = isMulticast;
            this.Encoder = encoder;
            this.retransmitSettings = retransmissionSettings;
            this.SendSockets = sendSockets;
            this.via = via;

            if (this.retransmitSettings.Enabled)
            {
                this.retransmitList = new Dictionary<UniqueId, IUdpRetransmitter>();
                this.randomNumberGenerator = new SynchronizedRandom(AppDomain.CurrentDomain.GetHashCode() | Environment.TickCount);
            }
        }

        private interface IUdpRetransmitter
        {
            bool IsMulticast { get; }

            void CancelRetransmission();
        }

        public override EndpointAddress RemoteAddress
        {
            get { return null; }
        }

        public override Uri Via
        {
            get { return this.via; }
        }

        internal bool IsMulticast
        {
            get;
            private set;
        }

        internal TimeSpan InternalSendTimeout
        {
            get { return this.DefaultSendTimeout; }
        }

        protected BufferManager BufferManager
        {
            get;
            private set;
        }

        protected MessageEncoder Encoder
        {
            get;
            private set;
        }

        protected UdpSocket[] SendSockets
        {
            get;
            private set;
        }

        [SuppressMessage("Microsoft.StyleCop.CSharp.ReadabilityRules", "SA1100:DoNotPrefixCallsWithBaseUnlessLocalImplementationExists", Justification = "StyleCop 4.5 does not validate this rule properly.")]
        public override T GetProperty<T>()
        {
            if (typeof(T) == typeof(IOutputChannel))
            {
                return (T)(object)this;
            }

            T messageEncoderProperty = this.Encoder.GetProperty<T>();
            if (messageEncoderProperty != null)
            {
                return messageEncoderProperty;
            }

            return base.GetProperty<T>();
        }

        internal void CancelRetransmission(UniqueId messageId)
        {
            if (messageId != null && this.retransmitList != null)
            {
                lock (this.ThisLock)
                {
                    if (this.retransmitList != null)
                    {
                        IUdpRetransmitter retransmitter;
                        if (this.retransmitList.TryGetValue(messageId, out retransmitter))
                        {
                            this.retransmitList.Remove(messageId);
                            retransmitter.CancelRetransmission();
                        }
                    }
                }
            }
        }

        protected static void LogMessage(ref Message message, ArraySegment<byte> messageData)
        {
            using (XmlDictionaryReader xmlDictionaryReader = XmlDictionaryReader.CreateTextReader(messageData.Array, messageData.Offset, messageData.Count, null, XmlDictionaryReaderQuotas.Max, null))
            {
                MessageLogger.LogMessage(ref message, xmlDictionaryReader, MessageLoggingSource.TransportSend);
            }
        }

        protected override void AddHeadersTo(Message message)
        {
            Fx.Assert(message != null, "Message can't be null");
            
            if (message is NullMessage)
            {
                return; 
            }

            if (message.Version.Addressing != AddressingVersion.None)
            {
                if (message.Headers.MessageId == null)
                {
                    message.Headers.MessageId = new UniqueId();
                }
            }
            else
            {
                if (this.retransmitSettings.Enabled == true)
                {
                    // we should only get here if some channel above us starts producing messages that don't match the encoder's message version.
                    throw FxTrace.Exception.AsError(new ProtocolException(SR.RetransmissionRequiresAddressingOnMessage(message.Version.Addressing.ToString())));
                }
            }
        }

        protected override IAsyncResult OnBeginSend(Message message, TimeSpan timeout, AsyncCallback callback, object state)
        {
            if (message is NullMessage)
            {
                return new CompletedAsyncResult(callback, state); 
            }
            
            return new SendAsyncResult(this, message, timeout, callback, state);
        }

        protected override void OnEndSend(IAsyncResult result)
        {
            if (result is CompletedAsyncResult)
            {
                CompletedAsyncResult.End(result); 
            }
            else 
            {
                SendAsyncResult.End(result);
            }
        }

        protected override void OnSend(Message message, TimeSpan timeout)
        {
            if (message is NullMessage)
            {
                return; 
            }
        
            TimeoutHelper timeoutHelper = new TimeoutHelper(timeout);

            IPEndPoint remoteEndPoint;
            UdpSocket[] sendSockets;
            Exception exceptionToBeThrown;
            sendSockets = this.GetSendSockets(message, out remoteEndPoint, out exceptionToBeThrown);

            if (exceptionToBeThrown != null)
            {
                throw FxTrace.Exception.AsError(exceptionToBeThrown);
            }

            if (timeoutHelper.RemainingTime() <= TimeSpan.Zero)
            {
                throw FxTrace.Exception.AsError(new TimeoutException(SR.SendTimedOut(remoteEndPoint, timeout)));
            }

            bool returnBuffer = false;
            ArraySegment<byte> messageData = default(ArraySegment<byte>);

            bool sendingMulticast = UdpUtility.IsMulticastAddress(remoteEndPoint.Address);
            SynchronousRetransmissionHelper retransmitHelper = null;
            RetransmitIterator retransmitIterator = null;

            bool shouldRetransmit = this.ShouldRetransmitMessage(sendingMulticast);
            
            try
            {
                if (shouldRetransmit)
                {
                    retransmitIterator = this.CreateRetransmitIterator(sendingMulticast);
                    retransmitHelper = new SynchronousRetransmissionHelper(sendingMulticast);
                    this.RetransmitStarting(message.Headers.MessageId, retransmitHelper);
                }

                messageData = this.EncodeMessage(message);
                returnBuffer = true;

                this.TransmitMessage(messageData, sendSockets, remoteEndPoint, timeoutHelper);

                if (shouldRetransmit)
                {
                    while (retransmitIterator.MoveNext())
                    {
                        // wait for currentDelay time, then retransmit
                        if (retransmitIterator.CurrentDelay > 0)
                        {
                            retransmitHelper.Wait(retransmitIterator.CurrentDelay);
                        }

                        if (retransmitHelper.IsCanceled)
                        {
                            ThrowIfAborted();
                            return;
                        }

                        // since we only invoke the encoder once just before the initial send of the message
                        // we need to handle logging the message in the retransmission case
                        if (MessageLogger.LogMessagesAtTransportLevel)
                        {
                            UdpOutputChannel.LogMessage(ref message, messageData);
                        }

                        this.TransmitMessage(messageData, sendSockets, remoteEndPoint, timeoutHelper);
                    }
                }
            }
            finally
            {
                if (returnBuffer)
                {
                    this.BufferManager.ReturnBuffer(messageData.Array);
                }

                if (shouldRetransmit)
                {
                    this.RetransmitStopping(message.Headers.MessageId);

                    if (retransmitHelper != null)
                    {
                        retransmitHelper.Dispose();
                    }
                }
            }
        }

        protected abstract UdpSocket[] GetSendSockets(Message message, out IPEndPoint remoteEndPoint, out Exception exceptionToBeThrown);

        protected override void OnAbort()
        {
            this.Cleanup(true, TimeSpan.Zero);
        }

        [SuppressMessage("Microsoft.StyleCop.CSharp.ReadabilityRules", "SA1100:DoNotPrefixCallsWithBaseUnlessLocalImplementationExists", Justification = "If BeginClose is overridden we still pass base.BeginClose here")]
        protected override IAsyncResult OnBeginClose(TimeSpan timeout, AsyncCallback callback, object state)
        {
            return new CloseAsyncResult(
                this,
                timeout,
                callback,
                state);
        }

        protected override void OnClose(TimeSpan timeout)
        {
            TimeoutHelper timeoutHelper = new TimeoutHelper(timeout);
            this.Cleanup(false, timeoutHelper.RemainingTime());
        }

        protected override void OnEndClose(IAsyncResult result)
        {
            CloseAsyncResult.End(result);
        }

        protected override IAsyncResult OnBeginOpen(TimeSpan timeout, AsyncCallback callback, object state)
        {
            this.OnOpen(timeout);
            return new CompletedAsyncResult(callback, state);
        }

        protected override void OnEndOpen(IAsyncResult result)
        {
            CompletedAsyncResult.End(result);
        }

        protected override void OnOpen(TimeSpan timeout)
        {
            for (int i = 0; i < this.SendSockets.Length; i++)
            {
                this.SendSockets[i].Open();
            }
        }

        protected ArraySegment<byte> EncodeMessage(Message message)
        {
            return this.Encoder.WriteMessage(message, int.MaxValue, this.BufferManager);
        }

        protected ObjectDisposedException CreateObjectDisposedException()
        {
            return new ObjectDisposedException(null, SR.ObjectDisposed(this.GetType().Name));
        }

        private RetransmitIterator CreateRetransmitIterator(bool sendingMulticast)
        {
            Fx.Assert(this.retransmitSettings.Enabled, "CreateRetransmitCalculator called when no retransmission set to happen");
            int lowerBound = this.retransmitSettings.GetDelayLowerBound();
            int upperBound = this.retransmitSettings.GetDelayUpperBound();
            int currentDelay = this.randomNumberGenerator.Next(lowerBound, upperBound);

            int maxDelay = this.retransmitSettings.GetMaxDelayPerRetransmission();
            int maxRetransmitCount = sendingMulticast ? this.retransmitSettings.MaxMulticastRetransmitCount : this.retransmitSettings.MaxUnicastRetransmitCount;

            return new RetransmitIterator(currentDelay, maxDelay, maxRetransmitCount);
        }

        private void RetransmitStarting(UniqueId messageId, IUdpRetransmitter retransmitter)
        {
            Fx.Assert(this.retransmitSettings.Enabled, "RetransmitStarting called when retransmission is disabled");

            lock (this.ThisLock)
            {
                ThrowIfDisposed();

                if (this.retransmitList.ContainsKey(messageId))
                {
                    // someone is sending a message with the same MessageId 
                    // while a retransmission is still in progress for that ID.
                    throw FxTrace.Exception.AsError(new InvalidOperationException(SR.RecycledMessageIdDuringRetransmission(messageId)));
                }
                else
                {
                    this.retransmitList[messageId] = retransmitter;
                }
            }
        }

        private void RetransmitStopping(UniqueId messageId)
        {
            Fx.Assert(this.retransmitSettings.Enabled, "RetransmitStopping called when retransmission is disabled");

            lock (this.ThisLock)
            {
                // Cleanup sets retransmitList to null, so check before using...
                if (this.retransmitList != null)
                {
                    this.retransmitList.Remove(messageId);

                    // if we are closing down, then we need to unblock the Cleanup code 
                    //  this.retransmissionDoneEvent only != null if on cleaning up; abort case means that it == null. 
                    if (this.retransmitList.Count == 0 && this.retransmissionDoneWaitHandle != null)
                    {
                        this.retransmissionDoneWaitHandle.Set();
                    }
                }
            }
        }

        private bool ShouldRetransmitMessage(bool sendingMulticast)
        {
            if (sendingMulticast)
            {
                return this.retransmitSettings.MaxMulticastRetransmitCount > 0;
            }
            else
            {
                return this.retransmitSettings.MaxUnicastRetransmitCount > 0;
            }
        }

        private void TransmitMessage(ArraySegment<byte> messageBytes, UdpSocket[] sockets, IPEndPoint remoteEndpoint, TimeoutHelper timeoutHelper)
        {
            Fx.Assert(messageBytes.Array != null, "message data array can't be null");
            Fx.Assert(sockets != null, "sockets can't be null");
            Fx.Assert(sockets.Length > 0, "sockets must contain at least one item");
            Fx.Assert(remoteEndpoint != null, "remoteEndPoint can't be null");

            for (int i = 0; i < sockets.Length; i++)
            {
                if (timeoutHelper.RemainingTime() <= TimeSpan.Zero)
                {
                    throw FxTrace.Exception.AsError(new TimeoutException(SR.SendTimedOut(remoteEndpoint, timeoutHelper.OriginalTimeout)));
                }

                sockets[i].SendTo(messageBytes.Array, messageBytes.Offset, messageBytes.Count, remoteEndpoint);
            }
        }

        // we're guaranteed by CommunicationObject that at most ONE of Close or BeginClose will be called once. 
        private void Cleanup(bool aborting, TimeSpan timeout)
        {
            bool needToWait = false;

            if (this.cleanedUp)
            {
                return;
            }

            lock (this.ThisLock)
            {
                if (this.cleanedUp)
                {
                    return;
                }

                if (!aborting && this.retransmitList != null && this.retransmitList.Count > 0)
                {
                    needToWait = true;
                    this.retransmissionDoneWaitHandle = new AsyncWaitHandle(EventResetMode.ManualReset);
                }
                else
                {
                    // copied this call here in order to avoid releasing then retaking lock 
                    this.CleanupAfterWait(aborting);
                }
            }

            if (needToWait)
            {
                if (!this.retransmissionDoneWaitHandle.Wait(timeout))
                {
                    throw FxTrace.Exception.AsError(new TimeoutException(SR.TimeoutOnOperation(timeout)));
                }

                lock (this.ThisLock)
                {
                    this.retransmissionDoneWaitHandle = null;

                    // another thread could have called Abort while Close() was waiting for retransmission to complete.
                    if (this.cleanedUp)
                    {
                        return;
                    }

                    this.CleanupAfterWait(aborting);
                }
            }
        }

        // must be called from within this.ThisLock
        private void CleanupAfterWait(bool aborting)
        {
            Fx.Assert(!this.cleanedUp, "We should only clean up once");

            if (this.retransmitList != null)
            {
                foreach (IUdpRetransmitter retransmitter in this.retransmitList.Values)
                {
                    retransmitter.CancelRetransmission();
                }

                if (aborting && this.retransmissionDoneWaitHandle != null)
                {
                    // If another thread has called close and is waiting for retransmission to complete,
                    // we need to make sure that thread gets unblocked.
                    this.retransmissionDoneWaitHandle.Set();
                }

                this.retransmitList = null;
            }

            for (int i = 0; i < this.SendSockets.Length; i++)
            {
                this.SendSockets[i].Close();
            }                

            this.cleanedUp = true;
        }

        private class RetransmitIterator
        {
            private int maxDelay;
            private int retransmitCount;
            private int initialDelay;

            internal RetransmitIterator(int initialDelay, int maxDelay, int retransmitCount)
            {
                Fx.Assert(initialDelay >= 0, "initialDelay cannot be negative");
                Fx.Assert(maxDelay >= initialDelay, "maxDelay must be >= initialDelay");
                Fx.Assert(retransmitCount > 0, "retransmitCount must be > 0");

                this.CurrentDelay = -1;
                this.initialDelay = initialDelay;
                this.maxDelay = maxDelay;
                this.retransmitCount = retransmitCount;
            }

            public int CurrentDelay
            {
                get;
                private set;
            }

            // should be called before each retransmission to determine if 
            // another one is needed.
            public bool MoveNext()
            {
                if (this.CurrentDelay < 0)
                {
                    this.CurrentDelay = this.initialDelay;
                    return true;
                }

                bool shouldContinue = --this.retransmitCount > 0;

                if (shouldContinue && this.CurrentDelay < this.maxDelay)
                {
                    this.CurrentDelay = Math.Min(this.CurrentDelay * 2, this.maxDelay);
                }

                return shouldContinue;
            }
        }

        private sealed class SynchronousRetransmissionHelper : IUdpRetransmitter, IDisposable
        {
            private ManualResetEvent cancelEvent;
            private object thisLock;
            private bool cleanedUp;

            public SynchronousRetransmissionHelper(bool isMulticast)
            {
                this.thisLock = new object();
                this.IsMulticast = isMulticast;
                this.cancelEvent = new ManualResetEvent(false);
            }

            public bool IsMulticast
            {
                get;
                private set;
            }

            public bool IsCanceled
            {
                get;
                private set;
            }

            public void Wait(int millisecondsTimeout)
            {
                if (this.ResetEvent())
                {
                    // Dispose should only be called by the same thread that
                    // is calling this function, making it so that we don't need a lock here...
                    this.cancelEvent.WaitOne(millisecondsTimeout);
                }
            }

            public void CancelRetransmission()
            {
                lock (this.thisLock)
                {
                    this.IsCanceled = true;

                    if (!this.cleanedUp)
                    {
                        this.cancelEvent.Set();
                    }
                }
            }

            public void Dispose()
            {
                lock (this.thisLock)
                {
                    if (!this.cleanedUp)
                    {
                        this.cleanedUp = true;
                        this.cancelEvent.Dispose();
                    }
                }
            }

            private bool ResetEvent()
            {
                lock (this.thisLock)
                {
                    if (!this.IsCanceled && !this.cleanedUp)
                    {
                        this.cancelEvent.Reset();
                        return true;
                    }
                }

                return false;
            }
        }

        private class SendAsyncResult : AsyncResult, IUdpRetransmitter
        {
            private static AsyncCallback onSocketSendComplete = Fx.ThunkCallback(new AsyncCallback(OnSocketSendComplete));
            private static Action<object> onRetransmitMessage = new Action<object>(OnRetransmitMessage);

            private UdpOutputChannel channel;
            private ArraySegment<byte> messageData;
            private TimeoutHelper timeoutHelper;
            private IPEndPoint remoteEndpoint;
            private int currentSocket;
            private UdpSocket[] sendSockets;
            private IOThreadTimer retransmitTimer;
            private RetransmitIterator retransmitIterator;
            private Message message;
            private bool retransmissionEnabled;

            public SendAsyncResult(UdpOutputChannel channel, Message message, TimeSpan timeout, AsyncCallback callback, object state)
                : base(callback, state)
            {
                this.timeoutHelper = new TimeoutHelper(timeout);

                this.channel = channel;
                this.message = message;
                bool throwing = true;
                bool completedSynchronously = false;

                try
                {
                    this.Initialize(message);

                    completedSynchronously = this.BeginTransmitMessage();

                    if (completedSynchronously && this.retransmissionEnabled)
                    {
                        // initial send completed sync, now we need to start the retransmission process...
                        completedSynchronously = this.BeginRetransmission();
                    }

                    throwing = false;
                }
                finally
                {
                    if (throwing)
                    {
                        this.Cleanup();
                    }
                }

                if (completedSynchronously)
                {
                    this.CompleteAndCleanup(true, null);
                }
            }

            private enum RetransmitState
            {
                WaitCompleted,
                TransmitCompleted,
            }

            public bool IsCanceled
            {
                get;
                private set;
            }

            public bool IsMulticast
            {
                get;
                private set;
            }

            public static void End(IAsyncResult result)
            {
                AsyncResult.End<SendAsyncResult>(result);
            }

            // tries to terminate retransmission early, but won't cancel async IO immediately
            public void CancelRetransmission()
            {
                this.IsCanceled = true;
            }

            private static void OnSocketSendComplete(IAsyncResult result)
            {
                if (result.CompletedSynchronously)
                {
                    return;
                }

                SendAsyncResult thisPtr = (SendAsyncResult)result.AsyncState;
                bool completeSelf = false;
                Exception completionException = null;

                try
                {
                    completeSelf = thisPtr.ContinueTransmitting(result);

                    if (completeSelf && thisPtr.retransmissionEnabled)
                    {
                        completeSelf = thisPtr.BeginRetransmission();
                    }
                }
                catch (Exception e)
                {
                    if (Fx.IsFatal(e))
                    {
                        throw;
                    }

                    completionException = e;
                    completeSelf = true;
                }

                if (completeSelf)
                {
                    thisPtr.CompleteAndCleanup(false, completionException);
                }
            }

            private static void OnRetransmitMessage(object state)
            {
                SendAsyncResult thisPtr = (SendAsyncResult)state;
                bool completeSelf = false;
                Exception completionException = null;

                try
                {
                    completeSelf = thisPtr.ContinueRetransmission(RetransmitState.WaitCompleted);
                }
                catch (Exception e)
                {
                    if (Fx.IsFatal(e))
                    {
                        throw;
                    }

                    completionException = e;
                    completeSelf = true;
                }

                if (completeSelf)
                {
                    thisPtr.CompleteAndCleanup(false, completionException);
                }
            }

            private bool BeginTransmitMessage()
            {
                this.currentSocket = 0;
                return this.ContinueTransmitting(null);
            }

            private bool ContinueTransmitting(IAsyncResult socketAsyncResult)
            {
                while (this.currentSocket < this.sendSockets.Length)
                {
                    if (socketAsyncResult == null)
                    {
                        socketAsyncResult = this.sendSockets[this.currentSocket].BeginSendTo(
                            this.messageData.Array,
                            this.messageData.Offset,
                            this.messageData.Count,
                            this.remoteEndpoint,
                            onSocketSendComplete,
                            this);

                        if (!socketAsyncResult.CompletedSynchronously)
                        {
                            return false;
                        }
                    }

                    this.sendSockets[this.currentSocket].EndSendTo(socketAsyncResult);

                    // check for timeout after calling socket.EndSendTo 
                    // so that we don't leave the socket in a bad state/leak async results
                    this.ThrowIfTimedOut();

                    if (this.IsCanceled)
                    {
                        // don't send on the next socket and return true to cause Complete to be called.
                        return true;
                    }

                    this.currentSocket++;
                    socketAsyncResult = null;
                }

                return true;
            }

            private bool BeginRetransmission()
            {
                // BeginRetransmission should only be called in the case where transmission of the message
                // completes synchronously.
                return this.ContinueRetransmission(RetransmitState.TransmitCompleted);
            }

            private bool ContinueRetransmission(RetransmitState state)
            {
                this.ThrowIfTimedOut();

                while (true)
                {
                    switch (state)
                    {
                        case RetransmitState.TransmitCompleted:
                            if (!this.retransmitIterator.MoveNext())
                            {
                                // We are done retransmitting
                                return true;
                            }

                            if (this.retransmitIterator.CurrentDelay > 0)
                            {
                                this.retransmitTimer.Set(this.retransmitIterator.CurrentDelay);
                                return false;
                            }

                            state = RetransmitState.WaitCompleted;
                            break;
                        case RetransmitState.WaitCompleted:
                            if (this.IsCanceled)
                            {
                                this.channel.ThrowIfAborted();
                                return true;
                            }

                            // since we only invoke the encoder once just before the initial send of the message
                            // we need to handle logging the message in the retransmission case
                            if (MessageLogger.LogMessagesAtTransportLevel)
                            {
                                UdpOutputChannel.LogMessage(ref this.message, this.messageData);
                            }

                            // !completedSync
                            if (!this.BeginTransmitMessage())
                            {
                                return false;
                            }

                            state = RetransmitState.TransmitCompleted;
                            break;

                        default:
                            Fx.Assert("Unknown RetransmitState value encountered");
                            return true;
                    }
                }
            }

            private void Initialize(Message message)
            {
                Exception exceptionToThrow;
                this.sendSockets = this.channel.GetSendSockets(message, out this.remoteEndpoint, out exceptionToThrow);

                if (exceptionToThrow != null)
                {
                    throw FxTrace.Exception.AsError(exceptionToThrow);
                }

                this.IsMulticast = UdpUtility.IsMulticastAddress(this.remoteEndpoint.Address);

                if (this.channel.ShouldRetransmitMessage(this.IsMulticast))
                {
                    this.retransmissionEnabled = true;
                    this.channel.RetransmitStarting(this.message.Headers.MessageId, this);
                    this.retransmitTimer = new IOThreadTimer(onRetransmitMessage, this, false);
                    this.retransmitIterator = this.channel.CreateRetransmitIterator(this.IsMulticast);
                }

                this.messageData = this.channel.EncodeMessage(message);
            }

            private void ThrowIfTimedOut()
            {
                if (this.timeoutHelper.RemainingTime() <= TimeSpan.Zero)
                {
                    throw FxTrace.Exception.AsError(new TimeoutException(SR.TimeoutOnOperation(this.timeoutHelper.OriginalTimeout)));
                }
            }

            private void Cleanup()
            {
                if (this.retransmissionEnabled)
                {
                    this.channel.RetransmitStopping(this.message.Headers.MessageId);
                    this.retransmitTimer.Cancel();
                }

                if (this.messageData.Array != null)
                {
                    this.channel.BufferManager.ReturnBuffer(this.messageData.Array);
                    this.messageData = default(ArraySegment<byte>);
                }
            }

            private void CompleteAndCleanup(bool completedSynchronously, Exception completionException)
            {
                this.Cleanup();
                this.Complete(completedSynchronously, completionException);
            }
        }

        // Control flow for async path
        // We use this mechanism to avoid initializing two async objects as logically cleanup+close is one operation. 
        // At any point in the Begin* methods, we may go async. The steps are: 
        // - Cleanup channel
        // - Close channel
        private class CloseAsyncResult : AsyncResult
        {
            private static Action<object, TimeoutException> completeCleanupCallback = new Action<object, TimeoutException>(CompleteCleanup);

            private UdpOutputChannel channel;
            private TimeoutHelper timeoutHelper;

            public CloseAsyncResult(UdpOutputChannel channel, TimeSpan timeout, AsyncCallback callback, object state)
                : base(callback, state)
            {
                this.channel = channel;
                this.timeoutHelper = new TimeoutHelper(timeout);

                if (this.BeginCleanup())
                {
                    this.Complete(true);
                }
            }

            internal static void End(IAsyncResult result)
            {
                AsyncResult.End<CloseAsyncResult>(result);
            }

            private static void CompleteCleanup(object state, TimeoutException exception)
            {
                CloseAsyncResult thisPtr = (CloseAsyncResult)state;
                Exception completionException = null;

                if (exception != null)
                {
                    Fx.Assert(exception.GetType() == typeof(TimeoutException), "Exception on callback should always be TimeoutException");
                    throw FxTrace.Exception.AsError(new TimeoutException(SR.TimeoutOnOperation(thisPtr.timeoutHelper.OriginalTimeout)));
                }

                try
                {
                    lock (thisPtr.channel.ThisLock)
                    {
                        thisPtr.channel.retransmissionDoneWaitHandle = null;

                        // another thread could have called Abort while Close() was waiting for retransmission to complete.
                        if (!thisPtr.channel.cleanedUp)
                        {
                            // never aborting here
                            thisPtr.channel.CleanupAfterWait(false);
                        }
                    }
                }
                catch (Exception e)
                {
                    if (Fx.IsFatal(e))
                    {
                        throw;
                    }

                    completionException = e;
                }

                thisPtr.Complete(false, completionException);
            }

            private bool BeginCleanup()
            {
                bool needToWait = false;

                if (this.channel.cleanedUp)
                {
                    return true;
                }

                lock (this.channel.ThisLock)
                {
                    if (this.channel.cleanedUp)
                    {
                        return true;
                    }

                    // we're never aborting in this case...
                    if (this.channel.retransmitList != null && this.channel.retransmitList.Count > 0)
                    {
                        needToWait = true;
                        this.channel.retransmissionDoneWaitHandle = new AsyncWaitHandle(EventResetMode.ManualReset);
                    }
                    else
                    {
                        this.channel.CleanupAfterWait(false);
                    }

                    // we're guaranteed by CommunicationObject that at most ONE of Close or BeginClose will be called once. 
                    // we don't null out retransmissionDoneEvent in the abort case; should be safe to use here. 
                    return !needToWait || this.channel.retransmissionDoneWaitHandle.WaitAsync(completeCleanupCallback, this, this.timeoutHelper.RemainingTime());
                }
            }
        }
    }
}