using Google.Protobuf;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.Linq;
using System.Reactive.Subjects;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Tango.Core;
using Tango.Logging;
using Tango.PMR;
using Tango.PMR.Common;
using System.Reactive.Linq;
using System.ServiceModel;
using Tango.Transport.Encoders;
using Tango.PMR.Connection;
using Tango.Core.Threading;

namespace Tango.Transport
{
    /// <summary>
    /// Represents an <see cref="ITransporter"/> base class.
    /// </summary>
    /// <seealso cref="Tango.Transport.ITransporter" />
    public abstract class TransporterBase : ExtendedObject, ITransporter
    {
        private const int MESSAGE_TOKEN_LENGTH = 36;
        private ProducerConsumerQueue<TransportMessageBase> _sendingQueue;
        private ConcurrentList<TransportMessageBase> _pendingRequests;
        private ProducerConsumerQueue<byte[]> _arrivedResponses;
        private Thread _pushThread;
        private Thread _pullThread;
        private Thread _keepAliveThread;
        private ITransportAdapter _adapter;
        private Dictionary<String, PendingResponse> _pendingResponses;
        private DateTime _lastKeepAliveTime;

        #region Events

        /// <summary>
        /// Occurs when a new request message has been received.
        /// </summary>
        public event EventHandler<MessageContainer> RequestReceived;

        /// <summary>
        /// Occurs when a new response message has been received.
        /// </summary>
        public event EventHandler<MessageContainer> PendingResponseReceived;

        /// <summary>
        /// Occurs when component state changes.
        /// </summary>
        public event EventHandler<TransportComponentState> StateChanged;

        #endregion

        #region Properties

        /// <summary>
        /// Gets or sets the <see cref="ITransportAdapter" /> used to read and write raw data.
        /// </summary>
        public ITransportAdapter Adapter
        {
            get { return _adapter; }
            set
            {
                var previous = _adapter;
                _adapter = value;
                OnAdapterChanged(previous, value);
                RaisePropertyChangedAuto();
            }
        }

        /// <summary>
        /// Gets or sets the transport encoder used to encode and decode tango messages.
        /// </summary>
        public ITransportEncoder Encoder { get; set; }

        private TransportComponentState _state;
        /// <summary>
        /// Gets the component state.
        /// </summary>
        public TransportComponentState State
        {
            get { return _state; }
            protected set
            {
                if (_state != value)
                {
                    _state = value;
                    OnStateChanged(_state);
                }
            }
        }

        /// <summary>
        /// Gets or sets the request timeout.
        /// </summary>
        public TimeSpan RequestTimeout { get; set; }

        private bool _useKeepAlive;
        /// <summary>
        /// Gets or sets a value indicating whether to use a keep alive mechanism.
        /// </summary>
        public bool UseKeepAlive
        {
            get { return _useKeepAlive; }
            set
            {
                _useKeepAlive = value;
                RaisePropertyChangedAuto();

                if (_useKeepAlive)
                {
                    LogManager.Log("KeepAlive is activated...");
                }
                else
                {
                    LogManager.Log("KeepAlive is deactivated.");
                }
            }
        }

        /// <summary>
        /// Gets or sets the keep alive timeout.
        /// </summary>
        public TimeSpan KeepAliveTimeout { get; set; }

        /// <summary>
        /// Gets or sets the keep alive retries.
        /// </summary>
        public int KeepAliveRetries { get; set; }

        /// <summary>
        /// Gets or sets a value indicating whether to auto respond to keep alive requests.
        /// </summary>
        public bool EnableKeepAliveAutoResponse { get; set; }

        /// <summary>
        /// Gets or sets a value indicating whether the transporter will get in to a failed state if any adapter has failed.
        /// </summary>
        public bool FailsWithAdapter { get; set; }

        /// <summary>
        /// Gets the last failed state exception/reason.
        /// </summary>
        public Exception FailedStateException { get; private set; }

        #endregion

        #region Virtual Methods

        /// <summary>
        /// Called when the <see cref="Adapter"/> has changed.
        /// </summary>
        /// <param name="newAdapter">The adapter.</param>
        protected async virtual void OnAdapterChanged(ITransportAdapter oldAdapter, ITransportAdapter newAdapter)
        {
            if (oldAdapter != newAdapter)
            {
                _pendingRequests.Clear();
                _pendingResponses.Clear();
                _arrivedResponses = new ProducerConsumerQueue<byte[]>();
                _sendingQueue = new ProducerConsumerQueue<TransportMessageBase>();
            }

            if (oldAdapter != null)
            {
                oldAdapter.StateChanged -= OnAdapterStateChanged;
                oldAdapter.DataAvailable -= OnAdapterDataAvailable;
            }

            if (newAdapter != null)
            {
                LogManager.Log(String.Format("Adapter Changed: Type = {0}, Address = {1}, State = {2}", newAdapter.GetType().Name, newAdapter.Address, newAdapter.State));

                newAdapter.StateChanged -= OnAdapterStateChanged;
                newAdapter.DataAvailable -= OnAdapterDataAvailable;
                newAdapter.StateChanged += OnAdapterStateChanged;
                newAdapter.DataAvailable += OnAdapterDataAvailable;

                if (State == TransportComponentState.Connected && newAdapter.State == TransportComponentState.Disconnected)
                {
                    await newAdapter.Connect();
                }
            }
            else
            {
                LogManager.Log("Adapter Changed: null");
            }
        }

        /// <summary>
        /// Called when the current adapter state has changed.
        /// </summary>
        /// <param name="sender">The sender.</param>
        /// <param name="e">The e.</param>
        protected virtual void OnAdapterStateChanged(object sender, TransportComponentState e)
        {
            if (e == TransportComponentState.Failed && FailsWithAdapter)
            {
                OnFailed(new CommunicationException("The adapter has failed. Going into a failed state..."));
            }
        }

        /// <summary>
        /// Called when there is data available from the adapter.
        /// </summary>
        /// <param name="sender">The sender.</param>
        /// <param name="data">The data.</param>
        protected virtual void OnAdapterDataAvailable(object sender, byte[] data)
        {
            _lastKeepAliveTime = DateTime.Now;
            EnqueueMessageIn(data);
        }

        /// <summary>
        /// Called when the component has failed.
        /// </summary>
        /// <param name="ex">The ex.</param>
        protected virtual void OnFailed(Exception ex)
        {
            FailedStateException = ex;
            State = TransportComponentState.Failed;
            LogManager.Log(ex, "Transporter failed.");
            Disconnect().Wait();
        }

        /// <summary>
        /// Called when a new request has been received.
        /// </summary>
        /// <param name="container">The request.</param>
        protected virtual void OnRequestReceived(MessageContainer container)
        {
            RequestReceived?.Invoke(this, container);
        }

        /// <summary>
        /// Called when a new response has been received.
        /// </summary>
        /// <param name="container">The request.</param>
        protected virtual void OnResponseReceived(MessageContainer container)
        {
            PendingResponseReceived?.Invoke(this, container);
        }

        /// <summary>
        /// Called when the component state has changed.
        /// </summary>
        /// <param name="state">The state.</param>
        protected virtual void OnStateChanged(TransportComponentState state)
        {
            StateChanged?.Invoke(this, state);
        }

        #endregion

        #region Constructors

        /// <summary>
        /// Initializes a new instance of the <see cref="TransporterBase"/> class.
        /// </summary>
        public TransporterBase()
        {
            Encoder = new ProtoEncoder();
            _pendingResponses = new Dictionary<string, PendingResponse>();
            _sendingQueue = new ProducerConsumerQueue<TransportMessageBase>();
            _pendingRequests = new ConcurrentList<TransportMessageBase>();
            _arrivedResponses = new ProducerConsumerQueue<byte[]>();
            RequestTimeout = TimeSpan.FromSeconds(5);
            EnableKeepAliveAutoResponse = true;
            KeepAliveTimeout = TimeSpan.FromSeconds(2);
            KeepAliveRetries = 1;
        }

        /// <summary>
        /// Initializes a new instance of the <see cref="TransporterBase"/> class.
        /// </summary>
        /// <param name="adapter">The transport adapter.</param>
        public TransporterBase(ITransportAdapter adapter) : this()
        {
            Adapter = adapter;
        }

        #endregion

        #region Public Methods

        /// <summary>
        /// Clears all message queues.
        /// </summary>
        public void ClearQueues()
        {
            _sendingQueue = new ProducerConsumerQueue<TransportMessageBase>();
            _pendingRequests = new ConcurrentList<TransportMessageBase>();
            _arrivedResponses = new ProducerConsumerQueue<byte[]>();
        }

        /// <summary>
        /// Connects the transport component.
        /// </summary>
        /// <returns></returns>
        public virtual async Task Connect()
        {
            if (Adapter != null)
            {
                await Adapter.Connect();
            }

            State = TransportComponentState.Connected;
            StartThreads();

            LogManager.Log("Transporter Connected...");
        }

        /// <summary>
        /// Disconnects the transport component.
        /// </summary>
        /// <returns></returns>
        public virtual async Task Disconnect()
        {
            State = TransportComponentState.Disconnected;

            try
            {
                if (_pullThread != null)
                {
                    _pullThread.Abort();
                    _pushThread.Abort();
                    _keepAliveThread.Abort();
                }
            }
            catch { }

            if (Adapter != null)
            {
                await Adapter.Disconnect();
            }
            LogManager.Log("Transporter Disconnected...");
        }

        /// <summary>
        /// Sends a request.
        /// </summary>
        /// <param name="request">The request.</param>
        /// <param name="timeout">Optional timeout. If not specified will use the <see cref="RequestTimeout" />.</param>
        /// <returns></returns>
        public Task<IMessage> SendRequest(IMessage request, TimeSpan? timeout = null)
        {
            String requestName = request.GetType().Name;
            String responseName = requestName.Replace("Request", "Response");

            MessageContainer container = new MessageContainer();
            container.Token = Guid.NewGuid().ToString();
            container.Data = request.ToByteString();
            container.Timeout = timeout.HasValue ? (UInt32)timeout.Value.TotalMilliseconds : (UInt32)RequestTimeout.TotalMilliseconds;
            container.Type = MessageFactory.ParseMessageType(requestName);

            LogManager.Log("Queuing request message: " + requestName + " Token: " + container.Token, LogCategory.Debug);
            LogManager.Log("Expected response: " + responseName, LogCategory.Debug);

            if (State != TransportComponentState.Connected)
            {
                throw LogManager.Log(new InvalidOperationException($"Could not send the request while transporter state is {State}."));
            }

            TaskCompletionSource<IMessage> source = new TaskCompletionSource<IMessage>();
            TransportMessage<IMessage> message = new TransportMessage<IMessage>(container.Token, request, TransportMessageDirection.Request, () => container.ToByteArray(), source);

            message.ActivateTimeout = () =>
            {
                TimeoutTask.StartNew(() =>
                {

                    if (!source.Task.IsCompleted)
                    {
                        TimeoutException ex = new TimeoutException("Request message: " + requestName + " had timed out after " + (timeout != null ? timeout.Value.TotalSeconds : RequestTimeout.TotalSeconds) + " seconds.");
                        LogManager.Log(ex);
                        LogManager.Log("Setting request task exception...", LogCategory.Debug);
                        source.SetException(ex);
                    }

                }, timeout != null ? timeout.Value : RequestTimeout);
            };

            EnqueueMessageOut(message);

            return source.Task;
        }

        /// <summary>
        /// Sends the response.
        /// </summary>
        /// <param name="response">The response.</param>
        /// <param name="token">The token.</param>
        /// <param name="completed">The completed.</param>
        /// <param name="errorCode">The error code.</param>
        /// <param name="errorMessage">The error message.</param>
        /// <returns></returns>
        /// <exception cref="System.InvalidOperationException">Matching request token was not found!</exception>
        public Task SendResponse(IMessage response, string token, bool? completed = default(bool?), ErrorCode? errorCode = default(ErrorCode?), string errorMessage = null)
        {
            String responseName = response.GetType().Name;

            MessageContainer container = new MessageContainer();
            container.Token = token;
            container.Data = response.ToByteString();
            container.Type = MessageFactory.ParseMessageType(responseName);

            if (errorCode.HasValue)
            {
                container.Error = errorCode.Value;
            }

            if (errorMessage != null)
            {
                container.ErrorMessage = errorMessage;
            }

            if (completed.HasValue)
            {
                container.Completed = completed.Value;
            }

            return SendResponse(container);
        }

        /// <summary>
        /// Sends the request.
        /// </summary>
        /// <param name="container">The container.</param>
        /// <returns></returns>
        public Task<MessageContainer> SendRequest(MessageContainer container)
        {
            String responseName = container.Type.ToString().Replace("Request", "Response");
            TimeSpan? timeout = GetContainerTimeoutOrDefault(container);

            LogManager.Log("Queuing request message: " + container.Type + " Token: " + container.Token, LogCategory.Debug);
            LogManager.Log("Expected response: " + responseName, LogCategory.Debug);

            if (State != TransportComponentState.Connected)
            {
                throw LogManager.Log(new InvalidOperationException($"Could not send the request while transporter state is {State}."));
            }

            TaskCompletionSource<MessageContainer> source = new TaskCompletionSource<MessageContainer>();
            TransportMessage<MessageContainer> message = new TransportMessage<MessageContainer>(container.Token, container, TransportMessageDirection.Request, () => container.ToByteArray(), source);

            message.ActivateTimeout = () =>
            {
                TimeoutTask.StartNew(() =>
                {

                    if (!source.Task.IsCompleted)
                    {
                        TimeoutException ex = new TimeoutException("Request message: " + container.Type + " had timed out after " + (timeout != null ? timeout.Value.TotalSeconds : RequestTimeout.TotalSeconds) + " seconds.");
                        LogManager.Log(ex);
                        LogManager.Log("Setting request task exception...", LogCategory.Debug);
                        source.SetException(ex);
                    }

                }, timeout != null ? timeout.Value : RequestTimeout);

            };

            EnqueueMessageOut(message);

            return source.Task;
        }

        /// <summary>
        /// Sends the response.
        /// </summary>
        /// <param name="container">The container.</param>
        /// <returns></returns>
        /// <exception cref="System.InvalidOperationException">Matching request token was not found!</exception>
        public Task SendResponse(MessageContainer container)
        {
            String token = container.Token;

            LogManager.Log("Queuing response message: " + container.Type, LogCategory.Debug);

            PendingResponse pendingResponse = null;

            if (State != TransportComponentState.Connected)
            {
                throw LogManager.Log(new InvalidOperationException($"Could not send the response while transporter state is {State}."));
            }

            LogManager.Log("Searching for matching request token: " + token, LogCategory.Debug);

            if (_pendingResponses.TryGetValue(token, out pendingResponse))
            {
                LogManager.Log("Found matching request token: " + token, LogCategory.Debug);

                if (!pendingResponse.IsContinuous)
                {
                    LogManager.Log("Removing matching request token.", LogCategory.Debug);
                    _pendingResponses.Remove(token);
                }
                else if (container.Completed)
                {
                    LogManager.Log("Response completed. Removing matching request token.", LogCategory.Debug);
                    _pendingResponses.Remove(token);
                }
            }
            else
            {
                //This should never happen.
                throw LogManager.Log(new InvalidOperationException("Matching request token was not found!"), LogCategory.Critical);
            }

            TaskCompletionSource<object> source = new TaskCompletionSource<object>();
            TransportMessage<object> message = new TransportMessage<object>(token, container, TransportMessageDirection.Response, () => container.ToByteArray(), source);
            EnqueueMessageOut(message);
            return source.Task;
        }

        /// <summary>
        /// Sends a request and expecting multiple response messages.
        /// </summary>
        /// <param name="request">The request.</param>
        /// <param name="timeout"></param>
        /// <returns></returns>
        public IObservable<IMessage> SendContinuousRequest(IMessage request, TimeSpan? timeout = default(TimeSpan?))
        {
            String requestName = request.GetType().Name;
            String responseName = requestName.Replace("Request", "Response");

            MessageContainer container = new MessageContainer();
            container.Token = Guid.NewGuid().ToString();
            container.Data = request.ToByteString();
            container.Type = MessageFactory.ParseMessageType(requestName);
            container.Timeout = timeout.HasValue ? (UInt32)timeout.Value.TotalMilliseconds : (UInt32)RequestTimeout.TotalMilliseconds;
            container.Continuous = true;

            LogManager.Log("Queuing continuous request message: " + requestName + " Token: " + container.Token, LogCategory.Debug);

            if (State != TransportComponentState.Connected)
            {
                throw LogManager.Log(new InvalidOperationException($"Could not send the request while transporter state is {State}."));
            }

            Subject<IMessage> subject = new Subject<IMessage>();

            LogManager.Log("Expected response: " + responseName, LogCategory.Debug);

            TransportMessage<IMessage> message = new TransportMessage<IMessage>(container.Token, request, TransportMessageDirection.Request, () => container.ToByteArray(), null)
            {
                IsContinuous = true,
                ContinuesResponseSubject = subject,
            };

            message.ActivateTimeout = () =>
            {

                TimeoutTask.StartNew(() =>
                {

                    if (!message.AtLeastOneResponseReceived)
                    {
                        TimeoutException ex = new TimeoutException("Request message: " + requestName + " had timed out after " + (timeout != null ? timeout.Value.TotalSeconds : RequestTimeout.TotalSeconds) + " seconds.");
                        LogManager.Log(ex);
                        LogManager.Log("Setting request exception...", LogCategory.Debug);
                        message.SetException(ex);
                    }

                }, timeout != null ? timeout.Value : RequestTimeout);

            };

            EnqueueMessageOut(message);

            return subject.AsObservable();
        }

        /// <summary>
        /// Sends a request.
        /// </summary>
        /// <typeparam name="Request">The type of the request.</typeparam>
        /// <typeparam name="Response">The type of the response.</typeparam>
        /// <param name="request">The request.</param>
        /// <param name="timeout">Optional timeout. If not specified will use the <see cref="RequestTimeout" />.</param>
        /// <returns></returns>
        public Task<TangoMessage<Response>> SendRequest<Request, Response>(TangoMessage<Request> request, TimeSpan? timeout = null) where Request : IMessage<Request> where Response : IMessage<Response>
        {
            LogManager.Log("Queuing request message: " + typeof(Request).Name + " Token: " + request.Container.Token, LogCategory.Debug);
            LogManager.Log("Expected response: " + typeof(Response).Name, LogCategory.Debug);

            if (State != TransportComponentState.Connected)
            {
                throw LogManager.Log(new InvalidOperationException($"Could not send the request while transporter state is {State}."));
            }

            request.Container.Timeout = timeout.HasValue ? (UInt32)timeout.Value.TotalMilliseconds : (UInt32)RequestTimeout.TotalMilliseconds;

            TaskCompletionSource<TangoMessage<Response>> source = new TaskCompletionSource<TangoMessage<Response>>();
            TransportMessage<TangoMessage<Response>> message = new TransportMessage<TangoMessage<Response>>(request.Container.Token, request, TransportMessageDirection.Request, () => Encoder.Encode(request), source);

            message.ActivateTimeout = () =>
            {
                TimeoutTask.StartNew(() =>
                {

                    if (!source.Task.IsCompleted)
                    {
                        TimeoutException ex = new TimeoutException("Request message: " + typeof(Request).Name + " had timed out after " + (timeout != null ? timeout.Value.TotalSeconds : RequestTimeout.TotalSeconds) + " seconds.");
                        LogManager.Log(ex);
                        LogManager.Log("Setting request task exception...", LogCategory.Debug);
                        source.SetException(ex);
                    }

                }, timeout != null ? timeout.Value : RequestTimeout);
            };

            EnqueueMessageOut(message);

            return source.Task;
        }

        /// <summary>
        /// Sends a request and expecting multiple response messages.
        /// </summary>
        /// <typeparam name="Request">The type of the request.</typeparam>
        /// <typeparam name="Response">The type of the response.</typeparam>
        /// <param name="request">The request.</param>
        /// <param name="timeout"></param>
        /// <returns></returns>
        public IObservable<TangoMessage<Response>> SendContinuousRequest<Request, Response>(TangoMessage<Request> request, TimeSpan? firstTimeout = null, TimeSpan? continousTimeout = null) where Request : IMessage<Request> where Response : IMessage<Response>
        {
            LogManager.Log("Queuing continuous request message: " + typeof(Request).Name + " Token: " + request.Container.Token, LogCategory.Debug);

            Subject<TangoMessage<Response>> subject = new Subject<TangoMessage<Response>>();

            LogManager.Log("Expected response: " + typeof(Response).Name, LogCategory.Debug);

            if (State != TransportComponentState.Connected)
            {
                throw LogManager.Log(new InvalidOperationException($"Could not send the request while transporter state is {State}."));
            }

            request.Container.Continuous = true;
            request.Container.Completed = false;

            request.Container.Timeout = firstTimeout.HasValue ? (UInt32)firstTimeout.Value.TotalMilliseconds : (UInt32)RequestTimeout.TotalMilliseconds;
            request.Container.ContinuousTimeout = continousTimeout.HasValue ? (UInt32)continousTimeout.Value.TotalMilliseconds : 0;

            TransportMessage<TangoMessage<Response>> message = new TransportMessage<TangoMessage<Response>>(request.Container.Token, request, TransportMessageDirection.Request, () => Encoder.Encode(request), null)
            {
                IsContinuous = true,
                ContinuesResponseSubject = subject,
            };

            message.ActivateTimeout = () =>
            {
                TimeoutTask.StartNew(() =>
                {

                    if (!message.AtLeastOneResponseReceived)
                    {
                        TimeoutException ex = new TimeoutException("Request message: " + typeof(Request).Name + " had timed out after " + (firstTimeout != null ? firstTimeout.Value.TotalSeconds : RequestTimeout.TotalSeconds) + " seconds.");
                        LogManager.Log(ex);
                        LogManager.Log("Setting request exception...", LogCategory.Debug);
                        message.SetException(ex);
                    }

                    if (continousTimeout != null)
                    {
                        Task.Factory.StartNew(async () =>
                        {
                            while (!message.Completed)
                            {
                                await Task.Delay(continousTimeout.Value).ContinueWith((y) =>
                                {
                                    if (!message.Completed)
                                    {
                                        if (DateTime.Now - message.LastResponseTime > continousTimeout.Value)
                                        {
                                            TimeoutException ex = new TimeoutException("Continuous request message: " + typeof(Request).Name + " had failed to provide a response for a period of " + (continousTimeout.Value.TotalSeconds) + " seconds and has timed out.");
                                            LogManager.Log(ex);
                                            LogManager.Log("Setting request exception...", LogCategory.Debug);
                                            message.SetException(ex);
                                            return;
                                        }
                                    }
                                });
                            }
                        });
                    }

                }, firstTimeout != null ? firstTimeout.Value : RequestTimeout);
            };

            EnqueueMessageOut(message);

            return subject.AsObservable();
        }

        /// <summary>
        /// Sends a continuous request.
        /// </summary>
        /// <param name="container">The container.</param>
        /// <returns></returns>
        public IObservable<MessageContainer> SendContinuousRequest(MessageContainer container)
        {
            TimeSpan? timeout = GetContainerTimeoutOrDefault(container);
            TimeSpan? continuousTimeout = container.ContinuousTimeout > 0 ? TimeSpan.FromMilliseconds(container.ContinuousTimeout) : default(TimeSpan?);

            String requestName = container.Type.ToString();
            String responseName = requestName.Replace("Request", "Response");

            LogManager.Log("Queuing continuous request message: " + requestName + " Token: " + container.Token, LogCategory.Debug);

            if (State != TransportComponentState.Connected)
            {
                throw LogManager.Log(new InvalidOperationException($"Could not send the request while transporter state is {State}."));
            }

            Subject<MessageContainer> subject = new Subject<MessageContainer>();

            LogManager.Log("Expected response: " + responseName, LogCategory.Debug);

            TransportMessage<MessageContainer> message = new TransportMessage<MessageContainer>(container.Token, container, TransportMessageDirection.Request, () => container.ToByteArray(), null)
            {
                IsContinuous = true,
                ContinuesResponseSubject = subject,
            };

            message.ActivateTimeout = () =>
            {
                TimeoutTask.StartNew(() =>
                {

                    if (!message.AtLeastOneResponseReceived)
                    {
                        TimeoutException ex = new TimeoutException("Request message: " + requestName + " had timed out after " + (timeout != null ? timeout.Value.TotalSeconds : RequestTimeout.TotalSeconds) + " seconds.");
                        LogManager.Log(ex);
                        LogManager.Log("Setting request exception...", LogCategory.Debug);
                        message.SetException(ex);
                    }

                    if (continuousTimeout != null)
                    {
                        Task.Factory.StartNew(async () =>
                        {
                            while (!message.Completed)
                            {
                                await Task.Delay(continuousTimeout.Value).ContinueWith((y) =>
                                {
                                    if (!message.Completed)
                                    {
                                        if (DateTime.Now - message.LastResponseTime > continuousTimeout.Value)
                                        {
                                            TimeoutException ex = new TimeoutException("Continuous request message: " + requestName + " had failed to provide a response for a period of " + (continuousTimeout.Value.TotalSeconds) + " seconds and has timed out.");
                                            LogManager.Log(ex);
                                            LogManager.Log("Setting request exception...", LogCategory.Debug);
                                            message.SetException(ex);
                                            return;
                                        }
                                    }
                                });
                            }
                        });
                    }

                }, timeout != null ? timeout.Value : RequestTimeout);
            };

            EnqueueMessageOut(message);

            return subject.AsObservable();
        }

        /// <summary>
        /// Sends a response.
        /// </summary>
        /// <typeparam name="Response">The type of the response.</typeparam>
        /// <param name="response">The response.</param>
        /// <returns></returns>
        public Task SendResponse<Response>(TangoMessage<Response> response) where Response : IMessage<Response>
        {
            return SendResponse<Response>(response, response.Container.Token);
        }

        /// <summary>
        /// Sends a response for the specified token.
        /// </summary>
        /// <typeparam name="Response">The type of the response.</typeparam>
        /// <param name="response">The response.</param>
        /// <param name="token">The token.</param>
        /// <param name="completed">The completed.</param>
        /// <param name="errorCode">The error code.</param>
        /// <param name="errorMessage">The error message.</param>
        /// <returns></returns>
        /// <exception cref="InvalidOperationException">Matching request token was not found!</exception>
        public Task SendResponse<Response>(TangoMessage<Response> response, String token, bool? completed = null, ErrorCode? errorCode = null, String errorMessage = null) where Response : IMessage<Response>
        {
            if (_pushThread == null || _pushThread.ThreadState == ThreadState.Aborted)
            {
                throw new InvalidOperationException("Transporter push thread is not in a running state.");
            }

            response.Container.Token = token;

            if (completed.HasValue)
            {
                response.Container.Completed = completed.Value;
            }

            if (errorCode.HasValue)
            {
                response.Container.Error = errorCode.Value;
            }

            if (!String.IsNullOrEmpty(errorMessage))
            {
                response.Container.ErrorMessage = errorMessage;
            }

            LogManager.Log("Queuing response message: " + typeof(Response).Name, LogCategory.Debug);

            if (State != TransportComponentState.Connected)
            {
                throw LogManager.Log(new InvalidOperationException($"Could not send the response while transporter state is {State}."));
            }

            PendingResponse pendingResponse = null;

            LogManager.Log("Searching for matching request token: " + token, LogCategory.Debug);

            if (_pendingResponses.TryGetValue(token, out pendingResponse))
            {
                LogManager.Log("Found matching request token: " + token, LogCategory.Debug);

                if (!pendingResponse.IsContinuous)
                {
                    LogManager.Log("Removing matching request token.", LogCategory.Debug);
                    _pendingResponses.Remove(token);
                }
                else if (response.Container.Completed)
                {
                    LogManager.Log("Response completed. Removing matching request token.", LogCategory.Debug);
                    _pendingResponses.Remove(token);
                }
            }
            else
            {
                //This should never happen.
                throw LogManager.Log(new InvalidOperationException("Matching request token was not found!"), LogCategory.Critical);
            }

            TaskCompletionSource<object> source = new TaskCompletionSource<object>();
            TransportMessage<object> message = new TransportMessage<object>(token, response, TransportMessageDirection.Response, () => Encoder.Encode(response), source);
            EnqueueMessageOut(message);
            return source.Task;
        }

        /// <summary>
        /// Sends a general error response agnostic to the type of request.
        /// </summary>
        /// <param name="exception">The exception.</param>
        /// <param name="token">The token.</param>
        /// <returns></returns>
        public Task SendErrorResponse(Exception exception, string token)
        {
            return SendResponse<ErrorResponse>(new ErrorResponse() { }, token, true, ErrorCode.GeneralError, exception.Message);
        }

        #endregion

        #region Private Methods

        /// <summary>
        /// Starts the pull and push threads.
        /// </summary>
        protected void StartThreads()
        {
            _pullThread = new Thread(PullThreadMethod);
            _pullThread.IsBackground = true;
            _pullThread.Start();

            _pushThread = new Thread(PushThreadMethod);
            _pushThread.IsBackground = true;
            _pushThread.Start();

            _keepAliveThread = new Thread(KeepAliveThreadMethod);
            _keepAliveThread.IsBackground = true;
            _keepAliveThread.Start();
        }

        /// <summary>
        /// Gets the container timeout or default.
        /// </summary>
        /// <param name="container">The container.</param>
        /// <returns></returns>
        private TimeSpan? GetContainerTimeoutOrDefault(MessageContainer container)
        {
            return container.Timeout > 0 ? TimeSpan.FromMilliseconds(container.Timeout) : default(TimeSpan?);
        }

        /// <summary>
        /// Enqueues the message and releases the push wait handle.
        /// </summary>
        /// <param name="message">The message.</param>
        private void EnqueueMessageOut(TransportMessageBase message)
        {
            _sendingQueue.BlockEnqueue(message);
        }

        /// <summary>
        /// Enqueues the message and releases the pull wait handle.
        /// </summary>
        /// <param name="data">The data.</param>
        private void EnqueueMessageIn(byte[] data)
        {
            _arrivedResponses.BlockEnqueue(data);
        }

        #endregion

        #region Push Thread

        /// <summary>
        /// Push thread loop.
        /// </summary>
        private void PushThreadMethod()
        {
            try
            {
                while (State == TransportComponentState.Connected)
                {
                    TransportMessageBase message = _sendingQueue.BlockDequeue();

                    try
                    {
                        if (Adapter.State == TransportComponentState.Connected)
                        {
                            if (message.Token.Length != MESSAGE_TOKEN_LENGTH)
                            {
                                message.SetException(LogManager.Log(new InvalidOperationException("Invalid message token length: " + message.Token)));
                                continue;
                            }

                            LogManager.Log("Sending message on adapter: " + Adapter.Address + "...", LogCategory.Debug, message.Message);

                            if (message.Direction == TransportMessageDirection.Request)
                            {
                                lock (_pendingRequests)
                                {
                                    _pendingRequests.Add(message);
                                }
                            }

                            Adapter.Write(message.Serialize());

                            message.ActivateTimeout?.Invoke();

                            LogManager.Log("Message sent on adapter: " + Adapter.Address + "...", LogCategory.Debug, message.Message);
                        }
                        else
                        {
                            if (message.Direction == TransportMessageDirection.Request)
                            {
                                message.SetException(LogManager.Log(new InvalidOperationException("Could not send message " + message.Message.GetType().Name + ". Adapter is disconnected.")));
                            }
                        }

                        if (message.Direction == TransportMessageDirection.Response)
                        {
                            message.SetResult(true, true);
                        }
                    }
                    catch (Exception ex)
                    {
                        message.SetException(ex);
                    }
                }
            }
            catch (ThreadAbortException)
            {
                LogManager.Log("Push thread has been aborted.");
            }
            catch (Exception ex)
            {
                OnFailed(ex);
            }
        }

        #endregion

        #region Pull Thread

        /// <summary>
        /// Pull thread loop.
        /// </summary>
        private void PullThreadMethod()
        {
            try
            {
                while (State == TransportComponentState.Connected)
                {
                    byte[] data = _arrivedResponses.BlockDequeue();

                    LogManager.Log("Message received on adapter: " + Adapter.Address, LogCategory.Debug);

                    LogManager.Log("Parsing message container...", LogCategory.Debug);
                    MessageContainer container = Encoder.DecodeContainer(data);

                    LogManager.Log("Message was identified as " + container.Type + ".", LogCategory.Debug);

                    if (container.Token.Length != MESSAGE_TOKEN_LENGTH)
                    {
                        LogManager.Log("Invalid message token length received: " + container.Token, LogCategory.Error);
                        continue;
                    }

                    LogManager.Log("Searching for pending request token: " + container.Token, LogCategory.Debug);

                    TransportMessageBase request = null;
                    lock (_pendingRequests)
                    {
                        request = _pendingRequests.ToList().SingleOrDefault(x => x.Token == container.Token);
                    }

                    if (request != null)
                    {
                        LogManager.Log("Found pending request: " + (request.Message.GetType().IsGenericType ? request.Message.GetType().GetGenericArguments()[0].Name : request.Message.GetType().Name), LogCategory.Debug);

                        if (!request.IsContinuous)
                        {
                            LogManager.Log("Pending request was identified as 'single response'. Removing pending request.", LogCategory.Debug);

                            _pendingRequests.Remove(request);

                            try
                            {
                                if (container.Error == ErrorCode.None)
                                {
                                    var message = Encoder.Decode(data);
                                    LogManager.Log("Parsing inner response message and setting pending request task result...", LogCategory.Debug, message);
                                    request.SetResult(message, true);
                                    LogManager.Log("Message enquirer released...", LogCategory.Debug);
                                }
                                else
                                {
                                    request.SetException(LogManager.Log(new ResponseErrorException(container), LogCategory.Warning));
                                }
                            }
                            catch (Exception ex)
                            {
                                request.SetException(LogManager.Log(ex, "Error parsing response message."));
                            }
                        }
                        else
                        {
                            LogManager.Log("Pending request was identified as 'continuous response'. keeping pending request.", LogCategory.Debug);

                            try
                            {
                                if (container.Error == ErrorCode.None)
                                {
                                    var message = Encoder.Decode(data);

                                    LogManager.Log("Parsing inner response message and invoking continuous response callback...", LogCategory.Debug, message);

                                    if (container.Completed)
                                    {
                                        LogManager.Log("Continuous sequence completed.", LogCategory.Debug);
                                        _pendingRequests.Remove(request);
                                    }
                                    request.SetResult(message, container.Completed);
                                }
                                else if (container.Error == ErrorCode.ContinuousResponseAborted)
                                {
                                    String m = "Continuous response " + container.Type + " has been aborted: " + container.Error.ToString();
                                    LogManager.Log(m, LogCategory.Info);
                                    _pendingRequests.Remove(request);
                                    request.SetException(new ContinuousResponseAbortedException(m));
                                }
                                else
                                {
                                    LogManager.Log("Continuous response has returned with error: " + container.Error.ToString(), LogCategory.Warning);
                                    _pendingRequests.Remove(request);
                                    request.SetException(new ResponseErrorException(container));
                                }
                            }
                            catch (Exception ex)
                            {
                                LogManager.Log(ex, "Error parsing response message.");
                            }
                        }

                        try
                        {
                            Task.Factory.StartNew(() => OnResponseReceived(container));
                        }
                        catch
                        {
                            //Ignore any exceptions that may raise on the client side..
                        }
                    }
                    else
                    {
                        if (container.Type.ToString().EndsWith("Response"))
                        {
                            LogManager.Log(String.Format("A response message with no awaiting request was identified. {0}, Token: {1}. Message ignored.", container.Type, container.Token), LogCategory.Warning);
                            continue;
                        }

                        LogManager.Log("Message was identified as a new request message: " + container.Type.ToString(), LogCategory.Debug);

                        try
                        {
                            LogManager.Log("Saving request token: " + container.Token, LogCategory.Debug);
                            _pendingResponses.Add(container.Token, new PendingResponse(container.Continuous));

                            if (container.Type == MessageType.KeepAliveRequest && EnableKeepAliveAutoResponse)
                            {
                                LogManager.Log("Submitting keep alive response...", LogCategory.Debug);
                                SendResponse<KeepAliveResponse>(new KeepAliveResponse(), container.Token);
                            }
                            else
                            {
                                LogManager.Log("Invoking RequestReceived event...", LogCategory.Debug, container);

                                try
                                {
                                    Task.Factory.StartNew(() => OnRequestReceived(container));
                                }
                                catch
                                {
                                    //Ignore any exceptions that may raise on the client side..
                                }
                            }
                        }
                        catch (Exception ex)
                        {
                            LogManager.Log(ex);
                        }
                    }
                }
            }
            catch (ThreadAbortException)
            {
                LogManager.Log("Pull thread has been aborted.");
            }
            catch (Exception ex)
            {
                OnFailed(ex);
            }
        }

        #endregion

        #region Keep Alive Thread

        /// <summary>
        /// Responsible for sending keep alive messages.
        /// </summary>
        private void KeepAliveThreadMethod()
        {
            bool aborted = false;

            int retryCounter = KeepAliveRetries;

            try
            {
                Thread.Sleep(2000);

                while (State == TransportComponentState.Connected)
                {
                    try
                    {
                        Thread.Sleep(2000);

                        if (UseKeepAlive)
                        {
                            if (_arrivedResponses.Count == 0)
                            {
                                retryCounter--;
                                var response = SendRequest<KeepAliveRequest, KeepAliveResponse>(new KeepAliveRequest(), KeepAliveTimeout).Result;
                                retryCounter = KeepAliveRetries;
                            }
                            else
                            {
                                LogManager.Log("Keep alive request was skipped due to busy response queue.", LogCategory.Debug);
                            }
                        }
                    }
                    catch (Exception ex) when (ex is TimeoutException || ex is AggregateException)
                    {
                        if (State != TransportComponentState.Connected || aborted || (Adapter.State == TransportComponentState.Failed && FailsWithAdapter)) return;

                        if (UseKeepAlive)
                        {
                            if (DateTime.Now - _lastKeepAliveTime > KeepAliveTimeout)
                            {
                                if (retryCounter == 0)
                                {
                                    OnFailed(new KeepAliveException("The transporter has not received a KeepAlive response within the given time."));
                                    return;
                                }
                                else
                                {
                                    LogManager.Log($"The transporter has not received a KeepAlive response within the given time. Retrying ({retryCounter}/{KeepAliveRetries})...", LogCategory.Warning);
                                }
                            }
                            else
                            {
                                retryCounter = KeepAliveRetries;
                                LogManager.Log("The transporter has not received a KeepAlive response within the given time, but was rescued due to other message received within the given time.", LogCategory.Warning);
                            }
                        }
                    }
                    catch (ThreadAbortException)
                    {
                        aborted = true;
                        LogManager.Log("KeepAlive thread has been aborted.");
                        return;
                    }
                    catch (Exception ex)
                    {
                        if (State != TransportComponentState.Connected || aborted) return;

                        if (UseKeepAlive)
                        {
                            OnFailed(ex);
                            return;
                        }
                    }
                }
            }
            catch (ThreadAbortException)
            {
                LogManager.Log("KeepAlive thread has been aborted.");
            }
        }

        #endregion

        #region Override Methods

        /// <summary>
        /// Returns a <see cref="System.String" /> that represents this instance.
        /// </summary>
        /// <returns>
        /// A <see cref="System.String" /> that represents this instance.
        /// </returns>
        public override string ToString()
        {
            return this.GetType().Name + ", Adapter: " + (Adapter != null ? Adapter.ToString() : "Null") + ", Encoder: " + (Encoder != null ? Encoder.ToString() : "Null");
        }

        #endregion

        #region Dispose

        /// <summary>
        /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
        /// </summary>
        public void Dispose()
        {
            Disconnect().Wait();
            State = TransportComponentState.Disposed;
        }

        #endregion
    }
}