修改 DataDocking4Socket 对接程序

ningshuxia

2022-10-28 3ccb7c60e1ed8b6748ed7fb8b64b1dbe50d62abf

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using IStation.Untity;
using SuperSocket;
 
namespace IStation.Server
{
    /// <summary>
    /// 接收信息辅助类
    /// </summary>
    internal class PackageHandleHelper
    {
        public static List<Model.DataDockingConfigureExSocket> ConfigureList { get; set; }
 
        //接收数据
        public async ValueTask Handle(IAppSession session, PackageInfo requestInfo)
        {
            await Task.Run(() =>
            {
                var mySession = (MySession)session;
                if (ConfigureList == null || ConfigureList.Count < 1)
                {
                    LogHelper.Info("Socket数据对接任务中，未检索到采用Socket方式的数据对接配置！");
                    return;
                }
 
                var message_str = BytesTransfer.ToString(requestInfo.Body);
                var message_bts = BitTransfer.ToString(requestInfo.Body);
                LogHelper.Info($"Socket数据对接服务端口:{session.Server.Options.Listeners.First().Port}，接收到数据：{message_bts}！");
 
                var configure = string.IsNullOrEmpty(mySession.RegisterCode) ? ConfigureList.Find(x => x.ConfigureParas.RegisterCode == message_str)
                    : ConfigureList.Find(x => x.ConfigureParas.RegisterCode == mySession.RegisterCode);
                if (configure == null)
                {
                    configure = string.IsNullOrEmpty(mySession.Heartbeat) ? ConfigureList.Find(x => x.ConfigureParas.Heartbeat == message_str)
                  : ConfigureList.Find(x => x.ConfigureParas.Heartbeat == mySession.Heartbeat);
                    if (configure != null)
                    {
                        mySession.Close($"关闭:[{configure.Name}]连接,未检测到注册码!");
                        return;
                    }
                }
                if (configure == null)
                {
                    LogHelper.Info($"Socket数据对接任务中,端口：{mySession.Server.Options.Listeners.First().Port},收到一条无法识别的消息：{message_str}");
                    return;
                }
 
                //session初始化
                mySession.SessionName = configure.Name;
                mySession.RegisterCode = configure.ConfigureParas.RegisterCode;
                mySession.Heartbeat = configure.ConfigureParas.Heartbeat;
 
                //创建对接对象
                var dataDocking = DataDockingSocketFactory.CreateSocket<IDataDockingSocket.IHandleHelper>(configure.ConfigureParas.DependencyFile);
                if (dataDocking == null)
                {
                    LogHelper.Info($"Socket数据对接中,{configure.Name},创建数据对接对象失败！");
                    return;
                }
                if (message_str == mySession.RegisterCode)
                {
                    LogHelper.Info($"Socket数据对接中,{configure.Name},接收到注册码：{message_str}");
                    dataDocking.HandleRegisterCode(mySession);
                }
                else if (message_str == mySession.Heartbeat)
                {
                    LogHelper.Info($"Socket数据对接中,{configure.Name},接收到心跳包：{message_str}");
                    dataDocking.HandleHeartbeat(mySession);
                }
                else
                {
                    LogHelper.Info($"Socket数据对接中,{configure.Name}，接收到数据：{message_bts}");
                    dataDocking.HandleData(mySession, requestInfo.Body, configure.Mappers, (receive_list) =>
                    {
                        if (receive_list == null || receive_list.Count < 1)
                        {
                            LogHelper.Info($"Socket数据对接中，注册码:{configure.Name},获取测点记录失败！");
                            return;
                        }
                        var queue = new RabbitMqQueueHelper();
                        queue.Push(ConfigHelper.QueueName, new Model.MonitorDataDockingCorpRecord()
                        {
                            CorpID = configure.CorpID,
                            ConfigureID = configure.ID,
                            Records = receive_list
                        });
                        LogHelper.Info($"Socket数据对接中:{configure.Name}，成功推入通道{receive_list.Count()}条数据！");
                    });
                }
 
 
            });
 
 
        }
 
 
    }
}