DotNetty 封装的 UdpClient

DotNetty 资料较少，UdpClient 和 TcpClient 略有不同

public class UdpCommunicator : ICommunicator
{
    private IChannel? _ClientChannel;
    private Bootstrap? _Bootstrap;
    IEventLoopGroup? _LoopGroup;
    private TaskCompletionSource<byte[]> _ResponseCompletionSource;
    CancellationTokenSource _TokenSource;
    NettyDataReceiveHandle _NettyDataReceiveHandle = new NettyDataReceiveHandle();
    public IPEndPoint? LocalEndPoint { get; private set; }
    public IPEndPoint? RemoteEndPoint { get; private set; }

    public bool IsConnected { get; private set; }

    public event Action<byte[]> OnDataReceive;
    internal UdpCommunicator()
    {
        LocalEndPoint = new IPEndPoint(IPAddress.Any, 0);
        InitNetty();
    }
    /// <summary>
    /// 用于绑定的本地端点
    /// </summary>
    /// <param name="iPEndPoint"></param>
    internal UdpCommunicator(IPEndPoint iPEndPoint)
    {
        LocalEndPoint = iPEndPoint;
        InitNetty();
    }
    void InitNetty()
    {
        _NettyDataReceiveHandle.OnBytesReceived += receiveBytes => Task.Run(() =>
        {
            OnDataReceive?.Invoke(receiveBytes);
            _ResponseCompletionSource?.TrySetResult(receiveBytes);
        });
        _NettyDataReceiveHandle.OnDisconnected += () => IsConnected = false;

        _LoopGroup = new MultithreadEventLoopGroup();
        
    }

    public bool Connect(IPEndPoint remoteEndPoint)
    {
        try
        {

            RemoteEndPoint = remoteEndPoint;

            _Bootstrap = new Bootstrap();
            _Bootstrap
                .Group(_LoopGroup)
                .Channel<SocketDatagramChannel>()
                .Option(ChannelOption.SoBroadcast, true)
                .Handler(new ActionChannelInitializer<IChannel>(channel =>
                {
                    IChannelPipeline pipeline = channel.Pipeline;
                    pipeline.AddLast(_NettyDataReceiveHandle);
                }));

            _ClientChannel = _Bootstrap.BindAsync(LocalEndPoint).Result;
            return true;
        }
        catch (Exception ex)
        {
            return false;
        }

    }

    public bool DisConnect()
    {
        _ClientChannel?.CloseAsync().Wait();
        _ClientChannel = null;
        return true;
    }

    public void Dispose()
    {
        DisConnect();
        _LoopGroup?.ShutdownGracefullyAsync(TimeSpan.FromMilliseconds(100), TimeSpan.FromSeconds(1)).Wait();
        _LoopGroup = null;
    }
    object sendLock = new object();
    public byte[]? Send(byte[]? content)
    {
        lock (sendLock)
        {
            _ResponseCompletionSource = new TaskCompletionSource<byte[]>();
            using (_TokenSource = new CancellationTokenSource(600))
            {
                _TokenSource.Token.Register(() => _ResponseCompletionSource.TrySetCanceled(), useSynchronizationContext: false);
                bool sendResult = SendAsync(content).Result;
                if(!sendResult) return null;
                try
                {
                    return _ResponseCompletionSource.Task.Result;
                }
                catch (TaskCanceledException)
                {
                    _ResponseCompletionSource = null;
                    return null;
                }
                catch (AggregateException)
                {
                    _ResponseCompletionSource = null;
                    return null;
                }
            }
        }
    }

    public async Task<bool> SendAsync(byte[]? content)
    {
        if(_ClientChannel is null) return false;
        IByteBuffer byteBuffer = Unpooled.WrappedBuffer(content);
        await _ClientChannel.WriteAndFlushAsync(new DatagramPacket(byteBuffer, RemoteEndPoint));
        return true;
    }
}
class NettyDataReceiveHandle : SimpleChannelInboundHandler<DatagramPacket>
{
    public override bool IsSharable => true;
    public event Func<byte[], Task>? OnBytesReceived;
    public event Action? OnDisconnected;
    protected override void ChannelRead0(IChannelHandlerContext ctx, DatagramPacket packet)
    {
        var byteBuffer = packet.Content;
        byte[] bytes = new byte[byteBuffer.ReadableBytes];
        byteBuffer.GetBytes(byteBuffer.ReaderIndex, bytes);
        OnBytesReceived?.Invoke(bytes);
    }
    public override void ChannelReadComplete(IChannelHandlerContext ctx)
    {
        ctx.Flush();
    }

    public override void ExceptionCaught(IChannelHandlerContext ctx, Exception e)
    {
        ctx.CloseAsync();
        OnDisconnected?.Invoke();
    }
}