iot/projects/IoT/IoT.Shared/Infrastructure/NodeService.cs

using Application.Domain.Entities;
using Application.Models;
using Infrastructure.Data;
using Infrastructure.Events;
using Infrastructure.Extensions;
using Microsoft.AspNetCore.SignalR.Client;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net.Http;
using System.Threading;
using System.Threading.Tasks;

namespace IoT.Shared.Infrastructure
{
    public class NodeService : IHostedService, IDisposable
    {
        private string _notifyHost;
        private HubConnection Connection;
        private readonly IServiceProvider applicationServices;
        private readonly IConfiguration _cfg;
        public string ConnectionId { get; private set; }

        public NodeService(IServiceProvider applicationServices, IConfiguration configuration)
        {
            this.applicationServices = applicationServices;
            this._cfg = configuration;
            this.ConnectionId = Guid.NewGuid().ToBase62();
        }

        public Task StartAsync(CancellationToken cancellationToken)
        {
            Task.Run(async () =>
            {
                while (!cancellationToken.IsCancellationRequested)
                {
                    this.Connect();
                    await Task.Delay(10 * 1000).ConfigureAwait(true);
                }
            });
            return Task.CompletedTask;
        }

        public Task StopAsync(CancellationToken cancellationToken)
        {
            return Task.CompletedTask;
        }

        [System.Diagnostics.CodeAnalysis.SuppressMessage("Design", "CA1031:不捕获常规异常类型", Justification = "<挂起>")]
        public void Connect()
        {
            if (this._cfg.GetValue<bool>("notify:enabled", false))
            {
                Console.WriteLine("notify is enabled");
                try
                {
                    if (Connection == null)
                    {
                        Console.WriteLine("connection is null");
                        InitConnection();
                    }
                    if (Connection.State == HubConnectionState.Disconnected)
                    {
                        Console.WriteLine("start connect");
                        if (this._notifyHost != this._cfg["notify:host"])
                        {
                            InitConnection();
                        }
                        Connection.StartAsync().Wait();
                        this.OnConnected();
                    }
                    else
                    {
                        if (this._notifyHost != this._cfg["notify:host"])
                        {
                            this.ReConnect(null);
                        }
                        else
                        {
                            Console.WriteLine($"connection has connected");
                        }
                    }
                }
                catch (Exception ex)
                {
                    ex.PrintStack();
                }
            }
            else
            {
                Console.WriteLine("notify is disabled");
                this.Close();
            }
        }

        public void Close()
        {
            if (this.Connection != null)
            {
                if (this.Connection.State == HubConnectionState.Connected)
                {
                    this.Connection.StopAsync();
                }
                this.Connection.DisposeAsync();
                this.Connection = null;
            }
        }

        private Task ReConnect(Exception arg)
        {
            this.Close();
            this.Connect();
            return Task.CompletedTask;
        }

        private void InitConnection()
        {
            this._notifyHost = this._cfg["notify:host"];
            var url = $"http://{this._notifyHost}/hub?group={this._cfg["sn"]}";
            Console.WriteLine($"init connection for {url}");
            if (this.Connection != null)
            {
                this.Connection.DisposeAsync();
            }
            this.Connection = new HubConnectionBuilder().WithUrl(url).Build();
            this.Connection.Closed += ReConnect;
            this.Connection.On(Methods.ServerToClient, (string method, string message, string fromConnectionId) => this.OnServerToClient(method, message, fromConnectionId));
        }

        public void ServerToClient(string method, string message, string fromConnectionId)
        {
            this.OnServerToClient(method, method, fromConnectionId);
        }

        [System.Diagnostics.CodeAnalysis.SuppressMessage("Design", "CA1031:不捕获常规异常类型", Justification = "<挂起>")]
        public void ClientToServer(string method, string message, string fromConnectionId = null)
        {
            Task.Run(() =>
            {
                try
                {
                    if (this.Connection != null && this.Connection.State == HubConnectionState.Connected)
                    {
                        this.Connection.SendAsync(Methods.ClientToServer, method, message, fromConnectionId ?? this._cfg["sn"]);
                    }
                    else
                    {
                        Console.WriteLine($"{_notifyHost} not connected");
                    }
                }
                catch (Exception ex)
                {
                    ex.PrintStack();
                }
            });
        }

        public void Dispose()
        {
            this.Close();
        }

        /////////////////////////////
        public void OnConnected()
        {
            Console.WriteLine($"{_notifyHost} OnConnected");
            this.UpdateServer();
        }

        [System.Diagnostics.CodeAnalysis.SuppressMessage("Design", "CA1031:不捕获常规异常类型", Justification = "<挂起>")]
        public void OnServerToClient(string method, string message, string fromConnectionId)
        {
            using var scope = this.applicationServices.CreateScope();
            if (method == Methods.HealthCheckRequest)
            {
                this.ClientToServer(Methods.HealthCheckResponse, this._cfg["sn"]);
            }
            else if (method == Methods.GetProductRequest)
            {
                var productNumber = message;
                var productRepo = scope.ServiceProvider.GetService<IRepository<Product>>();
                var product = productRepo.ReadOnlyTable().Include(o => o.Category).FirstOrDefault(o => o.Number == productNumber);
                if (product == null)
                {
                    throw new Exception($"not has infoNumber:{productNumber}");
                }
                var model = product.To<EditProductModel>();
                model.CategoryNumber = product.Category.Number;
                this.SendToServer(Methods.GetProductResponse, model);
            }
            else if (method == Methods.CallApi)
            {
                var cfg = scope.ServiceProvider.GetService<IConfiguration>();
                var port = cfg["server.urls"].Split(':')[2];
                var url = $"http://localhost:{port}{message}";
                var httpClient = scope.ServiceProvider.GetService<IHttpClientFactory>().CreateClient();
                var result = httpClient.GetStringAsync(url).Result;
                this.Connection.SendAsync(Methods.ClientToServer, Methods.ApiCallback, result, cfg["sn"]);
            }
            else if (method == Methods.CallScene)
            {
                var commandRepo = scope.ServiceProvider.GetService<IRepository<Command>>();
                var commands = commandRepo.ReadOnlyTable()
                    .Include(o => o.Api)
                    .Include(o => o.Device)
                    //.Where(o => o.Scene.Name == message)
                    .ToList();
                foreach (var command in commands)
                {
                    try
                    {
                        var cfg = scope.ServiceProvider.GetService<IConfiguration>();
                        var port = cfg["server.urls"].Split(':')[2];
                        var url = $"http://localhost:{port}{command.Api.Path}{command.Api.Command}?number={command.Device.Number}{(string.IsNullOrEmpty(command.QueryString) ? "" : "&")}{command.QueryString}";
                        var httpClient = scope.ServiceProvider.GetService<IHttpClientFactory>().CreateClient();
                        var result = httpClient.GetStringAsync(url).Result;
                    }
                    catch (Exception ex)
                    {
                        ex.PrintStack();
                    }
                }
            }
            else if (method == Methods.RefreshNodeRequest)
            {
                this.UpdateServer();
            }
            else if (method == Methods.EditNodeRequest)
            {
                var model = message.FromJson<EditNodeModel>();
                var nodeRepo = scope.ServiceProvider.GetService<IRepository<Node>>();
                var node = nodeRepo.Table().FirstOrDefault(o => o.Number == model.Number);
                if (node != null)
                {
                    node.FromDto(model);
                    nodeRepo.SaveChanges();
                    this.ClientToServer(Methods.EditNodeResponse, message);
                }
            }
            else if (method == Methods.EditDeviceRequest)
            {
                var model = message.FromJson<EditDeviceModel>();
                var deviceRepo = scope.ServiceProvider.GetService<IRepository<Device>>();
                var device = deviceRepo.Table().FirstOrDefault(o => o.Number == model.Number);
                if (device != null)
                {
                    device.FromDto(model);
                    deviceRepo.SaveChanges();
                    this.ClientToServer(Methods.EditDeviceResponse, message);
                }
            }
            else if (method == Methods.DeleteDeviceRequest)
            {
                var deviceRepo = scope.ServiceProvider.GetService<IRepository<Device>>();
                var model = message.FromJson<EditDeviceModel>();
                var device = deviceRepo.Table().FirstOrDefault(o => o.Number == model.Number);
                if (device != null)
                {
                    deviceRepo.Delete(device);
                    deviceRepo.SaveChanges();
                    this.ClientToServer(Methods.DeleteDeviceResponse, message);
                }
            }
            else if (method == Methods.EditDataRequest)
            {
                var model = message.FromJson<EditDataModel>();
                var dataRepo = scope.ServiceProvider.GetService<IRepository<Data>>();
                var data = dataRepo.Table().FirstOrDefault(o => o.Device.Number == model.DeviceNumber && o.Key == model.Key);
                if (data != null)
                {
                    data.FromDto(model);
                    dataRepo.SaveChanges();
                    this.ClientToServer(Methods.EditDataResponse, new List<EditDataModel>() { model }.ToJson());
                }
            }
            else if (method == Methods.DeleteDataRequest)
            {
                var model = message.FromJson<EditDataModel>();
                var dataRepo = scope.ServiceProvider.GetService<IRepository<Data>>();
                var data = dataRepo.Table().FirstOrDefault(o => o.Device.Number == model.DeviceNumber && o.Key == model.Key);
                if (data != null)
                {
                    dataRepo.Delete(data);
                    dataRepo.SaveChanges();
                    this.ClientToServer(Methods.DeleteDataResponse, message);
                }
            }
            else if (method == Methods.EditSceneRequest)
            {
                var model = message.FromJson<EditSceneModel>();
                var nodeRepo = scope.ServiceProvider.GetService<IRepository<Node>>();
                var sceneRepo = scope.ServiceProvider.GetService<IRepository<Scene>>();
                var scene = sceneRepo.Table().FirstOrDefault(o => o.Id == model.Id);
                if (scene == null)
                {
                    scene = new Scene
                    {
                        Id = model.Id
                    };
                    sceneRepo.Add(scene);
                }
                scene.FromDto(model);
                scene.NodeId = nodeRepo.ReadOnlyTable().FirstOrDefault(o => o.Number == model.NodeNumber).Id;
                sceneRepo.SaveChanges();
                this.ClientToServer(Methods.EditSceneResponse, message);
            }
            else if (method == Methods.DeleteSceneRequest)
            {
                var model = message.FromJson<EditSceneModel>();
                var sceneRepo = scope.ServiceProvider.GetService<IRepository<Scene>>();
                var scene = sceneRepo.Table().FirstOrDefault(o => o.Id == model.Id);
                if (scene != null)
                {
                    sceneRepo.Delete(scene);
                    sceneRepo.SaveChanges();
                }
                this.ClientToServer(Methods.DeleteSceneResponse, message);
            }
            else if (method == Methods.EditCommandRequest)
            {
                var model = message.FromJson<EditCommandModel>();
                var commandRepo = scope.ServiceProvider.GetService<IRepository<Command>>();
                var apiRepo = scope.ServiceProvider.GetService<IRepository<Api>>();
                var deviceRepo = scope.ServiceProvider.GetService<IRepository<Device>>();
                var device = deviceRepo.ReadOnlyTable().FirstOrDefault(o => o.Number == model.DeviceNumber);

                var command = commandRepo.Table().FirstOrDefault(o => o.Id == model.Id);
                if (command == null)
                {
                    command = new Command
                    {
                        Id = model.Id
                    };
                    commandRepo.Add(command);
                }
                command.FromDto(model);
                command.DeviceId = device.Id;
                command.ApiId = apiRepo.ReadOnlyTable().Where(o => o.ProductId == device.ProductId && o.Name == model.ApiName).FirstOrDefault().Id;
                commandRepo.SaveChanges();
                this.ClientToServer(Methods.EditCommandResponse, message);
            }
            else if (method == Methods.DeleteCommandRequest)
            {
                var model = message.FromJson<EditCommandModel>();
                var commandRepo = scope.ServiceProvider.GetService<IRepository<Command>>();
                var command = commandRepo.Table().FirstOrDefault(o => o.Id == model.Id);
                if (command != null)
                {
                    commandRepo.Delete(command);
                    commandRepo.SaveChanges();
                }
                this.ClientToServer(Methods.DeleteCommandResponse, message);
            }
            else if (method == Methods.EditSceneCommandRequest)
            {
                var model = message.FromJson<EditSceneCommandModel>();
                var sceneRepo = scope.ServiceProvider.GetService<IRepository<Scene>>();
                var commandRepo = scope.ServiceProvider.GetService<IRepository<Command>>();
                var scene = sceneRepo.Table().FirstOrDefault(o => o.Id == model.Id);
                var command = commandRepo.Table().FirstOrDefault(o => o.Id == model.Id);
                var sceneCommandRepo = scope.ServiceProvider.GetService<IRepository<SceneCommand>>();
                var sceneCommand = sceneCommandRepo.Table().FirstOrDefault(o => o.Id == model.Id);
                if (sceneCommand == null)
                {
                    sceneCommand = new SceneCommand
                    {
                        Id = model.Id
                    };
                    sceneCommandRepo.Add(sceneCommand);
                }
                sceneCommand.SceneId = model.SceneId.Value;
                sceneCommand.CommandId = model.CommandId.Value;
                sceneCommandRepo.SaveChanges();
                this.ClientToServer(Methods.EditSceneCommandResponse, message);
            }
            else if (method == Methods.DeleteSceneCommandRequest)
            {
                var model = message.FromJson<EditCommandModel>();
                var sceneCommandRepo = scope.ServiceProvider.GetService<IRepository<SceneCommand>>();
                var sceneCommand = sceneCommandRepo.Table().FirstOrDefault(o => o.Id == model.Id);
                if (sceneCommand != null)
                {
                    sceneCommandRepo.Delete(sceneCommand);
                    sceneCommandRepo.SaveChanges();
                }
                this.ClientToServer(Methods.DeleteSceneCommandResponse, message);
            }
            else if (method == Methods.EditTimerRequest)
            {
                var model = message.FromJson<EditTimerModel>();
                var nodeRepo = scope.ServiceProvider.GetService<IRepository<Node>>();
                var timerRepo = scope.ServiceProvider.GetService<IRepository<IoTTimer>>();
                var timer = timerRepo.Table().FirstOrDefault(o => o.Id == model.Id);
                if (timer == null)
                {
                    timer = new IoTTimer
                    {
                        Id = model.Id
                    };
                    timerRepo.Add(timer);
                }
                timer.FromDto(model);
                timer.NodeId = nodeRepo.ReadOnlyTable().FirstOrDefault(o => o.Number == model.NodeNumber).Id;
                timerRepo.SaveChanges();
                scope.ServiceProvider.GetService<IEventPublisher>().Publish(new EntityUpdatedEvent<IoTTimer>(timer));
                this.ClientToServer(Methods.EditTimerResponse, message);
            }
            else if (method == Methods.DeleteTimerRequest)
            {
                var model = message.FromJson<EditTimerModel>();
                var timerRepo = scope.ServiceProvider.GetService<IRepository<IoTTimer>>();
                var timer = timerRepo.Table().FirstOrDefault(o => o.Id == model.Id);
                if (timer != null)
                {
                    timerRepo.Delete(timer);
                    timerRepo.SaveChanges();
                    scope.ServiceProvider.GetService<IEventPublisher>().Publish(new EntityDeletedEvent<IoTTimer>(timer));
                }
                this.ClientToServer(Methods.DeleteTimerResponse, message);
            }
            else if (method == Methods.EditTiggerRequest)
            {
                var model = message.FromJson<EditTiggerModel>();
                var nodeRepo = scope.ServiceProvider.GetService<IRepository<Node>>();
                var repo = scope.ServiceProvider.GetService<IRepository<IoTTigger>>();
                var entity = repo.Table().FirstOrDefault(o => o.Id == model.Id);
                if (entity == null)
                {
                    entity = new IoTTigger
                    {
                        Id = model.Id
                    };
                    repo.Add(entity);
                }
                entity.FromDto(model);
                entity.NodeId = nodeRepo.ReadOnlyTable().FirstOrDefault(o => o.Number == model.NodeNumber).Id;
                repo.SaveChanges();
                scope.ServiceProvider.GetService<IEventPublisher>().Publish(new EntityUpdatedEvent<IoTTigger>(entity));
                this.ClientToServer(Methods.EditTiggerResponse, message);
            }
            else if (method == Methods.DeleteTiggerRequest)
            {
                var model = message.FromJson<EditTiggerModel>();
                var repo = scope.ServiceProvider.GetService<IRepository<IoTTigger>>();
                var entity = repo.Table().FirstOrDefault(o => o.Id == model.Id);
                if (entity != null)
                {
                    repo.Delete(entity);
                    repo.SaveChanges();
                    scope.ServiceProvider.GetService<IEventPublisher>().Publish(new EntityDeletedEvent<IoTTigger>(entity));
                }
                this.ClientToServer(Methods.DeleteTiggerResponse, message);
            }
        }

        private void UpdateServer()
        {
            using var scope = this.applicationServices.CreateScope();
            //同步Node
            var nodeRepo = scope.ServiceProvider.GetService<IRepository<Node>>();
            var nodeDto = nodeRepo.ReadOnlyTable().FirstOrDefault().To<EditNodeModel>();
            this.SendToServer(Methods.EditNodeResponse, nodeDto);

            //同步Product
            var productRepo = scope.ServiceProvider.GetService<IRepository<Product>>();
            var products = productRepo.ReadOnlyTable()
                .Include(o => o.Category)
                .ToList();
            foreach (var product in products)
            {
                var productDto = product.To<EditProductModel>();
                productDto.CategoryNumber = product.Category.Number;
                this.SendToServer(Methods.GetProductResponse, productDto);
            }

            var deviceRepo = scope.ServiceProvider.GetService<IRepository<Device>>();
            var devices = deviceRepo.ReadOnlyTable()
                .Include(o => o.Product)
                .Include(o => o.Node)
                .Include(o => o.Data).ToList();

            //从服务器删除节点已经删除的设备
            this.SendToServer(Methods.RefreshDeviceListResponse, devices.Select(o => o.Number).ToList());

            //同步设备及其数据
            foreach (var device in devices)
            {
                var deviceDto = device.To<EditDeviceModel>();
                deviceDto.ProductNumber = device.Product.Number;
                deviceDto.NodeNumber = device.Node.Number;
                this.SendToServer(Methods.EditDeviceResponse, deviceDto);
                var dataDtoList = new List<EditDataModel>();
                foreach (var deviceData in device.Data.Where(o => !o.Hidden))
                {
                    var dataDto = deviceData.To<EditDataModel>();
                    dataDto.DeviceNumber = device.Number;
                    dataDtoList.Add(dataDto);
                }
                this.SendToServer(Methods.EditDataResponse, dataDtoList);
            }

            var sceneRepo = scope.ServiceProvider.GetService<IRepository<Scene>>();
            var scenes = sceneRepo.ReadOnlyTable().Include(o => o.Node).ToList();
            //从服务器删除节点已经删除的场景
            this.SendToServer(Methods.RefreshSceneListResponse, scenes.Select(o => o.Id).ToList());
            //同步场景
            foreach (var scene in scenes)
            {
                var sceneDto = scene.To<EditSceneModel>();
                sceneDto.NodeNumber = scene.Node.Number;
                this.SendToServer(Methods.EditSceneResponse, sceneDto);
            }

            var commandRepo = scope.ServiceProvider.GetService<IRepository<Command>>();
            var commands = commandRepo.ReadOnlyTable().Include(o => o.Api).Include(o => o.Device).ThenInclude(o => o.Node).ToList();
            //从服务器删除节点已经删除的命令
            this.SendToServer(Methods.RefreshCommandListResponse, commands.Select(o => o.Id).ToList());
            //同步命令
            foreach (var command in commands)
            {
                var commandDto = command.To<EditCommandModel>();
                commandDto.NodeNumber = command.Device.Node.Number;
                commandDto.DeviceNumber = command.Device.Number;
                commandDto.ApiName = command.Api.Name;
                this.SendToServer(Methods.EditCommandResponse, commandDto);
            }
        }

        [System.Diagnostics.CodeAnalysis.SuppressMessage("Design", "CA1031:不捕获常规异常类型", Justification = "<挂起>")]
        public void SendToServer(string method, object data)
        {
            Task.Run(() =>
            {
                try
                {
                    Console.WriteLine($"send node to server {_notifyHost}");
                    this.ClientToServer(method, data.ToJson());
                }
                catch (Exception ex)
                {
                    ex.PrintStack();
                }
            });
        }
    }
}