qlhb/Ropin.IOT.MqttService/MqttWorkService.cs

using Autofac.Core;
using LinqKit;
using Microsoft.Extensions.Hosting;
using Ropin.Inspection.Model.Entities;
using Microsoft.Extensions.Logging;
using Ropin.Inspection.Repository;
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.EntityFrameworkCore.Internal;
using System.Linq;
using System.Data;
using Microsoft.Extensions.DependencyInjection;
using Quartz.Impl.AdoJobStore.Common;
using Ropin.Core.Extensions.Redis;
using AutoMapper.Configuration;
using Newtonsoft.Json;
using System.Security.Cryptography;
using log4net;
using Ropin.Inspection.Model;

namespace Ropin.IOT.MqttService
{
    public class MqttWorkService : IHostedService, IDisposable
    {
        private readonly ILogger _logger;
        private readonly IServiceProvider _provider;
        private readonly IRedisBasketRepository _redisService;
        private readonly Func<InspectionDbContext> _dbFuncContextFactory;
        public readonly ITdevBoxDevSpotRepository _tdevBoxDevSpotRepository;
        public readonly MqttOptions _mqttOptions;
        private static readonly ILog log = LogManager.GetLogger(typeof(MqttWorkService));
        public MqttWorkService(ILogger<MqttWorkService> logger,
            IServiceProvider provider, 
            MqttOptions mqttOptions, 
            IRedisBasketRepository redisService,
            Func<InspectionDbContext> dbContextFactory
            //, ITdevBoxDevSpotRepository tdevBoxDevSpotRepository
            )
        {
            _logger = logger;
            _provider = provider;
            using (var scope = _provider.GetRequiredService<IServiceScopeFactory>().CreateScope())
            {
                _tdevBoxDevSpotRepository = scope.ServiceProvider.GetService<ITdevBoxDevSpotRepository>();
            };
            _mqttOptions = mqttOptions;
            _redisService = redisService;
            _dbFuncContextFactory = dbContextFactory;
            //_tdevBoxDevSpotRepository = tdevBoxDevSpotRepository;
            _redisService.Set("mqtt_BoxDevSpot_IsUpdate", "1", TimeSpan.FromDays(7));
        }
        public void Dispose()
        {
            //throw new NotImplementedException();
        }

        public Task StartAsync(CancellationToken cancellationToken)
        {
            Task.Run(async () =>
            {
                while (true)
                {
                    try
                    {
                        var bol = await _redisService.Exist("mqtt_BoxDevSpot_IsUpdate");
                        log.Info($"[开始] MqttWorkService.【mqtt_BoxDevSpot_IsUpdate是否存在：{bol}】");
                        if (bol)
                        {
                            string val = await _redisService.GetValue("mqtt_BoxDevSpot_IsUpdate");
                            log.Info($"[开始] MqttWorkService.【mqtt_BoxDevSpot_IsUpdate是否修改值：{bol}】");
                            if (val=="1")
                            {
                                IList<TDEV_BoxDevSpot> boxDevSpot = null;
                                List <MqttDevicePoint> devicePointList = null;
                                using (var dbContext = _dbFuncContextFactory())
                                {
                                    //boxDevSpot = dbContext.TDEV_BoxDevSpot.AsQueryable().ToList();
                                    //if (boxDevSpot != null && boxDevSpot.Count > 0)
                                    //{
                                    //    boxDevSpot = boxDevSpot.Where(t => t.C_Status == "1").ToList();
                                    //}

                                    var query = from a in dbContext.TDEV_Box
                                                join b in dbContext.TDEV_BoxDevSpot
                                                on a.C_ID equals b.C_BoxCode
                                                where a.C_Status == "1" && b.C_Status == "1"
                                                //group b by b.C_ID into g
                                                select new MqttDevicePoint
                                                {
                                                    Id = b.C_ID,
                                                    ChineseName = b.C_ChineseName,
                                                    EnglishName = b.C_EnglishName,
                                                    BoxId = a.C_BoxNo,
                                                    GroupName = b.C_GroupName,
                                                };

                                    devicePointList = query.ToList();
                                }
                                log.Info($"[开始] MqttWorkService.TDEV_BoxDevSpot数据【{JsonConvert.SerializeObject(boxDevSpot)}】");
                                if (devicePointList != null&& devicePointList.Count>0)
                                {
                                   
                                    MqttClientService.AddMqttClient1(_mqttOptions, devicePointList);
                                    await _redisService.Set("mqtt_BoxDevSpot_IsUpdate", "0", TimeSpan.FromDays(7));
                                    log.Info($"[修改] mqtt_BoxDevSpot_IsUpdate=0");
                                }
                            }
                        }
                        await Task.Delay(6000);
                    }
                    catch (Exception ex)
                    {
                        log.Error("[MqttWorkService]Task1：" + ex.Message);
                        await Task.Delay(5000);//5秒
                    }
                }

            });
            return Task.CompletedTask;
        }

        public Task StopAsync(CancellationToken cancellationToken)
        {
            Dispose();
            _logger.LogInformation("内部任务计划结束");
            return Task.CompletedTask;
        }
    }
}