iot/projects/IoTCenter/Services/IoTCenterEventHandler.cs

using Application.Domain.Entities;
using CSScriptLib;
using Hangfire;
using Infrastructure.Data;
using Infrastructure.Events;
using Infrastructure.Extensions;
using Microsoft.AspNetCore.SignalR;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Vibrant.InfluxDB.Client;
using Vibrant.InfluxDB.Client.Rows;

namespace IoTCenter.Services
{
    public class IoTCenterEventHandler :
        IEventHander<EntityInsertedEvent<IoTTimer>>,
        IEventHander<EntityUpdatedEvent<IoTTimer>>,
        IEventHander<EntityDeletedEvent<IoTTimer>>,
        IEventHander<EntityInsertedEvent<IoTTigger>>,
        IEventHander<EntityUpdatedEvent<IoTTigger>>,
        IEventHander<EntityDeletedEvent<IoTTigger>>,
        IEventHander<EntityInsertedEvent<Node>>,
        IEventHander<EntityUpdatedEvent<Node>>,
        IEventHander<EntityDeletedEvent<Node>>,
        IEventHander<EntityInsertedEvent<Product>>,
        IEventHander<EntityUpdatedEvent<Product>>,
        IEventHander<EntityDeletedEvent<Product>>,
        IEventHander<EntityInsertedEvent<Device>>,
        IEventHander<EntityUpdatedEvent<Device>>,
        IEventHander<EntityDeletedEvent<Device>>,
        IEventHander<EntityInsertedEvent<Data>>,
        IEventHander<EntityUpdatedEvent<Data>>,
        IEventHander<EntityDeletedEvent<Data>>,
        IEventHander<EntityInsertedEvent<Command>>,
        IEventHander<EntityUpdatedEvent<Command>>,
        IEventHander<EntityDeletedEvent<Command>>,
        IEventHander<EntityInsertedEvent<Scene>>,
        IEventHander<EntityUpdatedEvent<Scene>>,
        IEventHander<EntityDeletedEvent<Scene>>
    {
        public static ConcurrentDictionary<Guid, IoTTigger> Tiggers = new ConcurrentDictionary<Guid, IoTTigger>();
        private readonly IServiceProvider _sp;

        public IoTCenterEventHandler(IServiceProvider sp)
        {
            this._sp = sp;
        }

        #region timer

        public void Handle(EntityInsertedEvent<IoTTimer> message)
        {
            var timer = message.Data;
            if (timer.NodeId == null)
            {
                RecurringJob.AddOrUpdate<IoTCenterJob>(timer.Id.ToString(), o => o.TimerHanle(timer.Id), timer.Cron, TimeZoneInfo.Local);
            }
        }

        public void Handle(EntityUpdatedEvent<IoTTimer> message)
        {
            var timer = message.Data;
            if (timer.NodeId == null)
            {
                RecurringJob.AddOrUpdate<IoTCenterJob>(timer.Id.ToString(), o => o.TimerHanle(timer.Id), timer.Cron, TimeZoneInfo.Local);
            }
        }

        public void Handle(EntityDeletedEvent<IoTTimer> message)
        {
            var timer = message.Data;
            if (timer.NodeId == null)
            {
                RecurringJob.RemoveIfExists(timer.Id.ToString());
            }
        }

        #endregion timer

        #region tigger

        public void Handle(EntityInsertedEvent<IoTTigger> message)
        {
            var tigger = message.Data;
            if (tigger.NodeId == null)
            {
                Tiggers.TryRemove(message.Data.Id, out IoTTigger iotTigger);
                Tiggers.TryAdd(tigger.Id, tigger);
            }
        }

        public void Handle(EntityUpdatedEvent<IoTTigger> message)
        {
            var tigger = message.Data;
            if (tigger.NodeId == null)
            {
                Tiggers.TryRemove(message.Data.Id, out IoTTigger iotTigger);
                Tiggers.TryAdd(tigger.Id, tigger);
            }
        }

        public void Handle(EntityDeletedEvent<IoTTigger> message)
        {
            var tigger = message.Data;
            if (tigger.NodeId == null)
            {
                Tiggers.TryRemove(tigger.Id, out IoTTigger iotTigger);
            }
        }

        #endregion tigger

        #region Product

        public void Handle(EntityInsertedEvent<Product> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityUpdatedEvent<Product> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityDeletedEvent<Product> message)
        {
            this.Notify(message);
        }

        #endregion Product

        #region Node

        public void Handle(EntityInsertedEvent<Node> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityUpdatedEvent<Node> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityDeletedEvent<Node> message)
        {
            this.Notify(message);
        }

        #endregion Node

        #region Device

        public void Handle(EntityInsertedEvent<Device> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityUpdatedEvent<Device> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityDeletedEvent<Device> message)
        {
            this.Notify(message);
        }

        #endregion Device

        #region Data

        public void Handle(EntityInsertedEvent<Data> message)
        {
            this.Notify(message);
            this.TiggerHandle(message);
            this.LogData(message.Data);
        }

        public void Handle(EntityUpdatedEvent<Data> message)
        {
            this.Notify(message);
            this.TiggerHandle(message);
            this.LogData(message.Data);
        }

        public void Handle(EntityDeletedEvent<Data> message)
        {
            this.Notify(message);
        }

        #endregion Data

        #region Command

        public void Handle(EntityInsertedEvent<Command> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityUpdatedEvent<Command> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityDeletedEvent<Command> message)
        {
            this.Notify(message);
        }

        #endregion Command

        #region Scene

        public void Handle(EntityInsertedEvent<Scene> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityUpdatedEvent<Scene> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityDeletedEvent<Scene> message)
        {
            this.Notify(message);
        }

        #endregion Scene

        private void TiggerHandle(BaseEvent<Data> message)
        {
            try
            {
                using var scope = this._sp.CreateScope();
                var job = scope.ServiceProvider.GetService<IoTCenterJob>();
                foreach (var item in Tiggers)
                {
                    var data = message.Data;
                    var tigger = item.Value;
                    if (tigger.DataId == data.Id)
                    {
                        var methodText = $"bool Valid(string name,string key,{data.Type.ToString().ToLower()} value,string description){{ return {tigger.Condition};}}";
                        try
                        {
                            dynamic method = CSScript.Evaluator.LoadMethod(methodText);
                            dynamic value = data.GetValue();
                            var result = method.Valid(data.Name, data.Key, value, data.Description);
                            if (result)
                            {
                                job.TiggerHandle(tigger.Id);
                            }
                        }
                        catch (Exception ex)
                        {
                            ex.PrintStack();
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                ex.PrintStack();
            }
        }

        private void Notify<T>(BaseEvent<T> message)
        {
            using var scope = this._sp.CreateScope();
            var hub = scope.ServiceProvider.GetRequiredService<IHubContext<IoTCenterHub>>();
            Task.Run(() =>
            {
                try
                {
                    hub.ServerToClient($"{typeof(T).Name}{message.Arg}", message.Data, "page", null);
                }
                catch (Exception ex)
                {
                    ex.PrintStack();
                }
            });
        }

        private void LogData(Data data)
        {
            using var scope = this._sp.CreateScope();
            var _cfg = scope.ServiceProvider.GetRequiredService<IConfiguration>();
            var url = _cfg["influxdb:url"];
            var usr = _cfg["influxdb:usr"];
            var pwd = _cfg["influxdb:pwd"];
            var deviceRepo = scope.ServiceProvider.GetRequiredService<IRepository<Device>>();
            var device = deviceRepo.ReadOnlyTable().FirstOrDefault(o => o.Id == data.DeviceId);
            Task.Run(async () =>
            {
                try
                {
                    if (string.IsNullOrEmpty(data.Value))
                    {
                        return;
                    }
                    if (data.Type != DeviceDataType.Int && data.Type != DeviceDataType.Float)
                    {
                        return;
                    }
                    try
                    {
                        var dbName = "iot";
                        var measurementName = "data";
                        using var client = new InfluxClient(new Uri(url), usr, pwd);
                        await client.CreateDatabaseAsync(dbName);
                        var row = new DynamicInfluxRow
                        {
                            Timestamp = DateTimeOffset.FromUnixTimeMilliseconds(data.Timestamp).DateTime
                        };
                        row.Fields.Add("DeviceNumber", device.Number);
                        row.Fields.Add("DeviceName", device.Name);
                        row.Fields.Add(data.Key, this.GetDataValue(data));
                        await client.WriteAsync(dbName, measurementName, new List<DynamicInfluxRow> { row });
                    }
                    catch (Exception ex)
                    {
                        ex.PrintStack();
                    }
                }
                catch (Exception ex)
                {
                    ex.PrintStack();
                }
            });
        }

        private object GetDataValue(Data model)
        {
            return model.Type switch
            {
                DeviceDataType.Int => Convert.ToInt32(model.Value),

                DeviceDataType.Float => Convert.ToSingle(model.Value),

                _ => model.Value,
            };
        }
    }
}