using DotNetty.Buffers;
using DotNetty.Codecs;
using DotNetty.Codecs.Protobuf;
using DotNetty.Common.Concurrency;
using DotNetty.Handlers.Timeout;
using DotNetty.Transport.Bootstrapping;
using DotNetty.Transport.Channels;
using DotNetty.Transport.Channels.Sockets;
using System;
using System.Collections;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
namespace Module.Socket.Tool
{
public class TcpClientChargerTool
{
#region 定义锁
private object lockObj = new object(); //线程同步锁
#endregion 定义锁
#region 字段属性
private string _client_ip;
///
/// 客户端IP地址
///
public string F_ClientIP
{
get { return _client_ip; }
set
{
lock (lockObj)
{
_client_ip = value;
}
}
}
private int _client_port;
///
/// 客户端端口号
///
public int F_ClientPort
{
get { return _client_port; }
set
{
lock (lockObj)
{
_client_port = value;
}
}
}
public event EventHandler ConnectedStatusChanged;
public event EventHandler DataSended;
public event EventHandler ConnectLogEvent;
#endregion
private AutoResetEvent ChannelInitilizedEvent = new AutoResetEvent(false);
private Bootstrap SocketBootstrap = new Bootstrap();
private MultithreadEventLoopGroup WorkGroup = new MultithreadEventLoopGroup();
public volatile bool Connected = false;
public volatile IChannel NettyChannel;
public Channel messageQueue;
#region 类结构体
public TcpClientChargerTool()
{
UnboundedChannelOptions options = new UnboundedChannelOptions();
options.SingleReader = true;
messageQueue = Channel.CreateUnbounded(options);
}
public TcpClientChargerTool(string ipAddr, int port)
{
UnboundedChannelOptions options = new UnboundedChannelOptions();
options.SingleReader = true;
messageQueue = Channel.CreateUnbounded(options);
_client_ip = ipAddr;
_client_port = port;
}
#endregion 类结构体
private void InitBootstrap()
{
IByteBuffer delimiter = Unpooled.CopiedBuffer(new byte[] { 0x68 ,0xEE});
SocketBootstrap = new Bootstrap();
SocketBootstrap.Group(WorkGroup)
.Channel()
.Option(ChannelOption.TcpNodelay, true)
.Option(ChannelOption.SoKeepalive, true)
.Handler(new ActionChannelInitializer(channel =>
{
IChannelPipeline pipeline = channel.Pipeline;
// 在管道中添加 DelimiterBasedFrameDecoder,指定分隔符
//pipeline.AddLast(new DelimiterBasedFrameDecoder(2048, delimiter));
//pipeline.AddLast(new CustomFrameDecoder1());
//pipeline.AddLast(new CustomFrameDecoder2(new IByteBuffer[]{ delimiter }, false,false ));//效果不好
//pipeline.AddLast(new CustomFrameDecoder3());
pipeline.AddLast(new CustomFrameDecoder4(new IByteBuffer[] { delimiter }, false, false));//效果不好
pipeline.AddLast("idleStateHandler", new IdleStateHandler(30, 0, 0)); // 触发读取超时
pipeline.AddLast(new ReconnectHandler(this));
pipeline.AddLast(new ClientHandler(this));
}));
}
public async Task Connect()
{
await DoConnect();
}
public void Disconnect()
{
WorkGroup.ShutdownGracefullyAsync(TimeSpan.FromMilliseconds(100), TimeSpan.FromSeconds(1));
}
public async Task Write(byte[] message)
{
try
{
await NettyChannel.WriteAndFlushAsync(Unpooled.WrappedBuffer(message));
}
catch (Exception e)
{
Console.WriteLine($"{_client_ip} Error writing message: {e.Message}");
}
}
private async Task DoConnect()
{
Connected = false;
SetConnectStatusEvent("连接初始化");
do
{
try
{
InitBootstrap();
var clientChannel =
await SocketBootstrap.ConnectAsync(new IPEndPoint(IPAddress.Parse(_client_ip), _client_port));
NettyChannel = clientChannel;
if (clientChannel.Open)
{
Connected = true;
SetConnectStatusEvent("连接成功");
}
else
{
Connected = false;
SetConnectStatusEvent("连接失败");
await Task.Delay(5 * 1000);
}
//ChannelInitilizedEvent.Set();
await Task.Delay(200);
}
catch (Exception ce)
{
SetConnectStatusEvent(ce.StackTrace);
ConnectLogEventHandle( ce.StackTrace);
Debug.Print("连接失败:" + ce.StackTrace);
Debug.Print("connect fail ,reconnect after 5 seconds..." + _client_ip);
await Task.Delay(5 * 1000);
}
} while (!Connected);
}
///
/// 设置连接状态字符串
///
/// 连接状态字符串。比如:“连接成功”
private void SetConnectStatusEvent(string status)
{
string strClient = "";
if (NettyChannel != null && NettyChannel.RemoteAddress != null)
{
strClient = NettyChannel.RemoteAddress.ToString();
}
string strTime = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss") + " ";
string strConnectLog = strTime + "客户端" + strClient + "与服务端" + _client_ip.ToString() + ":" +
_client_port.ToString() + " " + status + "!";
EventHandler ConnectedHandler;
ConnectedHandler = ConnectedStatusChanged;
if (ConnectedHandler != null)
{
ConnectedHandler.Invoke(this,
new ViewLogEventArgs() { ConnectedStatus = Connected, IsContent = strConnectLog });
}
}
internal void ConnectLogEventHandle(string message)
{
if (ConnectLogEvent != null)
{
ConnectLogEvent.Invoke(this, new ViewLogEventArgs() { IsContent = message });
}
}
}
public class CustomFrameDecoder : ByteToMessageDecoder
{
protected override void Decode(IChannelHandlerContext context, IByteBuffer input, List