prs/prs.server/Engines/GPS/GPSEngine.cs

using Comal.Classes;
using InABox.Clients;
using InABox.Core;
using InABox.DigitalMatter;
using InABox.IPC;
using netDxf.Tables;
using PRSServer.Engines;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading.Tasks;
using System.Timers;

namespace PRSServer
{
    public class GPSDeviceUpdate : ISerializeBinary
    {
        public string AuditTrail { get; set; }

        public GPSTrackerLocation Location { get; set; }

        public void SerializeBinary(CoreBinaryWriter writer)
        {
            writer.Write(AuditTrail ?? "");
            writer.WriteObject(Location);
        }

        public void DeserializeBinary(CoreBinaryReader reader)
        {
            AuditTrail = reader.ReadString();
            Location = reader.ReadObject<GPSTrackerLocation>();
        }
    }

    public class GPSUpdateQueue
    {
        public string QueuePath;

        public GPSUpdateQueue(string queuePath)
        {
            QueuePath = queuePath;
        }

        public void InitQueueFolder()
        {
            try
            {
                Directory.CreateDirectory(QueuePath);
            }
            catch (Exception e)
            {
                throw new Exception($"Could not create directory for device update queue: {QueuePath}", e);
            }
        }

        public int GetNumberOfItems()
        {
            using var profiler = new Profiler(true);
            return Directory.EnumerateFiles(QueuePath).Count();
        }

        /// <summary>
        /// Get the first (earliest) items of the directory.
        /// </summary>
        /// <returns>A list of (filename, update) tuples.</returns>
        public IEnumerable<Tuple<string, GPSDeviceUpdate>> GetFirstItems()
        {
            using var profiler = new Profiler(true);
            var files = Directory.EnumerateFiles(QueuePath).OrderBy(x => x);

            foreach (var filename in files)
            {
                GPSDeviceUpdate? deviceUpdate = null;
                try
                {
                    using var fileStream = new FileStream(filename, FileMode.Open, FileAccess.Read);
                    deviceUpdate = Serialization.ReadBinary<GPSDeviceUpdate>(fileStream, BinarySerializationSettings.Latest);
                }
                catch
                {
                    // File is probably in use.
                }
                if(deviceUpdate is not null)
                {
                    yield return new Tuple<string, GPSDeviceUpdate>(filename, deviceUpdate);
                }
            }
        }

        public void QueueUpdate(GPSDeviceUpdate deviceUpdate)
        {
            var filename = Path.Combine(QueuePath, $"{DateTime.UtcNow.Ticks} - {deviceUpdate.Location.Tracker.ID}");

            using var fileStream = new FileStream(filename, FileMode.OpenOrCreate, FileAccess.Write);
            Serialization.WriteBinary(deviceUpdate, fileStream, BinarySerializationSettings.Latest);
        }

        public void QueueUpdate(string auditTrail, GPSTrackerLocation location) => QueueUpdate(new GPSDeviceUpdate
        {
            AuditTrail = auditTrail,
            Location = location
        });
    }

    internal class GPSDeviceCache : ConcurrentDictionary<string, Device>
    {
        public void Refresh()
        {
            Logger.Send(LogType.Information, "", "Refreshing Tracker Cache");
            var table = new Client<GPSTracker>().Query(
                null,
                new Columns<GPSTracker>(x => x.ID, x => x.DeviceID, x => x.Type.BatteryFormula));

            Logger.Send(LogType.Information, "", string.Format("- Tracker Cache: {0} devices", table.Rows.Count));
            Clear();
            foreach (var row in table.Rows)
            {
                var formula = row.Get<GPSTracker, string?>(x => x.Type.BatteryFormula);
                var expression = string.IsNullOrWhiteSpace(formula) ? null : new CoreExpression<GPSBatteryFormulaModel, double>(formula);
                this[row.Get<GPSTracker, string>(x => x.DeviceID)] =
                    new Device(row.Get<GPSTracker, Guid>(x => x.ID), DateTime.MinValue, expression);
            }
        }
    }

    public class GPSEngine : Engine<GPSServerProperties>
    {
        private Listener<SigfoxHandler, SigfoxHandlerProperties> sigfoxListener;
        private OEMListener oemListener;

        private GPSDeviceCache DeviceCache = new();

        private Timer RefreshDevicesTimer;

        private Timer UpdateServerTimer;

        private GPSUpdateQueue UpdateQueue;

        public override void Configure(Server server)
        {
            base.Configure(server);

            UpdateQueue = new GPSUpdateQueue(Path.Combine(AppDataFolder, "device_queue"));
        }

        private void StartOEMListener()
        {
            if (Properties.ListenPort == 0)
                throw new Exception("Error: OEM Listen Port not Specified\n");

            Logger.Send(LogType.Information, "", "Starting OEM Listener on port " + Properties.ListenPort);

            oemListener = new OEMListener(Properties.ListenPort, DeviceCache, UpdateQueue);
            oemListener.Start();

            Logger.Send(LogType.Information, "", "OEM Listener started on port " + Properties.ListenPort);
        }

        private void StartSigfoxListener()
        {
            if (Properties.SigfoxListenPort == 0)
            {
                Logger.Send(LogType.Information, "", "No Sigfox listen port specified\n");
                return;
            }

            sigfoxListener = new Listener<SigfoxHandler, SigfoxHandlerProperties>(new SigfoxHandlerProperties(DeviceCache, UpdateQueue));
            sigfoxListener.InitPort((ushort)Properties.SigfoxListenPort);

            Logger.Send(LogType.Information, "", "Starting Sigfox Listener on port " + Properties.SigfoxListenPort);

            sigfoxListener.Start();

            Logger.Send(LogType.Information, "", "Sigfox Listener started on port " + Properties.SigfoxListenPort);
        }
        
        private void StartUpdateServerTask()
        {
            UpdateServerTimer = new Timer(Properties.UpdateTimer);
            UpdateServerTimer.Elapsed += (o, e) => UpdateServer();
            UpdateServerTimer.Start();
        }

        // List of (filename, update)
        private Queue<Tuple<string, GPSDeviceUpdate>> LocationQueueCache = new();

        private void GetLocationQueue(int nLocations)
        {
            LocationQueueCache.EnsureCapacity(LocationQueueCache.Count + nLocations);
            foreach(var item in UpdateQueue.GetFirstItems().Take(nLocations))
            {
                LocationQueueCache.Enqueue(item);
            }
        }

        private void UpdateServer()
        {
            // Cache a set of fifty, so that we're not running baack and forth to the filesystem all the time.
            if(LocationQueueCache.Count == 0)
            {
                GetLocationQueue(50);
            }

            if (LocationQueueCache.Count > 0)
            {
                var (filename, update) = LocationQueueCache.Dequeue();

                Logger.Send(LogType.Information, "",
                    string.Format("Updating Server ({0}): {1} - {2}", UpdateQueue.GetNumberOfItems(), update.Location.DeviceID, update.AuditTrail));
                new Client<GPSTrackerLocation>().Save(update.Location, update.AuditTrail, (_, exception) =>
                {
                    if (exception is not null)
                    {
                        Logger.Send(LogType.Error, "", $"Error saving GPS Tracker Location ({update.AuditTrail}): {CoreUtils.FormatException(exception)}");
                    }
                });

                try
                {
                    File.Delete(filename);
                }
                catch
                {
                    // Probably got deleted.
                }
            }
        }

        public override void Run()
        {
            if (string.IsNullOrWhiteSpace(Properties.Server))
            {
                Logger.Send(LogType.Error, "", "Server is blank!");
                return;
            }

            Logger.Send(LogType.Information, "", "Registering Classes");

            CoreUtils.RegisterClasses();
            ComalUtils.RegisterClasses();

            ClientFactory.SetClientType(typeof(IPCClient<>), Platform.GPSEngine, Version, DatabaseServerProperties.GetPipeName(Properties.Server));
            CheckConnection();

            UpdateQueue.InitQueueFolder();

            // Refresh device cache and set up timer.
            DeviceCache.Refresh();
            RefreshDevicesTimer = new Timer(5 * 60 * 1000);
            RefreshDevicesTimer.Elapsed += (o, e) => DeviceCache.Refresh();
            RefreshDevicesTimer.Start();

            DMFactory.Initialise(Properties.DumpFormat, Properties.DumpFile);

            StartOEMListener();
            StartSigfoxListener();
            StartUpdateServerTask();
        }

        private bool CheckConnection()
        {
            if (ClientFactory.UserGuid == Guid.Empty)
            {
                // Wait for server connection
                while (!Client.Ping())
                {
                    Logger.Send(LogType.Error, "", "Database server unavailable. Trying again in 30 seconds...");
                    Task.Delay(30_000).Wait();
                    Logger.Send(LogType.Information, "", "Retrying connection...");
                }
                ClientFactory.SetBypass();
            }

            return true;
        }

        public override void Stop()
        {
            oemListener.Stop();
            sigfoxListener.Stop();
            UpdateServerTimer.Stop();
            RefreshDevicesTimer.Stop();
        }
    }
}