path: root/Software/Visual_Studio/Tango.Transport/TransporterBase.cs

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;

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

        #region Events

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

        /// <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 { _adapter = value; OnAdapterChanged(value); }
        }

        /// <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
            {
                _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;

                if (UseKeepAlive)
                {
                    LogManager.Log("KeepAlive is now working...");
                }
                else
                {
                    LogManager.Log("KeepAlive is now shutdown.");
                }
            }
        }

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


        #endregion

        #region Virtual Methods

        /// <summary>
        /// Called when the <see cref="Adapter"/> has changed.
        /// </summary>
        /// <param name="adapter">The adapter.</param>
        protected async virtual void OnAdapterChanged(ITransportAdapter adapter)
        {
            LogManager.Log("Adapter Changed:");

            if (adapter != null)
            {
                LogManager.Log(adapter.GetType().Name + ", " + adapter.Address + ", " + adapter.State.ToString());

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

                if (State == TransportComponentState.Connected && adapter.State == TransportComponentState.Disconnected)
                {
                    await adapter.Connect();
                }
            }
        }

        /// <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)
        {
            _arrivedResponses.Enqueue(data);
        }

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

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

        /// <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 ConcurrentQueue<TransportMessageBase>();
            _pendingRequests = new List<TransportMessageBase>();
            _arrivedResponses = new ConcurrentQueue<byte[]>();
            RequestTimeout = TimeSpan.FromSeconds(5);
        }

        /// <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>
        /// 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;
            if (Adapter != null)
            {
                await Adapter.Disconnect();
            }
            LogManager.Log("Transporter Disconnected...");
        }

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

            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);
            _sendingQueue.Enqueue(message);
            Task.Delay(timeout != null ? timeout.Value : RequestTimeout).ContinueWith((x) =>
            {
                if (!source.Task.IsCompleted)
                {
                    LogManager.Log(new TimeoutException("Request message: " + typeof(Request).Name + " had timed out after " + (timeout != null ? timeout.Value.TotalSeconds : RequestTimeout.TotalSeconds) + " seconds."));
                    LogManager.Log("Setting request task exception...", LogCategory.Debug);
                    source.SetException(new TimeoutException());
                }
            });
            return source.Task;
        }

        public IObservable<TangoMessage<Response>> SendContinuousRequest<Request, Response>(TangoMessage<Request> request, TimeSpan? timeout = default(TimeSpan?)) 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);

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

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

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

            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>
        /// <returns></returns>
        public Task SendResponse<Response>(TangoMessage<Response> response, String token, bool? completed = null, ErrorCode? errorCode = null) where Response : IMessage<Response>
        {
            response.Container.Token = token;

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

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

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

            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);
            _sendingQueue.Enqueue(message);
            return source.Task;
        }

        #endregion

        #region Private Methods

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

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

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

        #endregion

        #region Push Thread

        /// <summary>
        /// Push thread loop.
        /// </summary>
        private void PushThreadMethod()
        {
            try
            {
                while (State == TransportComponentState.Connected)
                {
                    while (Adapter != null && _sendingQueue.Count > 0)
                    {
                        TransportMessageBase message;
                        if (_sendingQueue.TryDequeue(out message))
                        {
                            try
                            {
                                if (Adapter.State == TransportComponentState.Connected)
                                {
                                    Adapter.Write(message.Serialize());
                                    LogManager.Log("Message sent on adapter: " + Adapter.Address + "...", LogCategory.Debug);
                                }
                                else
                                {
                                    LogManager.Log("Could not send message " + message.Message.GetType().Name + ". Adapter is disconnected.", LogCategory.Warning);
                                }

                                if (message.Direction == TransportMessageDirection.Request)
                                {
                                    _pendingRequests.Add(message);
                                }
                                else
                                {
                                    message.SetResult(true, true);
                                }
                            }
                            catch (Exception ex)
                            {
                                message.SetException(ex);
                            }
                        }
                    }

                    Thread.Sleep(10);

                }
            }
            catch (Exception ex)
            {
                OnFailed(ex);
            }
        }

        #endregion

        #region Pull Thread

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

                    while (Adapter != null && _arrivedResponses.Count > 0)
                    {
                        if (_arrivedResponses.TryDequeue(out data))
                        {
                            LogManager.Log("Message received on adapter: " + Adapter.Address, LogCategory.Debug);

                            LogManager.Log("Parsing message container...", LogCategory.Debug);
                            MessageContainer container = Encoder.DecodeContainer(data);
                            LogManager.Log("Searching for pending request token: " + container.Token, LogCategory.Debug);
                            TransportMessageBase request = _pendingRequests.SingleOrDefault(x => x.Token == container.Token);

                            if (request != null)
                            {
                                LogManager.Log("Found pending request: " + request.Message.GetType().GetGenericArguments()[0].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)
                                        {
                                            LogManager.Log("Parsing inner response message and setting pending request task result...", LogCategory.Debug);
                                            request.SetResult(Encoder.Decode(data), true);
                                            LogManager.Log("Message enquirer released...", LogCategory.Debug);
                                        }
                                        else
                                        {
                                            LogManager.Log("Response has returned with error: " + container.Error.ToString(), LogCategory.Warning);
                                            request.SetException(new ResponseErrorException(container.Error, container.Type));
                                        }
                                    }
                                    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)
                                        {
                                            LogManager.Log("Parsing inner response message and invoking continuous response callback...", LogCategory.Debug);
                                            if (container.Completed)
                                            {
                                                LogManager.Log("Continuous sequence completed.", LogCategory.Debug);
                                            }
                                            request.SetResult(Encoder.Decode(data), container.Completed);
                                        }
                                        else
                                        {
                                            LogManager.Log("Response has returned with error: " + container.Error.ToString(), LogCategory.Warning);
                                            request.SetException(new ResponseErrorException(container.Error, container.Type));
                                        }
                                    }
                                    catch (Exception ex)
                                    {
                                        LogManager.Log(ex, "Error parsing response message.");
                                    }
                                }
                            }
                            else
                            {
                                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)
                                    {
                                        LogManager.Log("Submitting keep alive response...", LogCategory.Debug);
                                        SendResponse<KeepAliveResponse>(new KeepAliveResponse(), container.Token);
                                    }
                                    else
                                    {
                                        LogManager.Log("Invoking RequestReceived event...", LogCategory.Debug);

                                        try
                                        {
                                            Task.Factory.StartNew(() => OnRequestReceived(container));
                                        }
                                        catch
                                        {
                                            //Ignore any exceptions that may raise on the client side..
                                        }
                                    }
                                }
                                catch (Exception ex)
                                {
                                    LogManager.Log(ex);
                                }
                            }
                        }
                    }

                    Thread.Sleep(10);
                }
            }
            catch (Exception ex)
            {
                OnFailed(ex);
            }
        }

        #endregion

        #region Keep Alive Thread

        /// <summary>
        /// Responsible for sending keep alive messages.
        /// </summary>
        private void KeepAliveThreadMethod()
        {
            while (State == TransportComponentState.Connected)
            {
                try
                {
                    Thread.Sleep(5000);

                    if (UseKeepAlive)
                    {
                        var task = SendRequest<KeepAliveRequest, KeepAliveResponse>(new KeepAliveRequest(), TimeSpan.FromSeconds(5));
                        task.Wait();
                        var response = task.Result;

                    }
                }
                catch (TimeoutException)
                {
                    if (State != TransportComponentState.Connected) return;

                    if (UseKeepAlive)
                    {
                        OnFailed(new TimeoutException("The transporter has not received a KeepAlive response within the given time."));
                        return;
                    }
                }
                catch (Exception ex)
                {
                    if (State != TransportComponentState.Connected) return;

                    if (UseKeepAlive)
                    {
                        OnFailed(ex);
                        return;
                    }
                }

                if (State != TransportComponentState.Connected) return;

                if (State != TransportComponentState.Connected) return;
            }
        }

        #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
    }
}