JiShe.CollectBus/src/JiShe.CollectBus.Application/Samples/SampleAppService.cs

using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Apache.IoTDB.DataStructure;
using Apache.IoTDB;
using Confluent.Kafka;
using JiShe.CollectBus.Ammeters;
using JiShe.CollectBus.FreeSql;
using JiShe.CollectBus.IoTDBProvider;
using JiShe.CollectBus.IotSystems.PrepayModel;
using Microsoft.AspNetCore.Authorization;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.Options;
using JiShe.CollectBus.IoTDBProvider.Context;
using Microsoft.Extensions.Logging;
using JiShe.CollectBus.IotSystems.AFNEntity;
using JiShe.CollectBus.Protocol.Contracts.Interfaces;
using Microsoft.Extensions.DependencyInjection;
using JiShe.CollectBus.Common.Consts;
using JiShe.CollectBus.Common.Enums;
using System.Diagnostics.Metrics;
using JiShe.CollectBus.Common.DeviceBalanceControl;
using JiShe.CollectBus.Kafka.Attributes;
using System.Text.Json;
using JiShe.CollectBus.Kafka;
using JiShe.CollectBus.Application.Contracts;
using JiShe.CollectBus.Common.Models;
using System.Diagnostics;

namespace JiShe.CollectBus.Samples;

public class SampleAppService : CollectBusAppService, ISampleAppService, IKafkaSubscribe
{
    private readonly ILogger<SampleAppService> _logger;
    private readonly IIoTDBProvider _iotDBProvider;
    private readonly IoTDBRuntimeContext _dbContext;
    private readonly IoTDBOptions _options;
    private readonly IRedisDataCacheService _redisDataCacheService;

    public SampleAppService(IIoTDBProvider iotDBProvider, IOptions<IoTDBOptions> options,
        IoTDBRuntimeContext dbContext, ILogger<SampleAppService> logger, IRedisDataCacheService redisDataCacheService)
    {
        _iotDBProvider = iotDBProvider;
        _options = options.Value;
        _dbContext = dbContext;
        _logger = logger;
        _redisDataCacheService = redisDataCacheService;
    }

    /// <summary>
    /// 测试 UseSessionPool
    /// </summary>
    /// <param name="timestamps"></param>
    /// <returns></returns>
    [HttpGet]
    public async Task UseSessionPool(long timestamps)
    {
        string? messageHexString = null;
        if (timestamps == 0)
        {
            timestamps = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
            _logger.LogError($"timestamps_{timestamps}");
        }
        else
        {
            messageHexString = messageHexString + timestamps;
        }

        ElectricityMeter meter = new ElectricityMeter()
        {
            SystemName = "energy",
            DeviceId = "402440506",
            DeviceType = "Ammeter",
            Current = 10,
            MeterModel = "DDZY-1980",
            ProjectCode = "10059",
            Voltage = 10,
            IssuedMessageHexString = messageHexString,
            Timestamps = timestamps,
        };
        await _iotDBProvider.InsertAsync(meter);
    }

    /// <summary>
    /// 测试Session切换
    /// </summary>
    /// <returns></returns>
    [HttpGet]
    public async Task UseTableSessionPool()
    {
        ElectricityMeter meter2 = new ElectricityMeter()
        {
            SystemName = "energy",
            DeviceId = "402440506",
            DeviceType = "Ammeter",
            Current = 10,
            MeterModel = "DDZY-1980",
            ProjectCode = "10059",
            Voltage = 10,
            Timestamps = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
        };

        await _iotDBProvider.InsertAsync(meter2);

        _dbContext.UseTableSessionPool = true;

        ElectricityMeter meter = new ElectricityMeter()
        {
            SystemName = "energy",
            DeviceId = "402440506",
            DeviceType = "Ammeter",
            Current = 10,
            MeterModel = "DDZY-1980",
            ProjectCode = "10059",
            Voltage = 10,
            Timestamps = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
        };
        await _iotDBProvider.InsertAsync(meter);
    }

    /// <summary>
    /// 测试设备分组均衡控制算法
    /// </summary>
    /// <param name="deviceCount"></param>
    /// <returns></returns>
    [HttpGet]
    public async Task TestDeviceGroupBalanceControl(int deviceCount = 200000)
    {
        //var deviceList = new List<string>();
        //for (int i = 0; i < deviceCount; i++)
        //{
        //    deviceList.Add($"Device_{Guid.NewGuid()}");
        //}

        //// 初始化缓存
        //DeviceGroupBalanceControl.InitializeCache(deviceList);

        var timeDensity = "15";
        //获取缓存中的电表信息
        var redisKeyList = $"{string.Format(RedisConst.CacheMeterInfoHashKey, "Energy", "JiSheCollectBus", MeterTypeEnum.Ammeter.ToString(), timeDensity)}*";

        var oneMinutekeyList = await FreeRedisProvider.Instance.KeysAsync(redisKeyList);
        var meterInfos = await GetMeterRedisCacheListData<AmmeterInfo>(oneMinutekeyList, "Energy", "JiSheCollectBus", timeDensity, MeterTypeEnum.Ammeter);
        List<string> focusAddressDataLista = new List<string>();
        foreach (var item in meterInfos)
        {
            focusAddressDataLista.Add(item.FocusAddress);
        }

        DeviceGroupBalanceControl.InitializeCache(focusAddressDataLista);
          
        // 打印分布统计
        DeviceGroupBalanceControl.PrintDistributionStats();

        await Task.CompletedTask;
    }

    /// <summary>
    /// 测试设备分组均衡控制算法获取分组Id
    /// </summary>
    /// <param name="deviceAddress"></param>
    /// <returns></returns>
    [HttpGet]
    public async Task TestGetDeviceGroupBalanceControl(string deviceAddress)
    { 
        var groupId = DeviceGroupBalanceControl.GetDeviceGroupId(deviceAddress);
        Console.WriteLine(groupId);

        await Task.CompletedTask;
    }


    /// <summary>
    /// 测试单个测点数据项
    /// </summary>
    /// <param name="measuring"></param>
    /// <returns></returns>
    [HttpGet]
    public async Task TestSingleMeasuringAFNData(string measuring, string value)
    {
        var meter = new SingleMeasuringAFNDataEntity<string>()
        {
            SystemName = "energy",
            DeviceId = "402440506",
            DeviceType = "Ammeter",
            ProjectCode = "10059",
            Timestamps = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
            SingleMeasuring = new Tuple<string, string>(measuring, value)
        };
        await _iotDBProvider.InsertAsync(meter);
    }

    /// <summary>
    /// 测试Redis批量读取10万条数据性能
    /// </summary>
    /// <returns></returns>
    [HttpGet]
    public async Task TestRedisCacheGetAllPagedData()
    {
        var timeDensity = "15";
        string SystemType = "Energy";
        string ServerTagName = "JiSheCollectBus2";
        var redisCacheMeterInfoHashKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoHashKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}";
        var redisCacheMeterInfoSetIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoSetIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}";
        var redisCacheMeterInfoZSetScoresIndexKeyTemp = $"{string.Format(RedisConst.CacheMeterInfoZSetScoresIndexKey, SystemType, ServerTagName, MeterTypeEnum.Ammeter, timeDensity)}";

        var timer1 = Stopwatch.StartNew();
        decimal? cursor = null;
        string member = null;
        bool hasNext;
        List<AmmeterInfo> meterInfos = new List<AmmeterInfo>();
        do
        {
            var page = await _redisDataCacheService.GetAllPagedData<AmmeterInfo>(
                redisCacheMeterInfoHashKeyTemp,
                redisCacheMeterInfoZSetScoresIndexKeyTemp,
                pageSize: 1000,
                lastScore: cursor,
                lastMember: member);

            meterInfos.AddRange(page.Items);
            cursor = page.HasNext ? page.NextScore : null;
            member = page.HasNext ? page.NextMember : null;
            hasNext = page.HasNext;
        } while (hasNext);

        timer1.Stop();
        _logger.LogError($"读取数据更花费时间{timer1.ElapsedMilliseconds}毫秒");

        List<string> focusAddressDataLista = new List<string>();
        foreach (var item in meterInfos)
        {
            focusAddressDataLista.Add(item.FocusAddress);
        }

        DeviceGroupBalanceControl.InitializeCache(focusAddressDataLista);

        // 打印分布统计
        DeviceGroupBalanceControl.PrintDistributionStats();

        await Task.CompletedTask;
    }


    public Task<SampleDto> GetAsync()
    {
        return Task.FromResult(
            new SampleDto
            {
                Value = 42
            }
        );
    }

    [Authorize]
    public Task<SampleDto> GetAuthorizedAsync()
    {
        return Task.FromResult(
            new SampleDto
            {
                Value = 42
            }
        );
    }

    [AllowAnonymous]
    public async Task<List<Vi_BaseAmmeterInfo>> Test()
    {

        var ammeterList = await SqlProvider.Instance.Change(DbEnum.PrepayDB).Select<Vi_BaseAmmeterInfo>().Where(d => d.TB_CustomerID == 5).Take(10).ToListAsync();
        return ammeterList;
    }

    [AllowAnonymous]
    public bool GetTestProtocol()
    {
        var aa =  LazyServiceProvider.GetKeyedService<IProtocolPlugin>("TestProtocolPlugin");
        return aa == null;
    }

    [KafkaSubscribe(ProtocolConst.TESTTOPIC)]

    public async Task<ISubscribeAck> KafkaSubscribeAsync(object obj)
    {
        _logger.LogWarning($"收到订阅消息： {obj}");
       return SubscribeAck.Success();
    }
}