first commit

2026-09-02 16:31:50 +08:00
commit a461727193
911 changed files with 692450 additions and 0 deletions
@@ -0,0 +1,59 @@
+using DotNetty.Buffers;
+using DotNetty.Codecs;
+using DotNetty.Transport.Channels;
+using JSMachine.WMS.Infrastructure.Helper;
+using JSMachine.WMS.Netty.NettyAsClient.Command;
+using Newtonsoft.Json;
+using Newtonsoft.Json.Linq;
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace JSMachine.WMS.Netty.NettyAsClient.Codec
+{
+    public class Decoder : ByteToMessageDecoder
+    {
+        protected override void Decode(IChannelHandlerContext context, IByteBuffer input, List<object> output)
+        {
+            if (input != null)
+            {
+                int length = input.ReadableBytes;
+                byte[] bytes = new byte[length];
+
+                string msg = Encoding.UTF8.GetString(bytes);
+
+                LogHelper.Info(msg);
+
+                output.Add(input);
+
+                try
+                {
+                    //IMessagePackage message = MessagePackageMange.Instance.BinaryDeserializeToObject(bytes);
+                    //IMessagePackage message = JsonConvert.DeserializeObject<IMessagePackage>(Encoding.UTF8.GetString(bytes));
+
+                    string json = Encoding.UTF8.GetString(bytes);
+                    JObject jobj = JObject.Parse(json);
+                    //当找不到这个节点的值的时候则自动赋0，因此，命令类型不可以是0，否则无法解析的命令都自动归为0类型了
+                    //Common.MessagePack.CommandType cmdType = (Common.MessagePack.CommandType)jobj.Value<int>("ComType");
+
+                    //switch (cmdType)
+                    //{
+                    //    case Common.MessagePack.CommandType.g_eKeepAlive:
+                    //        baseCmd = JsonConvert.DeserializeObject<HeartBeartCmd>(json);
+                    //        break;
+                    //    default:
+                    //        break;
+                    //}
+
+                    //output.Add(baseCmd);
+                }
+                catch (Exception ex)
+                {
+                    LogHelper.Error($"解码远程地址：{context.Channel.RemoteAddress}----发送的数据异常");
+                }
+
+                input.Clear();
+            }
+        }
+    }
+}
@@ -0,0 +1,39 @@
+using DotNetty.Buffers;
+using DotNetty.Codecs;
+using DotNetty.Transport.Channels;
+using JSMachine.WMS.Infrastructure.Helper;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JSMachine.WMS.Netty.NettyAsClient.Codec
+{
+    public class EnCoderPrint : MessageToByteEncoder<byte[]>
+    {
+        private string _encoderWord;
+
+        public EnCoderPrint() : base() { }
+
+        public EnCoderPrint(string encoderWord)
+        {
+            _encoderWord = encoderWord;
+        }
+        protected override void Encode(IChannelHandlerContext context, byte[] message, IByteBuffer output)
+        {
+            try
+            {
+                byte[] body = message;
+                //byte[] tail = Encoding.UTF8.GetBytes(_encoderWord);
+
+                output.WriteBytes(body);
+                //output.WriteBytes(tail);
+            }
+            catch (Exception ex)
+            {
+                LogHelper.Error($"TCP编码异常,异常信息 {ex.Message}");
+            }
+        }
+    }
+}
@@ -0,0 +1,39 @@
+using DotNetty.Buffers;
+using DotNetty.Codecs;
+using DotNetty.Transport.Channels;
+using JSMachine.WMS.Infrastructure.Helper;
+using JSMachine.WMS.Netty.NettyAsClient.Command;
+using Newtonsoft.Json;
+using System;
+using System.Text;
+
+namespace JSMachine.WMS.Netty.NettyAsClient.Codec
+{
+    public class Encoder : MessageToByteEncoder<string>
+    {
+        private string _encoderWord;
+
+        public Encoder():base() { }
+
+        public Encoder(string encoderWord)
+        {
+            _encoderWord= encoderWord;
+        }
+        
+        protected override void Encode(IChannelHandlerContext context, string message, IByteBuffer output)
+        {
+            try
+            {
+                byte[] body = Encoding.UTF8.GetBytes(message);
+                byte[] tail = Encoding.UTF8.GetBytes(_encoderWord);
+
+                output.WriteBytes(body);
+                output.WriteBytes(tail);
+            }
+            catch (Exception ex)
+            {
+                LogHelper.Error($"TCP编码异常,异常信息 {ex.Message}");
+            }
+        }
+    }
+}
@@ -0,0 +1,16 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JSMachine.WMS.Netty.NettyAsClient.Command
+{
+    /// <summary>
+    /// 请求客户端读取条码的命令，消息标识为 T。
+    /// </summary>
+    public class BarCodeReadCmd : BaseCmd
+    {
+        public override string Msg { get; set; } = "T";
+    }
+}
@@ -0,0 +1,16 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JSMachine.WMS.Netty.NettyAsClient.Command
+{
+    /// <summary>
+    /// 条码读取异常命令
+    /// </summary>
+    public class BarCodeReadExceptionCmd : BaseCmd
+    {
+
+    }
+}
@@ -0,0 +1,10 @@
+namespace JSMachine.WMS.Netty.NettyAsClient.Command
+{
+    /// <summary>
+    /// 条码上传命令
+    /// </summary>
+    public class BarCodeUploadCmd : BaseCmd
+    {
+
+    }
+}
@@ -0,0 +1,19 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JSMachine.WMS.Netty.NettyAsClient.Command
+{
+    /// <summary>
+    /// Netty 客户端命令基类，定义待发送或处理的原始消息内容。
+    /// </summary>
+    public class BaseCmd
+    {
+        /// <summary>
+        /// 命令消息正文。
+        /// </summary>
+        public virtual string Msg { get; set; }
+    }
+}
@@ -0,0 +1,12 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace JSMachine.WMS.Netty.NettyAsClient.Command
+{
+    public class PenmoReadCmd : BaseCmd
+    {
+    }
+}
@@ -0,0 +1,41 @@
+using JSMachine.WMS.Common.Cache.Model;
+using JSMachine.WMS.Common.Cache;
+using System;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Caching.Memory;
+using JSMachine.WMS.Netty.NettyAsClient.Command;
+using JSMachine.WMS.Infrastructure.Helper;
+
+namespace JSMachine.WMS.Netty.NettyAsClient.CommandHandler
+{
+    public class BarCodeCmdHandlerL : BaseCmdhandler
+    {
+        public async override Task<bool> Handle(BaseCmd msg)
+        {
+            return await Task.Run(() =>
+             {
+                 if (msg is BarCodeUploadCmd barCodeUploadCmd)
+                 {
+                     GlobalMemoryCache.AddCach(
+                         RemoteIPInfo.IP,
+                         new BarCodeInfo
+                         {
+                             AddTime = DateTime.Now,
+                             BarCode = barCodeUploadCmd.Msg,
+                             IP = RemoteIPInfo.IP
+                         },
+                         new MemoryCacheEntryOptions
+                         {
+                             AbsoluteExpirationRelativeToNow = TimeSpan.FromSeconds(30),
+                         });
+
+                     LogHelper.Info($"收到条码信息({RemoteIPInfo.IP}): 【{barCodeUploadCmd.Msg}】");
+
+                     return true;
+                 }
+
+                 return false;
+             });
+        }
+    }
+}
@@ -0,0 +1,29 @@
+using JSMachine.WMS.Infrastructure.Helper;
+using JSMachine.WMS.Netty.NettyAsClient.Command;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using static System.Console;
+using System.Threading.Tasks;
+
+namespace JSMachine.WMS.Netty.NettyAsClient.CommandHandler
+{
+    public class BarCodeReadExceptionCmdHandler : BaseCmdhandler
+    {
+        public async override Task<bool> Handle(BaseCmd msg)
+        {
+            return await Task.Run(() =>
+            {
+                if (msg is BarCodeReadExceptionCmd barCodeReadExceptionCmd)
+                {
+                    LogHelper.Warn($"相机【{base.RemoteIPInfo.IP}】条码读取失败，返回信息 {barCodeReadExceptionCmd.Msg}");
+
+                    return true;
+                }
+
+                return false;
+            });
+        }
+    }
+}
@@ -0,0 +1,36 @@
+using DotNetty.Transport.Channels;
+using JSMachine.WMS.Netty.Common.Util;
+using JSMachine.WMS.Netty.NettyAsClient.Command;
+using Prism.Events;
+using System.ComponentModel.Composition;
+using System.Threading.Tasks;
+
+namespace JSMachine.WMS.Netty.NettyAsClient.CommandHandler
+{
+    /// <summary>
+    /// Netty 客户端命令处理器基类，提供通道上下文及收发端地址信息。
+    /// </summary>
+    [InheritedExport("NettyAsClient")]
+    public abstract class BaseCmdhandler
+    {
+        /// <summary>
+        /// 当前命令所属的 Netty 通道上下文。
+        /// </summary>
+        public IChannelHandlerContext Channel { get; set; }
+        /// <summary>
+        /// 远端地址信息。
+        /// </summary>
+        public IPInfo RemoteIPInfo => Channel.ToIPInfo(AddressOnwer.Remote);
+        /// <summary>
+        /// 本地地址信息。
+        /// </summary>
+        public IPInfo LocalIPInfo=> Channel.ToIPInfo(AddressOnwer.Local);
+        /// <summary>
+        /// 处理一条客户端命令。
+        /// </summary>
+        /// <param name="msg">待处理的命令。</param>
+        /// <returns>处理成功返回 <see langword="true"/>。</returns>
+        public abstract Task<bool> Handle(BaseCmd msg);
+
+    }
+}
@@ -0,0 +1,94 @@
+using DotNetty.Buffers;
+using DotNetty.Handlers.Timeout;
+using DotNetty.Transport.Channels;
+using JSMachine.WMS.Infrastructure.Helper;
+using JSMachine.WMS.Netty.Common.Model;
+using JSMachine.WMS.Netty.NettyAsClient.Transport;
+using Prism.Events;
+using Prism.Ioc;
+using System;
+
+namespace JSMachine.WMS.Netty.NettyAsClient
+{
+    public class EchoClientChannelHandler : ChannelHandlerAdapter
+    {
+        private NettyClientTransport _nettyTransport;
+        private TcpEngineInitParam _tcpEngineInitParam1;
+       IEventAggregator _eventAggregator= ContainerLocator.Container.Resolve<IEventAggregator>();
+
+        public EchoClientChannelHandler() : base() { }
+
+        public EchoClientChannelHandler(TcpEngineInitParam tcpEngineInitParam)
+        {
+            _tcpEngineInitParam1= tcpEngineInitParam;
+        }
+
+        public override async void ChannelActive(IChannelHandlerContext context)
+        {
+            _nettyTransport = new NettyClientTransport(context, _tcpEngineInitParam1);
+            TransportManager.DicClientChannels.AddOrUpdate(_tcpEngineInitParam1.IP, p => context, (s, p) => context);
+
+            base.ChannelActive(context);
+
+            //开启喷墨打印机
+            if(_tcpEngineInitParam1.DeviceType== DeviceType.PenmoPrint)
+            LogHelper.Info("已成功连接服务端");
+        }
+
+        public override void ChannelRead(IChannelHandlerContext context, object message)
+        {
+            //LogsTool.Default.Info("收到服务端TCP信息");
+            if (message is IByteBuffer byteBuffer)
+                _nettyTransport.ChannelReadHandle(byteBuffer);
+        }
+
+        public override void ChannelReadComplete(IChannelHandlerContext context) => context.Flush();
+
+        public override void ExceptionCaught(IChannelHandlerContext context, Exception exception)
+        {
+            //LogsTool.Default.Error("Exception: " + exception);
+            context.CloseAsync();
+        }
+
+        public override void HandlerAdded(IChannelHandlerContext context)
+        {
+            base.HandlerAdded(context);
+        }
+
+        public override void HandlerRemoved(IChannelHandlerContext context)
+        {
+            LogHelper.Warn($"跟服务端的连接断开");
+            base.HandlerRemoved(context);
+        }
+
+        public override void UserEventTriggered(IChannelHandlerContext context, object evt)
+        {
+            IdleStateEvent eventState = evt as IdleStateEvent;
+
+            //if (eventState.State == IdleState.ReaderIdle)
+            //{
+            //    //没有任何读取
+            //    Reconnect();
+            //}
+            //else if (eventState.State == IdleState.WriterIdle)
+            //{
+            //    //没有任何写入
+            //    Reconnect();
+            //}
+            //else 
+            if (eventState.State == IdleState.AllIdle)
+            {
+                //没有任何交互
+                Reconnect();
+            }
+
+            base.UserEventTriggered(context, evt);
+
+            void Reconnect()
+            {
+                //context.CloseAsync();
+                //new NettyEngine().InitEngine(NettyTransport.RemoteIpInfo.IP, NettyTransport.RemoteIpInfo.Port);
+            }
+        }
+    }
+}
@@ -0,0 +1,100 @@
+using DotNetty.Buffers;
+using DotNetty.Codecs;
+using DotNetty.Transport.Bootstrapping;
+using DotNetty.Transport.Channels;
+using DotNetty.Transport.Channels.Sockets;
+using JSMachine.WMS.Infrastructure.Helper;
+using JSMachine.WMS.Netty.Common.Model;
+using System;
+using System.Net;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace JSMachine.WMS.Netty.NettyAsClient
+{
+    /// <summary>
+    /// TCP客户端引擎
+    /// </summary>
+    public class TcpClientEngine
+    {
+        private Bootstrap SocketBootstrap = new();
+        private MultithreadEventLoopGroup WorkGroup = new();
+        private IChannel channel;
+
+        /// <summary>
+        /// 初始化 TCP 客户端编解码管道并启动后台连接监视。
+        /// </summary>
+        /// <param name="tcpEngineInitParam">服务端地址、端口及协议编解码配置。</param>
+        public async Task InitEngine(TcpEngineInitParam tcpEngineInitParam)
+        {
+            LogHelper.Info("正在启动TCP客户端引擎-----------");
+
+            SocketBootstrap
+                .Group(WorkGroup)
+                .Channel<TcpSocketChannel>()
+                .Option(ChannelOption.TcpNodelay, true)
+                .Option(ChannelOption.ConnectTimeout, TimeSpan.FromSeconds(2))
+                .Option(ChannelOption.SoKeepalive, true)
+                //以下两种方式都是设置一次性能接收的数据包大小
+                //.Option(ChannelOption.SoRcvbuf, 4 * 1024)
+                .Option(ChannelOption.RcvbufAllocator, new FixedRecvByteBufAllocator(25))
+                .Handler(new ActionChannelInitializer<ISocketChannel>(channel =>
+                {
+                    IChannelPipeline pipeline = channel.Pipeline;
+
+                    if (!string.IsNullOrEmpty(tcpEngineInitParam.DecoderWord))
+                    {
+                        IByteBuffer delimiter = Unpooled.WrappedBuffer(Encoding.UTF8.GetBytes("\r"));
+                        //简单的协议可以直接使用自带的分隔符解码器
+                        pipeline.AddLast("framing-dec", new DelimiterBasedFrameDecoder(32, true, delimiter));
+                    }
+                    else
+                    {
+                        //pipeline.AddLast("framing-enc", new LengthFieldPrepender(2));
+                        pipeline.AddLast("framing-dec", new LengthFieldBasedFrameDecoder(ushort.MaxValue, 0, 2, 0, 2));
+                    }
+                    if (!string.IsNullOrEmpty(tcpEngineInitParam.EncoderWord))
+                    {
+                        //自定义编码器，编解码器必须写在处理器之前，否则数据包将不会经过编解码器
+                        pipeline.AddLast(new JSMachine.WMS.Netty.NettyAsClient.Codec.Encoder(tcpEngineInitParam.EncoderWord));
+                    }
+                    else
+                    {
+                        pipeline.AddLast(new JSMachine.WMS.Netty.NettyAsClient.Codec.EnCoderPrint(tcpEngineInitParam.EncoderWord));
+                    }
+
+
+                    pipeline.AddLast("echo", new EchoClientChannelHandler(tcpEngineInitParam));
+                }));
+
+            await MonitorTcpConnection(tcpEngineInitParam.IP, tcpEngineInitParam.Port);
+        }
+
+        private async Task MonitorTcpConnection(string serverIP, int port)
+        {
+            // 连接断开后持续重试，保证扫码设备临时掉线时能够自动恢复通信。
+           await Task.Run(async () => 
+            {
+                while (true)
+                {
+                    if (channel == null || !channel.Active)
+                    {
+                        try
+                        {
+                            channel = await SocketBootstrap.ConnectAsync(new IPEndPoint(IPAddress.Parse(serverIP), port));
+                        }
+                        catch (Exception ex)
+                        {
+                            LogHelper.Error($"连接服务端 {serverIP} 失败,错误信息 {ex.Message} 即将重新连接");
+                        }
+                    }
+                    Thread.Sleep(5000);
+                }
+
+            });
+
+            
+        }
+    }
+}
@@ -0,0 +1,84 @@
+using DotNetty.Buffers;
+using DotNetty.Transport.Channels;
+using JSMachine.WMS.Infrastructure.Helper;
+using JSMachine.WMS.Netty.Common.Model;
+using JSMachine.WMS.Netty.Common.Util;
+using JSMachine.WMS.Netty.NettyAsClient.Command;
+using JSMachine.WMS.Netty.NettyAsClient.CommandHandler;
+using System;
+using System.Collections.Generic;
+using System.ComponentModel.Composition;
+using System.ComponentModel.Composition.Hosting;
+using System.Reflection;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace JSMachine.WMS.Netty.NettyAsClient.Transport
+{
+    public class NettyClientTransport
+    {
+        private IChannelHandlerContext ChannelHandlerContext;
+        private IPInfo RemoteIpInfo;
+        private TcpEngineInitParam _tcpEngineInitParam;
+
+        [ImportMany("NettyAsClient")]
+        private List<BaseCmdhandler> AllCmds { get; set; }
+
+        private Lazy<CompositionContainer> _Container = new(() =>
+        {
+            AggregateCatalog catalog = new();
+            catalog.Catalogs.Add(new AssemblyCatalog(Assembly.GetExecutingAssembly()));
+
+            return new CompositionContainer(catalog);
+        });
+
+        public NettyClientTransport(IChannelHandlerContext channel, TcpEngineInitParam tcpEngineInitParam)
+        {
+            _tcpEngineInitParam= tcpEngineInitParam;
+
+            _Container.Value.ComposeParts(this);
+
+            ChannelHandlerContext = channel;
+
+            foreach (BaseCmdhandler cmd in AllCmds)
+            {
+                cmd.Channel = channel;
+            }
+
+            RemoteIpInfo = channel.ToIPInfo(AddressOnwer.Remote);
+        }
+
+        public Task ChannelReadHandle(IByteBuffer byteBuffer)
+        {
+            string msg = byteBuffer.ToString(Encoding.UTF8);
+
+            //int length = byteBuffer.ReadableBytes;
+            //byte[] array = new byte[length];
+            //byteBuffer.GetBytes(byteBuffer.ReaderIndex, array);
+
+            //string msg = Encoding.UTF8.GetString(array);
+
+            return Task.Run(async () =>
+            {
+                foreach (BaseCmdhandler cmd in AllCmds)
+                {
+                    BaseCmd baseCmd;
+                    if (msg.Length < _tcpEngineInitParam.MinCodeLength)
+                        baseCmd = new BarCodeReadExceptionCmd() { Msg = msg };
+                    else
+                    {
+                        if (_tcpEngineInitParam.DeviceType != DeviceType.PenmoPrint)
+                            baseCmd = new BarCodeUploadCmd() { Msg = msg };
+                        else
+                            baseCmd = new PenmoReadCmd() { Msg = msg };
+                        //baseCmd = new BarCodeUploadCmd() { Msg = msg };
+                    }
+                    bool isMsgHasBeenHandled = await cmd.Handle(baseCmd);
+                    if (isMsgHasBeenHandled)
+                        break;
+                }
+            });
+        }
+    }
+}
@@ -0,0 +1,119 @@
+using DotNetty.Transport.Channels;
+using JSMachine.WMS.Infrastructure.Helper;
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Data;
+using System.Linq;
+using System.Reactive.Linq;
+using System.Threading.Tasks;
+
+namespace JSMachine.WMS.Netty.NettyAsClient.Transport
+{
+    public static class TransportManager
+    {
+        /// <summary>
+        /// 已连接的客户端
+        /// key-ip 一个客户端仅允许一个连接
+        /// </summary>
+
+        public static readonly ConcurrentDictionary<string, IChannelHandlerContext> DicClientChannels = new();
+
+        /// <summary>
+        /// 移除有问题通道
+        /// </summary>
+        /// <param name="channel"></param>
+        public static void RemoveInActiveChannel(IChannelHandlerContext channel)
+        {
+            try
+            {
+                DicClientChannels
+               .Where(p => p.Value == channel)
+               .Select(p => p.Key)
+               .ToList()
+               ?.ForEach(p => DicClientChannels.Remove(p, out IChannelHandlerContext channelHandlerContext));
+            }
+            catch (Exception ex)
+            {
+                LogHelper.Error($"移除有问题通道出错: {ex.Message} \n {ex.StackTrace}");
+            }
+        }
+
+        /// <summary>
+        /// 指定客户端发送消息
+        /// </summary>
+        /// <param name="ip"></param>
+        /// <param name="msg"></param>
+        /// <returns></returns>
+        public async static Task<bool> SendMsg(string ip, string msg)
+        {
+            try
+            {
+                if (!DicClientChannels.ContainsKey(ip))
+                {
+                    LogHelper.Error($"{ip}连接失败");
+                    return false;
+                }
+
+                IChannelHandlerContext channel = DicClientChannels[ip];
+
+                await channel.Channel.WriteAndFlushAsync(msg);
+                return true;
+            }
+            catch (Exception ex)
+            {
+                LogHelper.Error($"发送消息出错: {ex.Message} \n {ex.StackTrace}");
+            }
+
+            return false;
+        }
+
+        /// <summary>
+        /// 指定客户端发送byte数组
+        /// </summary>
+        /// <param name="ip"></param>
+        /// <param name="msg"></param>
+        /// <returns></returns>
+        public async static Task<bool> SendMsgByte(string ip, byte[] msg)
+        {
+            try
+            {
+                if (!DicClientChannels.ContainsKey(ip))
+                {
+                    LogHelper.Error($"{ip}连接失败");
+                    return false;
+                }
+
+                IChannelHandlerContext channel = DicClientChannels[ip];
+
+                await channel.Channel.WriteAndFlushAsync(msg);
+                return true;
+            }
+            catch (Exception ex)
+            {
+                LogHelper.Error($"发送消息出错: {ex.Message} \n {ex.StackTrace}");
+            }
+
+            return false;
+        }
+
+        /// <summary>
+        /// 发送广播消息
+        /// </summary>
+        /// <param name="messagePackage"></param>
+        public static void BroadcastMsg(string brodCastMsg)
+        {
+            try
+            {
+                DicClientChannels.Keys.ToList().ForEach(async key =>
+                {
+                    await DicClientChannels[key].Channel.WriteAndFlushAsync(brodCastMsg);
+                });
+            }
+            catch (Exception ex)
+            {
+                LogHelper.Error($"发送广播消息出错: {ex.Message} \n {ex.StackTrace}");
+            }
+        }
+    }
+}