server/CSserver/clientTest/Program_Chat.cs

using System;
using System.IO;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Channels;
using System.Linq;
using System.Net.Http;
using System.Collections.Generic;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using System.Net;
using System.Net.Sockets;
using System.Diagnostics;
using ProtoDataBuff;
using Chat;
using pb = global::Google.Protobuf;

namespace clientTest.chat
{
    public static class TaskWaitingExtensions
    {
        public static async Task<TResult> WaitAsync<TResult>(this Task<TResult> task, TimeSpan timeout)
        {
            using (var timeoutCancellationTokenSource = new CancellationTokenSource())
            {
                var delayTask = Task.Delay(timeout, timeoutCancellationTokenSource.Token);
                if (await Task.WhenAny(task, delayTask) == task)
                {
                    timeoutCancellationTokenSource.Cancel();
                    return await task;
                }
                throw new TimeoutException("The operation has timed out.");
            }
        }
        public static async Task WaitAsync(this Task task, TimeSpan timeout)
        {
            using (var timeoutCancellationTokenSource = new CancellationTokenSource())
            {
                var delayTask = Task.Delay(timeout, timeoutCancellationTokenSource.Token);
                if (await Task.WhenAny(task, delayTask) == task)
                {
                    timeoutCancellationTokenSource.Cancel();
                    return;
                }
                throw new TimeoutException("The operation has timed out.");
            }
        }
    }
    class Program
    {
        static readonly string[] names = new string[] { "你猜(尼采)", "黑哥儿(黑格尔)", "孟德四舅" };
        static readonly string[] texts = new string[] {
            "一个人可以失败很多次，但是只要他没有开始责怪旁人，他还不是一个失败者。",
            "这个世界既不是有钱人的世界，也不是有权人的世界，它是有心人的世界。",
            "伟人之所以伟大，是因为他与别人共处逆境时，别人失去了信心，他却下决心实现自己的目标。",
            "世上没有绝望的处境，只有对处境绝望的人。 ",
            "时间就像一张网,你撒在哪里,收获就在哪里。",
            "心若计较，处处都是怨言；心不计较，时时都是晴天。",
            "能使我们感觉快乐的，不是环境，而是态度。",
            "学问是用来实践的，不是拿来用嘴说的。",
            "自己打败自己是最可悲的失败,自己战胜自己是最可贵的胜利。",
            "我们得成功，其实成功有一条很简单的定律：“只要站起来的次数比被击倒的次数多一次就行。",
            " 弱者坐待良机，强者创造时机。",
            " 没有不会做的事, 只有不想做的事。",
            "任何的限制, 都是从自己内心开始的。",
            "只要还有明天, 今日就永远是起跑线。",
            "既然认准一条道路, 何必去打听要走多久。",
            "现在站在什么地方不重要，重要的是你往什么方向移动？",
            "如果什么都想要，只会什么都得不到。"

        };
        static readonly Random r = new Random();

        /// <summary>
        /// 消息分发
        /// </summary>
        static private Dictionary<eProtocalCommand, Action<sSocketData>> callbacks = new Dictionary<eProtocalCommand, Action<sSocketData>>();
        static void Main(string[] args)
        {

            callbacks.Add(eProtocalCommand.ScChatNewMsg, On_update);
            callbacks.Add(eProtocalCommand.ScChatLogin, On_LoginOver);
            var n = 10;
            var list = new Task[n];
            for (int i = 0; i < n; i++)
            {
                list[i] = Task.Run(async () => await send());
            }
            Task.Run(Dispatch);

            Task.WaitAll(list);
        }

        static string ChannelName(ChatChannel c) => c switch
        {
            ChatChannel.System => "系统",
            ChatChannel.World => "世界",
            ChatChannel.Guild => "公会",
            _ => "x",
        };

        /// <summary>
        /// 处理客户端上报伤害请求
        /// </summary>
        static void On_update(sSocketData data)
        {
            var msg = SC_ChatNewMsg.Parser.ParseFrom(data._data);

            Console.WriteLine($"[{ChannelName(msg.FromChannel)}]-[{msg.SenderName}] : " + msg.Msg);
        }

        /// <summary>
        /// 处理客户端上报伤害请求
        /// </summary>
        static void On_LoginOver(sSocketData data)
        {
            var msg = SC_ChatLogin.Parser.ParseFrom(data._data);
            Console.WriteLine("登录结果: " + msg.Code);
            //Task.Delay(3000).ContinueWith(t => Environment.Exit(0));
        }
        static async Task send()
        {

            var port = 6000;

            var endPoint = new IPEndPoint(IPAddress.Parse("192.168.10.17"), port);
            //var endPoint = new IPEndPoint(IPAddress.Loopback, port);
            //var endPoint = new IPEndPoint(IPAddress.Parse("115.159.121.129"), port);

            using (var client = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
            {
                try
                {
                    await client.ConnectAsync(endPoint);
                }
                catch (Exception ee)
                {
                    Debug.WriteLine(ee.Message);
                }
                var t = Task.Run(() => recv(client));
                await login(client);
                var i = 0;
                while (i++ < 10)
                {
                    try
                    {
                        await sendMsg(client);
                        Thread.Sleep(r.Next(1000, 1100));
                    }
                    catch (Exception e)
                    {
                        client.Close();
                        break;
                    }
                }
            }
        }

        async static Task login(Socket Sock)
        {
            try
            {
                var msg = new CS_ChatLogin() { Uid = Guid.NewGuid().ToString(), Zoneid = 1, Name = names[r.Next(0, 3)] + r.Next() };
                var data = SocketDataToBytes(BytesToSocketData(eProtocalCommand.CsChatLogin, IMsg2Bytes(msg)));
                await Sock.SendAsync(new ArraySegment<byte>(data), SocketFlags.None);
            }
            catch (Exception e)
            {
                Sock.Close();
            }
        }

        async static Task sendMsg(Socket sock)
        {
            var c = (ChatChannel)r.Next(0, 3);

            var info = texts[r.Next(0, texts.Length)];
            var msg = new CS_ChatSendMsg() { ToChannel = c, Msg = info };

            var data = SocketDataToBytes(BytesToSocketData(eProtocalCommand.CsChatSendMsg, IMsg2Bytes(msg)));
            await sock.SendAsync(new ArraySegment<byte>(data), SocketFlags.None);
        }

        static byte[] IMsg2Bytes(pb::IMessage msg)
        {
            using var ms = new MemoryStream();
            using var goutstream = new pb::CodedOutputStream(ms);
            msg.WriteTo(goutstream);
            goutstream.Flush();
            ms.Seek(0, SeekOrigin.Begin);
            return ms.ToArray();
        }



        static async void Dispatch()
        {
            while (true)
            {
                var msg = await recvDataBuffer.Reader.ReadAsync();
                if (callbacks.ContainsKey(msg._protocallType))
                {
                    callbacks[msg._protocallType](msg);
                }
                else
                {
                    // 未找到消息处理逻辑
                    Console.WriteLine("未识别的消息类型:" + msg._protocallType.ToString());
                }
            }
        }

        /// <summary>
        /// 向客户端写入消息
        /// </summary>
        static async void WriteToserver(Socket Sock)
        {
            while (true)
            {
                var msg = await sendDataBuffer.Reader.ReadAsync();
                var data = SocketDataToBytes(msg);
                await Sock.SendAsync(new ArraySegment<byte>(data), SocketFlags.None);
            }
        }
        /// <summary>
        /// 网络结构转数据
        /// </summary>
        /// <param name="tmpSocketData"></param>
        /// <returns></returns>
        static private byte[] SocketDataToBytes(sSocketData tmpSocketData)
        {
            byte[] _tmpBuff = new byte[tmpSocketData._buffLength];
            byte[] _tmpBuffLength = BitConverter.GetBytes(tmpSocketData._buffLength);
            byte[] _tmpDataLenght = BitConverter.GetBytes((UInt16)tmpSocketData._protocallType);

            Array.Copy(_tmpBuffLength, 0, _tmpBuff, 0, Constants.HEAD_DATA_LEN);//缓存总长度
            Array.Copy(_tmpDataLenght, 0, _tmpBuff, Constants.HEAD_DATA_LEN, Constants.HEAD_TYPE_LEN);//协议类型
            Array.Copy(tmpSocketData._data, 0, _tmpBuff, Constants.HEAD_LEN, tmpSocketData._dataLength);//协议数据

            return _tmpBuff;
        }


        /// <summary>
        /// 数据转网络结构
        /// </summary>
        /// <param name="_protocalType"></param>
        /// <param name="_data"></param>
        /// <returns></returns>
        static private sSocketData BytesToSocketData(eProtocalCommand _protocalType, byte[] _data)
        {
            sSocketData tmpSocketData = new sSocketData();
            tmpSocketData._buffLength = Constants.HEAD_LEN + _data.Length;
            tmpSocketData._dataLength = _data.Length;
            tmpSocketData._protocallType = _protocalType;
            tmpSocketData._data = _data;
            return tmpSocketData;
        }


        /// <summary>
        /// 接收buffer
        /// </summary>
        static private Channel<sSocketData> recvDataBuffer = Channel.CreateUnbounded<sSocketData>();
        /// <summary>
        /// 发送buffer
        /// </summary>
        static private Channel<sSocketData> sendDataBuffer = Channel.CreateUnbounded<sSocketData>();

        /// <summary>
        /// 接收客户端发来的信息，客户端套接字对象
        /// </summary>
        /// <param name="socketclientpara"></param>    
        static async void recv(Socket socketServer)
        {
            socketServer.ReceiveTimeout = 800;                                             // 接收等待超时时间设为800毫秒
            var _databuffer = new DataBuffer();
            byte[] arrServerRecMsg = new byte[4096];                                       // 创建一个内存缓冲区，其大小为4k字节  
            while (true)
            {
                try
                {
                    var length = await socketServer.ReceiveAsync(new ArraySegment<byte>(arrServerRecMsg), SocketFlags.None);   // 将接收到的信息存入到内存缓冲区，并返回其字节数组的长度    
                    if (length <= 0)                                                      // 视为客户端已经close连接
                    {
                        break;
                    }
                    _databuffer.AddBuffer(arrServerRecMsg, length);                        //将收到的数据添加到缓存器中
                    while (_databuffer.GetData(out sSocketData _socketData))                         //取出一条完整数据
                    {
                        await recvDataBuffer.Writer.WriteAsync(_socketData);                 // 放入channel
                    }
                    Array.Clear(arrServerRecMsg, 0, length);
                }
                catch (SocketException e)
                {
                    if (e.ErrorCode == 10060)                                          // 超时的时候错误号码是10060
                    {
                        continue;                                                      // 继续等待  
                    }
                    break;
                }
                catch (Exception)
                {
                    break;
                }
            }

            socketServer.Close();                                                     // 关闭之前accept出来的和客户端进行通信的套接字 
        }
    }


}