using System; using System.Collections.Concurrent; using System.Threading; using Xylem.Common.CommonCore.ThreadWatcher; using Xylem.Common.Hardware.Interfaces.Ports.PortCore.EventArguments; using Xylem.Common.Hardware.Interfaces.Protocols.ProtocolCore.EventArguments; using Xylem.Common.Hardware.Interfaces.Protocols.TransmitProtocol; namespace Xylem.Common.Hardware.Interfaces.Protocols.ProtocolCore { /// /// /// Base for all protocol types /// public abstract class BaseProtocol : IProtocol { private readonly CancellationTokenSource _decodingToken = new CancellationTokenSource(); /// public abstract event EventHandler OnRecordIsDecoded; /// public virtual event EventHandler OnRecordReadyToSend; private readonly ConcurrentQueue _decodingFifo = new ConcurrentQueue(); //initially do not signal event private readonly AutoResetEvent _onSyncDecodingThread = new AutoResetEvent(false); private readonly Thread _decodingThread; /// /// Ident is a combined string of Slot, Port, Protocol and Type /// protected readonly String Ident; private ITransmitProtocol _transmitProtocol; /// /// Starting decoding thread /// /// /// - Initial /// protected BaseProtocol(String ident) { Ident = ident; //assign thread to loop _decodingThread = new Thread(DecodingThreadLoop) { Name = $"{Ident} Decoding thread" }; //start DecodingThread ThreadWatcher.Instance.Start(_decodingThread); } /// /// /// Kill the decoding thread /// /// /// - Initial /// /// /// - try catch block. /// public virtual void Dispose() { try { //Cancel receive tokens _decodingToken.Cancel(); //Run thread again to notice CancellationToken has changed _onSyncDecodingThread.Set(); //this timeout counter is being used for dispose only var timeoutCounter = 100; while (_decodingThread.ThreadState != ThreadState.Stopped && timeoutCounter > 0) { Thread.Sleep(1); timeoutCounter -= 1; //Run thread again to notice CancellationToken has changed _onSyncDecodingThread.Set(); } if (_decodingThread.ThreadState != ThreadState.Stopped) { //if thread still running, he stuck so try to abort // try to avoid abort (takes age to run and is not safe) _decodingThread.Abort(); } } catch (Exception ex) { throw new ApplicationException(ex.Message); } } /// /// Fill FIFO with received data /// /// /// - Initial /// public void FillDecodingBuffer(IPortDataEventArgs data) { if (_decodingToken.IsCancellationRequested) return; //put data to FIFO _decodingFifo.Enqueue(data); //put DecodingThread state from WaitSleepJoin to Running _onSyncDecodingThread.Set(); } /// /// Decoding thread loop calling the individual decoding /// /// /// - Initial /// private void DecodingThreadLoop() { try { while (!_decodingToken.IsCancellationRequested) { while (_decodingFifo.TryDequeue(out var receivedRecord)) { DecodeRecord(receivedRecord); } //put DecodingThread to WaitSleepJoin until next data arrived _onSyncDecodingThread.WaitOne(); } } catch (ThreadAbortException) { } } /// /// The decoding routine /// /// protected abstract void DecodeRecord(IPortDataEventArgs data); /// public ITransmitProtocol GetTransmitProtocol() { return _transmitProtocol; } /// public void SetTransmitProtocol(ITransmitProtocol transmitProtocol) { _transmitProtocol = transmitProtocol; } } }