git_bsmd/AIS/bsmd.AIS2Service/AISDecoder.cs

using log4net;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Text;
using System.Threading;

namespace bsmd.AIS2Service
{
    internal class AISDecoder : IAISThread
    {

        #region fields

        private readonly ConcurrentQueue<string> _inputLines;
        private readonly ConcurrentQueue<AISClass> _outputAISClasses;
        private Thread _thread;
        private readonly Dictionary<string, List<AISQueueElement>> fragmentDict = new Dictionary<string, List<AISQueueElement>>();
        private const int sleepMS = 250;
        private bool _stopFlag = false;

        private static readonly ILog _log = LogManager.GetLogger(typeof(AISDecoder));
        #endregion

        #region construction

        public AISDecoder(ConcurrentQueue<string> input, ConcurrentQueue<AISClass> output)
        {
            _inputLines = input;
            _outputAISClasses = output;
        }

        #endregion

        #region class AISQueueElement

        public class AISQueueElement
        {
            public int seq_nr;
            public int total_nr;
            public string id;
            public string data;         
        }

        #endregion

        #region private methods

        /// <summary>
        /// Thread entry
        /// </summary>
        private void ReadData()
        {
            NMEA.Status status = NMEA.Status.OK;
            try
            {
                while (!_stopFlag)
                {
                    if (_inputLines.TryDequeue(out string line))
                    {
                        NMEA decodedSentence = NMEA.Decode(line, ref status);
                        if(decodedSentence != null)
                        {
                            if(decodedSentence is NMEA_AIS_Sentence aisSentence)
                            {
                                if(aisSentence.Total_Sentence_Nr == 1)
                                {
                                    DecodeData(aisSentence.AIS_Message);   
                                } 
                                else
                                {
                                    if(aisSentence.Seq_Message_Ident.Length == 0)
                                    {
                                        _log.WarnFormat("message sequence ident is empty, but we have multipart message. Ignoring message");
                                    }
                                    else
                                    {
                                        if(!fragmentDict.ContainsKey(aisSentence.Seq_Message_Ident))                                        
                                            fragmentDict[aisSentence.Seq_Message_Ident] = new List<AISQueueElement>();
                                        fragmentDict[aisSentence.Seq_Message_Ident].Add(new AISQueueElement { data = aisSentence.AIS_Message, id = aisSentence.Seq_Message_Ident, seq_nr = aisSentence.Msg_Sentence_Nr, total_nr = aisSentence.Total_Sentence_Nr });
                                        if((fragmentDict[aisSentence.Seq_Message_Ident].Count > 1) && FragmentsComplete(fragmentDict[aisSentence.Seq_Message_Ident]))
                                        {
                                            string concatData = ConcatenateFragments(fragmentDict[aisSentence.Seq_Message_Ident]);
                                            fragmentDict.Remove(aisSentence.Seq_Message_Ident);
                                            DecodeData(concatData);
                                        }
                                    }
                                }
                            }
                            else if(decodedSentence is NMEA_PNMLS_Sentence pnmlsSentence)
                            {
                                _log.Warn("cannot decode PNMLS sentence at this point");
                            }
                        }
                        else
                        {
                            _log.WarnFormat("NMEA decode failed with {0}", status);
                        }
                    }
                    else
                    {
                        Thread.Sleep(sleepMS);
                    }
                }
            }

            catch (Exception ex)
            {
                _log.ErrorFormat("Something bad has happened: {0}", ex.Message);
                this.FatalErrorOccurred?.Invoke(this, new EventArgs());
            }

        }

        private void DecodeData(string data)
        {
            AISClass.Status aisStatus = AISClass.Status.OK;
            AISClass decodedClass = AISClass.Decode(data, ref aisStatus);
            if(aisStatus == AISClass.Status.OK)
            {
                _outputAISClasses.Enqueue(decodedClass);
                _log.InfoFormat("Enqueuing AIS message for MMSI {0}", decodedClass.MMSI);
            }
            else
            {
                _log.WarnFormat("failed to decode AIS data: {0}", aisStatus);
            }
        }

        #endregion

        #region private helpers

        /// <summary>
        /// check to see if all fragments are available
        /// </summary>
        private static bool FragmentsComplete(List<AISQueueElement> elements)
        {
            if (elements == null || elements.Count == 0) return false;
            int num = elements[0].total_nr;

            for (int i = 1; i <= num; i++)
            {
                bool foundElements = false;
                for (int j = 0; j < elements.Count; j++)
                {
                    if (elements[j].seq_nr == i)
                        foundElements = true;
                }
                if (!foundElements) return false; // etwas fehlt noch
            }
            return true;
        }

        /// <summary>
        /// assembles message fragments. Care must be taken since fragments can appear
        /// out of order
        /// </summary>
        private static string ConcatenateFragments(List<AISQueueElement> elements)
        {
            if (elements == null || elements.Count == 0) return string.Empty;
            int num = elements[0].total_nr;
            StringBuilder sb = new StringBuilder();

            for (int i = 1; i <= num; i++)
            {
                for (int j = 0; j < elements.Count; j++)
                    if (elements[j].seq_nr == i)
                        sb.Append(elements[j].data);
            }
            return sb.ToString();
        }

        #endregion

        #region IAISThread implementation

        public event EventHandler FatalErrorOccurred;

        public void Start()
        {
            if (_thread != null) return; // may not run twice
            ThreadStart runReader = new ThreadStart(this.ReadData);
            _thread = new Thread(runReader);
            _thread.Start();
        }

        public void Stop()
        {
            if (_thread == null) return;
            _stopFlag = true;
            _thread.Join();
            _thread = null;
        }

        public string Name
        {
            get { return "AIS decoder"; }
        }

        #endregion

    }
}