git_bsmd/AIS/bsmd.AIS2Service/SerialTCPReader.cs

using log4net;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Net.Sockets;
using System.Text;
using System.Threading;

namespace bsmd.AIS2Service
{
    internal class SerialTCPReader : IAISThread
    {
        private readonly string _host;
        private readonly uint _port;
        private readonly ConcurrentQueue<string> _inputQueue;
        private bool _stopFlag = false;

        private TcpClient tcpSocket;
        private Thread _thread;

        private const int sleepRetry = 60000;
        private const int maxRetries = 3;

        private static readonly ILog _log = LogManager.GetLogger(typeof(SerialTCPReader));


        public SerialTCPReader(string host, uint port, ConcurrentQueue<string> inputQueue)
        {
            _host = host; _port = port; _inputQueue = inputQueue;
        }

        private void ReadData()
        {

            while (!_stopFlag)
            {
                if (this.tcpSocket == null || !this.tcpSocket.Connected)
                {
                    if (!this.Connect())
                    {
                        _log.Error("Cannot connect to datasource, stopping service");
                        this.FatalErrorOccurred?.Invoke(this, new EventArgs());
                    }
                }

                try
                {
                    foreach (string line in ReadLines(this.tcpSocket.GetStream(), Encoding.ASCII))
                    {
                        _inputQueue.Enqueue(line);
                        if (_stopFlag) return;
                    }
                }
                catch (Exception ex)
                {
                    _log.WarnFormat("Exception occurred on serial reader: {0}", ex.Message);
                }
                finally
                {
                    this.tcpSocket?.Close();
                }
            }
        }

        private bool Connect()
        {
            int retryCounter = 0;
            int retryTime = sleepRetry;
            while (retryCounter < maxRetries)
            {
                try
                {
                    this.tcpSocket = new TcpClient(_host, (int)_port);
                    _log.InfoFormat("TCP stream connected ({0}:{1})", _host, _port);
                    return true;
                }
                catch (Exception ex)
                {
                    retryCounter++;
                    retryTime *= 2;
                    _log.WarnFormat("connect failed: {0}, retrying {1}", ex.Message, retryCounter);
                    Thread.Sleep(retryTime);
                }
            }
            return false;
        }

        private static IEnumerable<string> ReadLines(Stream source, Encoding encoding)
        {
            using(StreamReader reader = new StreamReader(source, encoding))
            {
                string line;
                
                while((line = reader.ReadLine()) != null)
                {
                    yield return line;
                }
                _log.InfoFormat("passed behind yield and closing StreamReader");
            }
        }

        #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 "Serial stream reader"; }
        }

        #endregion

    }
}