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;
using System.IO;
using Tango.Core.ExtensionMethods;
using Tango.PMR.Integration;
using Newtonsoft.Json;
using System.Diagnostics;
using System.Reactive.Concurrency;

namespace Tango.Transport
{
    /// <summary>
    /// Represents an <see cref="ITransporter"/> base class.
    /// </summary>
    /// <seealso cref="Tango.Transport.ITransporter" />
    public abstract class TransporterBase : ExtendedObject, ITransporter
    {
        protected static long _component_counter = 1;

        private class RequestHandler
        {
            public Type RequestType { get; set; }
            public Action<ITransporter, Object, String> Callback { get; set; }
            public object RegisteredCallback { get; set; }
        }

        private const int MESSAGE_TOKEN_LENGTH = 36;
        private PriorityProducerConsumerQueue<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;
        private List<RequestHandler> _requestHandlers;
        private static JsonSerializerSettings _genericMessageSettings = new JsonSerializerSettings()
        {
            TypeNameHandling = TypeNameHandling.All,
        };

        #region Events

        /// <summary>
        /// Occurs when a new request message has been received.
        /// </summary>
        public event EventHandler<RequestReceivedEventArgs> 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;

        /// <summary>
        /// Occurs when a request has been sent.
        /// </summary>
        public event EventHandler<IMessage> RequestSent;

        /// <summary>
        /// Occurs when a request response has been received.
        /// </summary>
        public event EventHandler<IMessage> ResponseReceived;

        /// <summary>
        /// Occurs when a request has failed.
        /// </summary>
        public event EventHandler<RequestFailedEventArgs> RequestFailed;

        #endregion

        #region Properties

        /// <summary>
        /// Gets or sets the name of the transport component.
        /// </summary>
        public String ComponentName { get; set; }

        /// <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($"{GetExtendedComponentName()}: KeepAlive is activated...");
                }
                else
                {
                    LogManager.Log($"{GetExtendedComponentName()}: 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; }

        private GenericMessageProtocol _genericProtocol;
        /// <summary>
        /// Gets or sets the generic protocol used to serialize/deserialize generic messages.
        /// </summary>
        public GenericMessageProtocol GenericProtocol
        {
            get { return _genericProtocol; }
            set { _genericProtocol = value; RaisePropertyChangedAuto(); }
        }

        #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 PriorityProducerConsumerQueue<TransportMessageBase>();
            }

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

            if (newAdapter != null)
            {
                LogManager.Log($"{GetExtendedComponentName()}: Adapter Changed: Type = {newAdapter.GetType().Name}, Address = {newAdapter.Address}, State = {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($"{GetExtendedComponentName()}: 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 with exception '{Adapter.FailedStateException.Message}' and the transporter is configured to fail with the adapter."));
            }
        }

        /// <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 async void OnFailed(Exception ex)
        {
            if (State != TransportComponentState.Failed)
            {
                FailedStateException = ex;
                LogManager.Log(ex, $"{GetExtendedComponentName()}: Transporter failed.");
                State = TransportComponentState.Failed;

                await OnPostDisconnection();
            }
            else
            {
                LogManager.Log(ex, LogCategory.Warning, $"{GetExtendedComponentName()}: OnFailed called while state is already failed!");
            }
        }

        protected virtual async Task OnPostDisconnection()
        {
            try
            {
                if (_pullThread != null)
                {
                    _pullThread.Abort();
                    _pushThread.Abort();
                    _keepAliveThread.Abort();
                }
            }
            catch { }

            if (Adapter != null)
            {
                await Adapter.Disconnect();
            }

            NotifyContinuousRequestMessagesDisconnection();
        }

        /// <summary>
        /// Called when a new request has been received.
        /// </summary>
        /// <param name="container">The request.</param>
        protected virtual void OnRequestReceived(RequestReceivedEventArgs e)
        {
            var container = e.Container;

            if (_requestHandlers.Count > 0)
            {
                if (container.Type != MessageType.GenericRequest)
                {
                    var handlers = _requestHandlers.Where(x => x.RequestType.Name == container.Type.ToOriginalName()).ToList();

                    if (handlers.Count > 0) //Handle
                    {
                        e.Handled = true;

                        var request = MessageFactory.ExtractMessageFromContainer(container);

                        foreach (var handler in handlers)
                        {
                            try
                            {
                                handler.Callback.Invoke(this, request, container.Token);
                            }
                            catch
                            {
                                //Ignore any exception on the client side.
                            }
                        }
                    }
                }
                else
                {
                    var genericRequest = MessageFactory.ExtractMessageFromContainer<GenericRequest>(container);

                    var handlers = _requestHandlers.Where(x => x.RequestType.AssemblyQualifiedName == genericRequest.Type).ToList();

                    if (handlers.Count > 0)
                    {
                        e.Handled = true;

                        var innerRequest = GenericMessageSerializer.DeserializeFromByteString(handlers[0].RequestType, genericRequest.Data, GenericProtocol);

                        foreach (var handler in handlers)
                        {
                            try
                            {
                                handler.Callback.Invoke(this, innerRequest, container.Token);
                            }
                            catch
                            {
                                //Ignore any exception on the client side.
                            }
                        }
                    }
                }
            }

            RequestReceived?.Invoke(this, e);
        }

        /// <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);
        }

        /// <summary>
        /// Returns a string representing the component name. If an adapter is attached it will be part of the string.
        /// </summary>
        /// <returns></returns>
        protected virtual String GetExtendedComponentName()
        {
            return Adapter != null ? $"{ComponentName} ({Adapter.Address})" : ComponentName;
        }

        /// <summary>
        /// Notifies all the continuous request messages about disconnection.
        /// </summary>
        protected virtual void NotifyContinuousRequestMessagesDisconnection()
        {
            LogManager.Log("Notifying all continuous request messages about disconnection...");
            foreach (var request in _pendingRequests.ToList().Where(x => x.Direction == TransportMessageDirection.Request && x.IsContinuous))
            {
                try
                {
                    _pendingRequests.Remove(request);
                    LogManager.Log($"Notifying continuous request '{(request.Message as ITangoMessage).Type}'...");
                    OnRequestFailed(request, new TransporterDisconnectedException("Transporter disconnected."));
                    request.SetException(new TransporterDisconnectedException("Transporter disconnected."));
                }
                catch (Exception e)
                {
                    System.Diagnostics.Debug.WriteLine(e.ToString());
                }
            }
        }

        /// <summary>
        /// Called when the request has been sent
        /// </summary>
        /// <param name="response">The request.</param>
        private void OnRequestSent(TransportMessageBase request)
        {
            if (request.ShouldLog)
            {
                IMessage message = request.GetActualMessage() as IMessage;

                if (message != null)
                {
                    RequestSent?.Invoke(this, message);
                }
            }
        }

        /// <summary>
        /// Called when the response has been received
        /// </summary>
        /// <param name="response">The response.</param>
        private void OnResponseReceived(IMessage response)
        {
            if (response != null)
            {
                ResponseReceived?.Invoke(this, response);
            }
        }

        /// <summary>
        /// Called when the request has been failed
        /// </summary>
        /// <param name="request">The request.</param>
        private void OnRequestFailed(TransportMessageBase request, Exception exception)
        {
            IMessage message = request.GetActualMessage() as IMessage;

            if (message != null)
            {
                RequestFailed?.Invoke(this, new RequestFailedEventArgs(message, exception));
            }
        }

        #endregion

        #region Constructors

        /// <summary>
        /// Initializes a new instance of the <see cref="TransporterBase"/> class.
        /// </summary>
        public TransporterBase()
        {
            ComponentName = "Not Set";
            Encoder = new ProtoEncoder();
            _pendingResponses = new Dictionary<string, PendingResponse>();
            _sendingQueue = new PriorityProducerConsumerQueue<TransportMessageBase>();
            _pendingRequests = new ConcurrentList<TransportMessageBase>();
            _arrivedResponses = new ProducerConsumerQueue<byte[]>();
            _requestHandlers = new List<RequestHandler>();
            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 PriorityProducerConsumerQueue<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($"{GetExtendedComponentName()}: Transporter Connected...");
        }

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

                await OnPostDisconnection();

                LogManager.Log($"{GetExtendedComponentName()}: Transporter Disconnected...");
            }
        }

        #endregion

        #region Public Request Methods

        /// <summary>
        /// Sends a request.
        /// </summary>
        /// <param name="request">The request.</param>
        /// <param name="config">Request configuration.</param>
        /// <returns></returns>
        /// <exception cref="InvalidOperationException"></exception>
        public Task<IMessage> SendRequest(IMessage request, TransportRequestConfig config = null)
        {
            MessageContainer container = new MessageContainer();
            container.Token = Guid.NewGuid().ToString();
            container.Data = request.ToByteString();
            container.Timeout = config.Timeout.HasValue ? (UInt32)config.Timeout.Value.TotalMilliseconds : (UInt32)RequestTimeout.TotalMilliseconds;
            container.Type = MessageFactory.ParseMessageType(request.GetType().Name);

            return SendRequestInternal<IMessage>(container.Token, container, config).Task;
        }

        /// <summary>
        /// Sends the request.
        /// </summary>
        /// <param name="container"></param>
        /// <param name="config">Request configuration.</param>
        /// <returns></returns>
        /// <exception cref="InvalidOperationException"></exception>
        public Task<MessageContainer> SendRequest(MessageContainer container, TransportRequestConfig config = null)
        {
            return SendRequestInternal<MessageContainer>(container.Token, container, config).Task;
        }

        /// <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="config">Request configuration.</param>
        /// <returns></returns>
        /// <exception cref="InvalidOperationException"></exception>
        public Task<TangoMessage<Response>> SendRequest<Request, Response>(TangoMessage<Request> request, TransportRequestConfig config = null) where Request : IMessage<Request> where Response : IMessage<Response>
        {
            return SendRequestInternal<TangoMessage<Response>>(request.Container.Token, request, config).Task;
        }

        private TaskCompletionSource<T> SendRequestInternal<T>(String token, Object request, TransportRequestConfig config = null)
        {
            config = config ?? new TransportRequestConfig();

            TaskCompletionSource<T> source = new TaskCompletionSource<T>();

            String requestName = TransportMessageBase.GetActualMessageTypeName(request);

            Func<byte[]> toBytes = null;

            if (request is ITangoMessage tangoMessage)
            {
                toBytes = () => Encoder.Encode(tangoMessage);
            }
            else if (request is IMessage protoMessage)
            {
                toBytes = () => protoMessage.ToByteArray();
            }

            TransportMessage<T> message = new TransportMessage<T>(token, request, TransportMessageDirection.Request, toBytes, source);
            message.ShouldLog = config.ShouldLog;
            message.Immidiate = config.Immediate;
            message.Priority = config.Priority;
            message.TransportComponentName = GetExtendedComponentName();
            message.ThreadingMode = config.ThreadingMode;

            TimeSpan? timeout = config.Timeout;

            if (request is MessageContainer container)
            {
                timeout = GetContainerTimeoutOrDefault(container);
            }

            if (timeout == null)
            {
                timeout = RequestTimeout;
            }

            if (request is ITangoMessage tanMessage)
            {
                tanMessage.Container.Timeout = (uint)timeout.Value.TotalMilliseconds;
            }

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

            LogManager.Log($"{GetExtendedComponentName()}: Queuing request message: {requestName} Token: {token}", LogCategory.Debug);
            LogManager.Log($"{GetExtendedComponentName()}: Expected response: {responseName}", LogCategory.Debug);

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


            Action timeoutAction = () => 
            {
                if (!source.Task.IsCompleted)
                {
                    TimeoutException ex = new TimeoutException($"{GetExtendedComponentName()}: Request message '{requestName}' had timed out after {timeout.Value.TotalSeconds} seconds.");
                    OnRequestFailed(message, ex);
                    message.SetException(ex);
                }
            };

            if (config.ThreadingMode == TransportThreadingMode.NewThread)
            {
                message.ActivateTimeout = () =>
                {
                    TimeoutTask.StartNew(() =>
                    {
                        timeoutAction();
                    }, timeout.Value);
                };
            }
            else
            {
                message.ActivateTimeout = () =>
                {
                    Task.Delay(timeout.Value).ContinueWith((x) =>
                    {
                        timeoutAction();
                    });
                };
            }

            EnqueueMessageOut(message);

            return source;
        }

        /// <summary>
        /// Sends a request and expecting multiple response messages.
        /// </summary>
        /// <param name="request"></param>
        /// <param name="config">Request configuration.</param>
        /// <returns></returns>
        /// <exception cref="InvalidOperationException"></exception>
        public IObservable<IMessage> SendContinuousRequest(IMessage request, TransportContinuousRequestConfig config = null)
        {
            MessageContainer container = new MessageContainer();
            container.Token = Guid.NewGuid().ToString();
            container.Data = request.ToByteString();
            container.Type = MessageFactory.ParseMessageType(request.GetType().Name);

            //We need to assign this timeout because when the internal method detects a MessageContainer it will bypass the container continuous timeout assignment.
            container.ContinuousTimeout = config.ContinuousTimeout != null ? (UInt32)config.ContinuousTimeout.Value.TotalMilliseconds : 0;

            container.Continuous = true;

            return SendContinuousRequestInternal<IMessage>(container.Token, container, config);
        }

        /// <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"></param>
        /// <param name="config">Request configuration.</param>
        /// <returns></returns>
        /// <exception cref="InvalidOperationException"></exception>
        public IObservable<TangoMessage<Response>> SendContinuousRequest<Request, Response>(TangoMessage<Request> request, TransportContinuousRequestConfig config = null) where Request : IMessage<Request> where Response : IMessage<Response>
        {
            request.Container.Continuous = true;
            request.Container.Completed = false;
            return SendContinuousRequestInternal<TangoMessage<Response>>(request.Container.Token, request, config);
        }

        /// <summary>
        /// Sends a continuous request.
        /// </summary>
        /// <param name="container">The container.</param>
        /// <param name="config">Request configuration.</param>
        /// <returns></returns>
        /// <exception cref="InvalidOperationException"></exception>
        public IObservable<MessageContainer> SendContinuousRequest(MessageContainer container, TransportContinuousRequestConfig config = null)
        {
            return SendContinuousRequestInternal<MessageContainer>(container.Token, container, config);
        }

        private IObservable<T> SendContinuousRequestInternal<T>(String token, Object request, TransportContinuousRequestConfig config = null)
        {
            config = config ?? new TransportContinuousRequestConfig();

            String requestName = TransportMessageBase.GetActualMessageTypeName(request);

            Func<byte[]> toBytes = null;

            if (request is ITangoMessage tangoMessage)
            {
                toBytes = () => Encoder.Encode(tangoMessage);
            }
            else if (request is IMessage protoMessage)
            {
                toBytes = () => protoMessage.ToByteArray();
            }

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

            TransportMessage<T> message = new TransportMessage<T>(token, request, TransportMessageDirection.Request, toBytes, null);
            message.ShouldLog = config.ShouldLog;
            message.Immidiate = config.Immediate;
            message.Priority = config.Priority;
            message.TransportComponentName = GetExtendedComponentName();
            message.IsContinuous = true;
            message.ContinuesResponseSubject = subject;

            TimeSpan? timeout = config.Timeout;
            TimeSpan? continuousTimeout = config.ContinuousTimeout;

            if (request is MessageContainer container)
            {
                timeout = GetContainerTimeoutOrDefault(container);

                if (timeout == null)
                {
                    timeout = config.Timeout != null ? config.Timeout.Value : RequestTimeout;
                    container.Timeout = (uint)timeout.Value.TotalMilliseconds;
                }

                continuousTimeout = GetContainerContinuousTimeoutOrDefault(container);
            }

            if (timeout == null)
            {
                timeout = RequestTimeout;
            }

            if (request is ITangoMessage tanMessage)
            {
                tanMessage.Container.Continuous = true;
                tanMessage.Container.Completed = false;
                tanMessage.Container.Timeout = (uint)timeout.Value.TotalMilliseconds;
                tanMessage.Container.ContinuousTimeout = continuousTimeout != null ? (UInt32)continuousTimeout.Value.TotalMilliseconds : 0;
            }

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

            LogManager.Log($"{GetExtendedComponentName()}: Queuing continuous request message: {requestName} Token: {token}", LogCategory.Debug);
            LogManager.Log($"{GetExtendedComponentName()}: Expected response: {responseName}", LogCategory.Debug);

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

            Action timeoutAction = () => 
            {
                if (!message.AtLeastOneResponseReceived)
                {
                    TimeoutException ex = new TimeoutException($"{GetExtendedComponentName()}: Request message '{requestName}' had timed out after {timeout.Value.TotalSeconds} seconds.");
                    OnRequestFailed(message, ex);
                    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($"{GetExtendedComponentName()}: Continuous request message '{requestName}' had failed to provide a response for a period of {continuousTimeout.Value.TotalSeconds} seconds and has timed out.");
                                        OnRequestFailed(message, ex);
                                        message.SetException(ex);
                                        return;
                                    }
                                }
                            });
                        }
                    });
                }
            };

            if (config.ThreadingMode == TransportThreadingMode.NewThread)
            {
                message.ActivateTimeout = () =>
                {
                    TimeoutTask.StartNew(() =>
                    {
                        timeoutAction();
                    }, timeout.Value);
                };
            }
            else
            {
                message.ActivateTimeout = () =>
                {
                    Task.Delay(timeout.Value).ContinueWith((x) =>
                    {
                        timeoutAction();
                    });
                };
            }

            EnqueueMessageOut(message);

            return subject.AsObservable();
        }

        /// <summary>
        /// Sends a generic request of any type.
        /// </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="config">The configuration.</param>
        /// <returns></returns>
        public async Task<Response> SendGenericRequest<Request, Response>(Request request, TransportRequestConfig config = null) where Request : class where Response : class
        {
            GenericRequest genericRequest = new GenericRequest();
            genericRequest.Type = request.GetType().AssemblyQualifiedName;
            genericRequest.Data = GenericMessageSerializer.SerializeToByteString<Request>(request, GenericProtocol);

            var response = await SendRequest<GenericRequest, GenericResponse>(genericRequest, config);
            var responseObject = GenericMessageSerializer.DeserializeFromByteString<Response>(response.Message.Data, GenericProtocol);
            return responseObject;
        }

        /// <summary>
        /// Sends a generic request and expecting multiple generic response messages.
        /// </summary>
        /// <typeparam name="Request">The type of the request.</typeparam>
        /// <typeparam name="Response">The type of the response.</typeparam>
        /// <param name="config">Request configuration.</param>
        /// <returns></returns>
        public IObservable<Response> SendGenericContinuousRequest<Request, Response>(Request request, TransportContinuousRequestConfig config = null) where Request : class where Response : class
        {
            GenericRequest genericRequest = new GenericRequest();
            genericRequest.Type = request.GetType().AssemblyQualifiedName;

            genericRequest.Data = GenericMessageSerializer.SerializeToByteString<Request>(request, GenericProtocol);

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

            SendContinuousRequest<GenericRequest, GenericResponse>(genericRequest, config).Subscribe((response) =>
            {

                try
                {
                    var responseObject = GenericMessageSerializer.DeserializeFromByteString<Response>(response.Message.Data, GenericProtocol);
                    subject.OnNext(responseObject);
                }
                catch (Exception ex)
                {
                    Debugger.Break();
                    System.Diagnostics.Debug.WriteLine($"Exception thrown by the generic request continuous handler.\n{ex.ToString()}");
                    //Ignore exception at the client side.
                }

            }, (ex) =>
            {

                try
                {
                    subject.OnError(ex);
                }
                catch (Exception xx)
                {
                    Debugger.Break();
                    System.Diagnostics.Debug.WriteLine($"Exception thrown by the generic request error handler.\n{xx.ToString()}");
                    //Ignore exception at the client side.
                }

            }, () =>
            {
                try
                {
                    subject.OnCompleted();
                }
                catch (Exception ex)
                {
                    Debugger.Break();
                    System.Diagnostics.Debug.WriteLine($"Exception thrown by the generic request completed handler.\n{ex.ToString()}");
                    //Ignore exception at the client side.
                }
            });

            return subject.AsObservable();
        }

        #endregion

        #region Public Response Methods

        /// <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 the response.
        /// </summary>
        /// <param name="response"></param>
        /// <param name="token">Request token.</param>
        /// <param name="config">Response configuration.</param>
        /// <returns></returns>
        public Task SendResponse(IMessage response, String token, TransportResponseConfig config = null)
        {
            config = config ?? new TransportResponseConfig();

            String responseName = response.GetType().Name;

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

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

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

            container.EventCode = (int)config.EventType;

            container.Completed = config.Completed;

            return SendResponse(container);
        }

        /// <summary>
        /// Sends the response.
        /// </summary>
        /// <param name="container">The container.</param>
        /// <returns></returns>
        /// <exception cref="InvalidOperationException">
        /// </exception>
        public Task SendResponse(MessageContainer container, TransportResponseConfig config = null)
        {
            config = config ?? new TransportResponseConfig();

            if (_pushThread == null || _pushThread.ThreadState == System.Threading.ThreadState.Aborted)
            {
                throw new InvalidOperationException("Transporter push thread is not in a running state.");
            }

            String token = container.Token;

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

            PendingResponse pendingResponse = null;

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

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

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

                if (!pendingResponse.IsContinuous)
                {
                    LogManager.Log($"{GetExtendedComponentName()}: Removing matching request token.", LogCategory.Debug);
                    _pendingResponses.Remove(token);
                }
                else if (container.Completed)
                {
                    LogManager.Log($"{GetExtendedComponentName()}: Response completed. Removing matching request token.", LogCategory.Debug);
                    _pendingResponses.Remove(token);
                }
            }
            else
            {
                //This should never happen.
                throw LogManager.Log(new InvalidOperationException($"{GetExtendedComponentName()}: 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);
            message.ShouldLog = config.ShouldLog;
            message.Immidiate = config.Immediate;
            message.Priority = config.Priority;
            EnqueueMessageOut(message);
            return source.Task;
        }

        /// <summary>
        /// Sends a response.
        /// </summary>
        /// <typeparam name="Response">The type of the response.</typeparam>
        /// <param name="response">The response.</param>
        /// <param name="token">Request token.</param>
        /// <param name="config">Response configuration.</param>
        /// <returns></returns>
        /// <exception cref="InvalidOperationException">
        /// Transporter push thread is not in a running state.
        /// </exception>
        public Task SendResponse<Response>(TangoMessage<Response> response, String token, TransportResponseConfig config = null) where Response : IMessage<Response>
        {
            config = config ?? new TransportResponseConfig();

            if (_pushThread == null || _pushThread.ThreadState == System.Threading.ThreadState.Aborted)
            {
                throw new InvalidOperationException("Transporter push thread is not in a running state.");
            }

            response.Container.Token = token;

            response.Container.Completed = config.Completed;

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

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

            response.Container.EventCode = (int)config.EventType;

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

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

            PendingResponse pendingResponse = null;

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

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

                if (!pendingResponse.IsContinuous)
                {
                    LogManager.Log($"{GetExtendedComponentName()}: Removing matching request token.", LogCategory.Debug);
                    _pendingResponses.Remove(token);
                }
                else if (response.Container.Completed)
                {
                    LogManager.Log($"{GetExtendedComponentName()}: Response completed. Removing matching request token.", LogCategory.Debug);
                    _pendingResponses.Remove(token);
                }
            }
            else
            {
                //This should never happen.
                throw LogManager.Log(new InvalidOperationException($"{GetExtendedComponentName()}: 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);
            message.ShouldLog = config.ShouldLog;
            message.Immidiate = config.Immediate;
            message.Priority = config.Priority;
            EnqueueMessageOut(message);
            return source.Task;
        }

        /// <summary>
        /// Sends a generic response.
        /// </summary>
        /// <typeparam name="Response">The type of the response.</typeparam>
        /// <param name="response">The response.</param>
        /// <param name="token">The request token.</param>
        /// <param name="config">The response configuration.</param>
        /// <returns></returns>
        public async Task SendGenericResponse<Response>(Response response, String token, TransportResponseConfig config = null) where Response : class
        {
            GenericResponse genericResponse = new GenericResponse();

            genericResponse.Type = response.GetType().AssemblyQualifiedName;
            genericResponse.Data = GenericMessageSerializer.SerializeToByteString<Response>(response, GenericProtocol);
            await SendResponse<GenericResponse>(genericResponse, token, config);
        }

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

        #endregion

        #region Request Handlers

        /// <summary>
        /// Registers a custom request handler.
        /// </summary>
        /// <typeparam name="Request">The type of the request.</typeparam>
        /// <param name="callback">The callback.</param>
        public void RegisterRequestHandler<Request>(RequestHandlerCallbackDelegate<Request> callback) where Request : class
        {
            RequestHandler handler = new RequestHandler();
            handler.RequestType = typeof(Request);
            handler.RegisteredCallback = callback;
            handler.Callback = (transporter, obj, token) =>
            {
                callback?.Invoke(transporter, obj as Request, token);
            };

            _requestHandlers.Add(handler);
        }

        /// <summary>
        /// Unregisters a custom request handler.
        /// </summary>
        /// <typeparam name="Request">The type of the request.</typeparam>
        /// <param name="callback">The callback.</param>
        public void UnregisterRequestHandler<Request>(RequestHandlerCallbackDelegate<Request> callback) where Request : class
        {
            var handler = _requestHandlers.FirstOrDefault(x => (x.RegisteredCallback as RequestHandlerCallbackDelegate<Request>) == callback);
            if (handler != null)
            {
                _requestHandlers.Remove(handler);
            }
        }

        /// <summary>
        /// Copies this instance request handlers to the specified instance.
        /// </summary>
        /// <param name="transporter">The transporter to copy the handlers to.</param>
        public void CopyRequestHandlers(ITransporter transporter)
        {
            foreach (var handler in _requestHandlers.ToList())
            {
                (transporter as TransporterBase)._requestHandlers.Add(handler);
                _requestHandlers.Remove(handler);
            }
        }

        #endregion

        #region Private Methods

        /// <summary>
        /// Starts the pull and push threads.
        /// </summary>
        protected void StartThreads()
        {
            try
            {
                _pullThread = new Thread(PullThreadMethod);
                _pullThread.Name = $"{GetExtendedComponentName()} Pull Thread";
                _pullThread.IsBackground = true;
                _pullThread.Start();

                _pushThread = new Thread(PushThreadMethod);
                _pushThread.Name = $"{GetExtendedComponentName()} Push Thread";
                _pushThread.IsBackground = true;
                _pushThread.Start();

                _keepAliveThread = new Thread(KeepAliveThreadMethod);
                _keepAliveThread.Name = $"{GetExtendedComponentName()} KeepAlive Thread";
                _keepAliveThread.IsBackground = true;
                _keepAliveThread.Start();
            }
            catch (Exception ex)
            {
                LogManager.Log(ex, "Error starting transporter threads.");
            }
        }

        /// <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>
        /// Gets the container timeout or default.
        /// </summary>
        /// <param name="container">The container.</param>
        /// <returns></returns>
        private TimeSpan? GetContainerContinuousTimeoutOrDefault(MessageContainer container)
        {
            return container.ContinuousTimeout > 0 ? TimeSpan.FromMilliseconds(container.ContinuousTimeout) : 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, message.Priority);
        }

        /// <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)
                            {
                                var ex = LogManager.Log(new InvalidOperationException($"{GetExtendedComponentName()}: Invalid message token length: " + message.Token));
                                OnRequestFailed(message, ex);
                                message.SetException(ex);
                                continue;
                            }

                            if (message.Direction == TransportMessageDirection.Request)
                            {
                                if (message.ShouldLog)
                                {
                                    LogManager.Log($"{GetExtendedComponentName()}: Sending request '{message.GetActualMessageTypeName()}'...\n{message.GetActualMessage().ToJsonString()}", LogCategory.Info);
                                    OnRequestSent(message);
                                }

                                lock (_pendingRequests)
                                {
                                    _pendingRequests.Add(message);
                                }
                            }
                            else
                            {
                                if (message.ShouldLog)
                                {
                                    LogManager.Log($"{GetExtendedComponentName()}: Sending response '{message.GetActualMessageTypeName()}'...\n{message.GetActualMessage().ToJsonString()}", LogCategory.Info);
                                }
                            }

                            Adapter.Write(message.Serialize(), message.Immidiate);

                            message.ActivateTimeout?.Invoke();

                            LogManager.Log($"{GetExtendedComponentName()}: Message sent...", LogCategory.Debug, message.Message);
                        }
                        else
                        {
                            if (message.Direction == TransportMessageDirection.Request)
                            {
                                var ex = new InvalidOperationException($"{GetExtendedComponentName()}: Could not send message " + message.GetActualMessageTypeName() + ". Adapter is disconnected.");
                                OnRequestFailed(message, ex);
                                message.SetException(ex);
                            }
                        }

                        if (message.Direction == TransportMessageDirection.Response)
                        {
                            message.SetResult(true, true);
                        }
                    }
                    catch (ThreadAbortException)
                    {
                        Exception requestException = FailedStateException != null ? FailedStateException : new TransporterDisconnectedException("The transporter push thread has been aborted.");
                        OnRequestFailed(message, requestException);
                        message.SetException(requestException);
                    }
                    catch (Exception ex)
                    {
                        OnRequestFailed(message, ex);
                        message.SetException(ex);
                    }
                }
            }
            catch (ThreadAbortException)
            {
                LogManager.Log($"{GetExtendedComponentName()}: 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($"{GetExtendedComponentName()}: Message received...", LogCategory.Debug);

                    LogManager.Log($"{GetExtendedComponentName()}: Parsing message container...", LogCategory.Debug);

                    MessageContainer container = null;

                    try
                    {
                        container = Encoder.DecodeContainer(data);
                    }
                    catch (Exception ex)
                    {
                        LogManager.Log(ex, $"{GetExtendedComponentName()}: Error parsing message container. Skipping incoming message...");
                        continue;
                    }

                    LogManager.Log($"{GetExtendedComponentName()}: Message was identified as " + container.Type + ".", LogCategory.Debug);

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

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

                    TransportMessageBase request = null;
                    lock (_pendingRequests)
                    {
                        try
                        {
                            var requests = _pendingRequests.ToList().Where(x => x.Token == container.Token).ToList();

                            try
                            {
                                if (requests.Count > 1)
                                {
                                    LogManager.Log($"{GetExtendedComponentName()}: {requests.Count} requests with the same token were detected - {requests.First().GetActualMessageTypeName()}.", LogCategory.Warning);
                                }
                            }
                            catch (Exception ex)
                            {
                                LogManager.Log(ex, LogCategory.Warning, "Something bad happened please investigate.");
                            }

                            request = requests.LastOrDefault();
                        }
                        catch (Exception ex)
                        {
                            LogManager.Log(ex, LogCategory.Warning, "Something bad happened please investigate.");
                        }
                    }

                    if (request != null)
                    {
                        LogManager.Log($"{GetExtendedComponentName()}: Found pending request: " + request.GetActualMessageTypeName(), LogCategory.Debug);

                        if (!request.IsContinuous)
                        {
                            LogManager.Log($"{GetExtendedComponentName()}: 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);

                                    if (request.ShouldLog)
                                    {
                                        try
                                        {
                                            String responseType = message.Type.ToString();
                                            var messageContent = message.GetType().GetProperty("Message").GetValue(message);

                                            try
                                            {
                                                if (messageContent.GetType() == typeof(GenericResponse))
                                                {
                                                    Type genericType = Type.GetType((messageContent as GenericResponse).Type);
                                                    responseType = genericType.Name;
                                                    messageContent = GenericMessageSerializer.DeserializeFromByteString(genericType, (messageContent as GenericResponse).Data, GenericProtocol);
                                                }
                                            }
                                            catch { }

                                            LogManager.Log($"{GetExtendedComponentName()}: Response received '{responseType}'...\n{messageContent.ToJsonString()}", LogCategory.Info);
                                            OnResponseReceived(messageContent as IMessage);
                                        }
                                        catch
                                        {
                                            LogManager.Log("Error logging response received.", LogCategory.Warning);
                                        }
                                    }

                                    LogManager.Log($"{GetExtendedComponentName()}: Parsing inner response message and setting pending request task result...", LogCategory.Debug, message);
                                    request.SetResult(message, true);
                                    LogManager.Log($"{GetExtendedComponentName()}: Message enquirer released...", LogCategory.Debug);
                                }
                                else
                                {
                                    var ex = new ResponseErrorException(container, request.GetActualMessageTypeName());
                                    OnRequestFailed(request, ex);
                                    request.SetException(ex);
                                }
                            }
                            catch (Exception ex)
                            {
                                OnRequestFailed(request, ex);
                                request.SetException(LogManager.Log(ex, $"{GetExtendedComponentName()}: Error parsing response message."));
                            }
                        }
                        else
                        {
                            LogManager.Log($"{GetExtendedComponentName()}: Pending request was identified as 'continuous response'. keeping pending request.", LogCategory.Debug);

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

                                    if (request.ShouldLog && !request.AtLeastOneResponseReceived)
                                    {
                                        try
                                        {
                                            var messageContent = message.GetType().GetProperty("Message").GetValue(message);
                                            LogManager.Log($"{GetExtendedComponentName()}: Response received '{message.Type}'...\n{messageContent.ToJsonString()}", LogCategory.Info);
                                            OnResponseReceived(messageContent as IMessage);
                                        }
                                        catch
                                        {
                                            LogManager.Log("Error logging response received.", LogCategory.Warning);
                                        }
                                    }

                                    LogManager.Log($"{GetExtendedComponentName()}: Parsing inner response message and invoking continuous response callback...", LogCategory.Debug, message);

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

                                    var exx = new ResponseErrorException(container, request.GetActualMessageTypeName());
                                    OnRequestFailed(request, exx);
                                    request.SetException(exx);
                                }
                            }
                            catch (Exception ex)
                            {
                                LogManager.Log(ex, $"{GetExtendedComponentName()}: 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($"{GetExtendedComponentName()}: A response message with no awaiting request was identified. {container.Type}, Token: {container.Token}. Message ignored.", LogCategory.Warning);
                            continue;
                        }

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

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

                            if (container.Type == MessageType.KeepAliveRequest && EnableKeepAliveAutoResponse)
                            {
                                LogManager.Log($"{GetExtendedComponentName()}: Submitting keep alive response...", LogCategory.Debug);
                                try
                                {
                                    SendResponse<KeepAliveResponse>(new KeepAliveResponse(), container.Token, new TransportResponseConfig()
                                    {
                                        Priority = QueuePriority.High
                                    });
                                }
                                catch { }
                            }
                            else
                            {
                                LogManager.Log($"{GetExtendedComponentName()}: Invoking RequestReceived event...", LogCategory.Debug, container);

                                try
                                {
                                    Task.Factory.StartNew(() => OnRequestReceived(new RequestReceivedEventArgs(container)));
                                }
                                catch
                                {
                                    //Ignore any exceptions that may raise on the client side..
                                }
                            }
                        }
                        catch (Exception ex)
                        {
                            LogManager.Log(ex);
                        }
                    }
                }
            }
            catch (ThreadAbortException)
            {
                LogManager.Log($"{GetExtendedComponentName()}: 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--;

                                if (State == TransportComponentState.Connected)
                                {
                                    var response = SendRequest<KeepAliveRequest, KeepAliveResponse>(new KeepAliveRequest(), new TransportRequestConfig()
                                    {
                                        Timeout = KeepAliveTimeout,
                                        Priority = QueuePriority.High
                                    }).Result;
                                    retryCounter = KeepAliveRetries;
                                }
                                else
                                {
                                    continue;
                                }
                            }
                            else
                            {
                                LogManager.Log($"{GetExtendedComponentName()}: 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($"{GetExtendedComponentName()}: 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($"{GetExtendedComponentName()}: KeepAlive thread has been aborted.");
                        return;
                    }
                    catch (Exception ex)
                    {
                        if (State != TransportComponentState.Connected || aborted) return;

                        if (UseKeepAlive)
                        {
                            OnFailed(ex);
                            return;
                        }
                    }
                }
            }
            catch (ThreadAbortException)
            {
                LogManager.Log($"{GetExtendedComponentName()}: 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
    }
}