common/Hardware/Common.Hardware.SIRT/SIRTBroadcaster.cs

namespace Common.Hardware.SIRT
{
    using Common.Hardware.SIRT.Tasks;

    using System;
    using System.Collections.Concurrent;
    using System.Threading;
    using System.Threading.Tasks;

    public class SIRTBroadcaster
    {
        private readonly SIRTChanel[] chanels;
        private readonly ConcurrentDictionary<uint, SIRTChanel> chanelsMap;
        private readonly ConcurrentDictionary<uint, SIRTTask> tasks;
        private readonly SIRTStream sirtStream;
        private readonly bool log_RX_TX;

        private CancellationTokenSource cancellationTokenSource;
        private CancellationToken cancellationToken;

        public SIRTBroadcaster(string portName, bool log_RX_TX = true)
        {
            this.chanels = new SIRTChanel[]
            {
                new SIRTChanel(),
                new SIRTChanel(),
                new SIRTChanel(),
                new SIRTChanel(),
                new SIRTChanel(),
            };

            this.tasks = new ConcurrentDictionary<uint, SIRTTask>();
            this.sirtStream = new SIRTStream(portName, false);
            this.log_RX_TX = log_RX_TX;
            this.Id = "00-00-00-00-00-00-00-00";
        }

        public event Action<SIRTBroadcaster> Connected;
        public event Action<SIRTBroadcaster> Disconnected;
        public event Action<SIRTMessage> MessageReceived;

        public string Id { get; private set; }

        public int Frequency { get; private set; }

        public bool IsOpen => this.sirtStream.IsOpen;

        public string PortName => this.sirtStream.PortName;

        public void Close()
        {
            this.sirtStream.Close();

            try
            {
                this.tasks.Clear();
            }
            catch (Exception e)
            {
                this.LogMessage(e);
            }

            try
            {
                this.cancellationTokenSource.Cancel();
            }
            catch (Exception e)
            {
                this.LogMessage(e);
            }

            foreach (var chanel in this.chanels)
            {
                chanel.Captured -= this.Chanel_Captured;
                chanel.Released -= this.Chanel_Released;
            }

            this.Disconnected?.Invoke(this);
        }

        public void Open()
        {
            this.sirtStream.Open();

            if (this.sirtStream.IsOpen)
            {
                this.cancellationTokenSource = new CancellationTokenSource();
                this.cancellationToken = this.cancellationTokenSource.Token;
                _ = Task.Factory
                    .StartNew(this.ReceiveAsync, this.cancellationToken)
                    .ContinueWith(this.LogContinuationError);

                this.LogMessage("Activating SIRT ...");

                if (this.TryActivate(out var id, out var frequency))
                {
                    this.Id = id;
                    this.Frequency = frequency;

                    this.Connected?.Invoke(this);
                    this.LogMessage("Activated");

                    foreach (var chanel in this.chanels)
                    {
                        chanel.Captured += this.Chanel_Captured;
                        chanel.Released += this.Chanel_Released;
                    }
                }
                else
                {
                    this.Close();
                    this.LogMessage("Activation failed!");
                }
            }
        }

        /// <summary>
        /// Checks if the given task can be sent then replaces existing SIRTTask (if any) and starts the new one.
        /// </summary>
        public bool Start(SIRTTask task)
        {
            if (task.Frequency == this.Frequency)
            {
                task.SirtId = this.Id;
                task = this.tasks.AddOrUpdate(task.RequestAddress, addr => task, (addr, oldTask) => task);
                task = this.tasks.AddOrUpdate(task.ResponseAddress, addr => task, (addr, oldTask) => task);
                var bytes = task.Start();

                this.sirtStream.Write(bytes);

                if (this.log_RX_TX)
                {
                    this.LogMessage(task);
                }

                return true;
            }

            return false;
        }

        private void Chanel_Released(SIRTChanel chanel) 
            => this.chanelsMap.TryRemove(chanel.Address, out var _);

        private void Chanel_Captured(SIRTChanel chanel)
        {
            this.chanelsMap.TryRemove(chanel.Address, out var _);
            this.chanelsMap.TryAdd(chanel.Address, chanel);
        }

        private void LogContinuationError(Task task)
        {
            if (task?.Exception != null)
            {
                this.LogMessage(task.Exception);
            }
        }

        private void ReceiveAsync()
        {
            this.LogMessage($"Message receiving started!");

            var cancellationToken = this.cancellationToken;

            foreach (SIRTMessage message in this.sirtStream.ReadMessages())
            {
                message.Timestamp = DateTimeOffset.UtcNow;
                message.Frequency = this.Frequency;
                message.SirtId = this.Id;

                //if (this.chanelsMap.TryGetValue(message.Address, out var chanel))
                //{
                //    chanel.Receive(message);
                //}

                this.MessageReceived?.Invoke(message);

                if (this.tasks.TryGetValue(message.Address, out var task))
                {
                    task.Receive(message);

                    if (task.State == SIRTTaskState.Completed || task.State == SIRTTaskState.Cancelled)
                    {
                        this.tasks.TryRemove(task.RequestAddress, out _);
                        this.tasks.TryRemove(task.ResponseAddress, out _);
                    }
                }

                if (this.log_RX_TX)
                {
                    this.LogMessage(message);
                }

                cancellationToken.ThrowIfCancellationRequested();
            }

            this.LogMessage($"Message receiving stopped!");
        }

        private void LogMessage(object message)
            => SIRTLogger.LogMessage($"{this.PortName}|{nameof(SIRTBroadcaster)}|{message}");

        public bool TryCaptureChanel(uint address, out SIRTChanel chanel)
        {
            chanel = null;

            if (this.chanelsMap.TryGetValue(address, out chanel))
            {
                return true;
            }

            foreach (var _chanel in this.chanels)
            {
                if (_chanel.IsFree)
                {
                    _chanel.Capture(address);

                    this.chanelsMap.AddOrUpdate(address, _chanel, (_1, _2) => _chanel);

                    chanel = _chanel;

                    break;
                }
            }

            return chanel != null;
        }
    }
}