FYCore/LanSheng/LanShengService/Tcp/ChannelHandler.cs

using CommonModel;
using DotNetty.Transport.Channels;
using DotNetty.Transport.Channels.Groups;
using Google.Api;
using LanShengInterface;
using LanShengModel;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Text.Json;
using System.Threading.Tasks;

namespace LanShengService.Tcp
{
    public class ChannelHandler : ChannelHandlerAdapter
    {
        public ChannelHandler(IServiceProvider services)
        {
            Services = services;
            Logger = Services.GetRequiredService<ILogger<ChannelHandler>>();
            DeviceService = Services.GetRequiredService<IDeviceService>();
            TcpService = Services.GetRequiredService<ITcpService>();
        }

        private readonly IServiceProvider Services;

        private readonly IDeviceService DeviceService;

        private readonly ILogger Logger;

        private readonly ITcpService TcpService;

        private int TryCount = 0;

        private string DTU_Id = "";

        public override async void ChannelRead(IChannelHandlerContext context, object message)
        {
            var packet = message as TcpDataLog<DeviceData>;
            DTU_Id = packet!.DTU_ID!;
            Logger.LogDebug($"接收到数据：{JsonSerializer.Serialize<TcpDataLog<DeviceData>>(packet!)}");
            await TcpService.AddChannel(packet!.DTU_ID!, context);
            await DeviceService.InsertDataLog(packet!);

            if (packet!.Content != null)
            {
                await DeviceService.InsertData(packet!.Content);
            }

            if (packet.Msg_ID == "1001" || packet.Msg_ID == "1103")
            {
                var Reply = new TcpDataLog<DeviceData>
                {
                    Frame_START = packet.Frame_START,
                    Frame_number = packet.Frame_number,
                    DTU_ID = packet.DTU_ID,
                    Frame_END = packet.Frame_END
                };
                if (packet.Msg_ID == "1001")
                {
                    Reply.Msg_ID = "9001";
                }
                if (packet.Msg_ID == "1103")
                {
                    Reply.Msg_ID = "9103";
                }
                try
                {
                    await context.WriteAndFlushAsync(Reply);
                }
                catch
                {
                    await context.CloseAsync();
                }
            }
            if (packet.Msg_ID == "9104")
            {
                await TcpService.RemoveMessageItem(packet);
            }
            TryCount = 0;
        }

        public override void ChannelReadComplete(IChannelHandlerContext context)
        {
            context.Flush();
        }

        public override void ExceptionCaught(IChannelHandlerContext context, Exception exception)
        {
            TryCount++;

            Logger.LogError(exception.ToString());

            if (TryCount >= 5)
            {
                context.CloseAsync();
            }
        }

        public override async void ChannelActive(IChannelHandlerContext context)
        {
            await TcpService.AddChannel("all", context);
            base.ChannelActive(context);
        }

        public override async void ChannelInactive(IChannelHandlerContext context)
        {
            if (!string.IsNullOrEmpty(DTU_Id))
            {
                await DeviceService.Offline(DTU_Id);
            }
            await context.CloseAsync();
            base.ChannelInactive(context);
        }
    }
}