[悬赏300(比较麻烦的话, 可以谈)][ET6 webSocket][帮我远程, 或者有现成webSocket方面能跑的代码帮助我解决网络层问题]

ET_MCC

[悬赏300(比较麻烦的话, 可以谈)][帮我远程, 或者有现成webSocket方面能跑的代码帮助我解决网络层问题]
ET6, 用之前的登录流程, 只是参考(直接抄)了论坛一个哥哥的代码NetWSComponent,然后WService跟WChannel还是用原来ET6的代码, 然后我现在的情况是, 比如Account登录完后, 我就Dispose掉了跟Account服的Session, 重新
创建跟realm服的session, 然后会报错WebSocket error (Code: ConnectionClosedPrematurely): The remote party closed the WebSocket connection without completing the close handshake…

ET_MCC

`namespace ET

{
[ObjectSystem]
public class NetWSComponentAwakeSystem : AwakeSystem<NetWSComponent, int>

{
    public override void Awake(NetWSComponent self, int sessionStreamDispatcherType)

    {
        self.SessionStreamDispatcherType = sessionStreamDispatcherType;

        self.Service = new WService(NetThreadComponent.Instance.ThreadSynchronizationContext);
        self.Service.ErrorCallback += (channelId, error) => self.OnError(channelId, error);
        self.Service.ReadCallback += (channelId, Memory) => self.OnRead(channelId, Memory);

        NetThreadComponent.Instance.Add(self.Service);
    }
}


[ObjectSystem]
public class NetWSComponentAwake1System : AwakeSystem<NetWSComponent, IEnumerable<string>, int>

{
    public override void Awake(NetWSComponent self, IEnumerable<string> prefixs, int sessionStreamDispatcherType)

    {
        self.SessionStreamDispatcherType = sessionStreamDispatcherType;

        self.Service = new WService(NetThreadComponent.Instance.ThreadSynchronizationContext, prefixs);

        self.Service.ErrorCallback += (channelId, error) => self.OnError(channelId, error);

        self.Service.ReadCallback += (channelId, Memory) => self.OnRead(channelId, Memory);

        self.Service.AcceptCallback += (channelId, IPAddress) => self.OnAccept(channelId, IPAddress);

        NetThreadComponent.Instance.Add(self.Service);
    }
}

[ObjectSystem]
public class NetWSComponentDestroySystem : DestroySystem<NetWSComponent>

{
    public override void Destroy(NetWSComponent self)

    {
        NetThreadComponent.Instance.Remove(self.Service);

        self.Service.Destroy();
    }
}


public static class NetWSComponentSystem

{
    public static void OnRead(this NetWSComponent self, long channelId, MemoryStream memoryStream)

    {
        Session session = self.GetChild<Session>(channelId);

        if (session == null)

        {
            return;
        }

        session.LastRecvTime = TimeHelper.ClientNow();

        SessionStreamDispatcher.Instance.Dispatch(self.SessionStreamDispatcherType, session, memoryStream);
    }

    public static void OnError(this NetWSComponent self, long channelId, int error)

    {
        Session session = self.GetChild<Session>(channelId);

        if (session == null)

        {
            return;
        }

        session.Error = error;

        session.Dispose();
    }

    // 这个channelId是由CreateAcceptChannelId生成的

    public static void OnAccept(this NetWSComponent self, long channelId, IPEndPoint ipEndPoint)

    {
        Session session = self.AddChildWithId<Session, AService>(channelId, self.Service);

        session.RemoteAddress = ipEndPoint;

        // 挂上这个组件，5秒就会删除session，所以客户端验证完成要删除这个组件。该组件的作用就是防止外挂一直连接不发消息也不进行权限验证

        session.AddComponent<SessionAcceptTimeoutComponent>();

        // 客户端连接，2秒检查一次recv消息，10秒没有消息则断开

        session.AddComponent<SessionIdleCheckerComponent, int>(NetThreadComponent.checkInteral);
    }

    public static Session Get(this NetWSComponent self, long id)

    {
        Session session = self.GetChild<Session>(id);

        return session;
    }

    public static Session Create(this NetWSComponent self, IPEndPoint realIPEndPoint)

    {
        long channelId = RandomHelper.RandInt64();

        Session session = self.AddChildWithId<Session, AService>(channelId, self.Service);

        session.RemoteAddress = realIPEndPoint;

        session.AddComponent<SessionIdleCheckerComponent, int>(NetThreadComponent.checkInteral);

        self.Service.GetOrCreate(session.Id, realIPEndPoint);

        return session;
    }
}

ET_MCC

`using System.Collections.Generic;

namespace ET

{
public class NetWSComponent : Entity, IAwake<int>, IAwake<IEnumerable<string>, int>, IDestroy

{
    public AService Service;

    public int SessionStreamDispatcherType { get; set; }
}

ET_MCC

`using System;
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Net.WebSockets;

namespace ET
{
public class WService: AService
{
private long idGenerater = 200000000;

    private HttpListener httpListener;
    
    private readonly Dictionary<long, WChannel> channels = new Dictionary<long, WChannel>();

    public WService(ThreadSynchronizationContext threadSynchronizationContext, IEnumerable<string> prefixs)
    {
        this.ThreadSynchronizationContext = threadSynchronizationContext;
        
        this.httpListener = new HttpListener();

        StartAccept(prefixs).Coroutine();
    }
    
    public WService(ThreadSynchronizationContext threadSynchronizationContext)
    {
        this.ThreadSynchronizationContext = threadSynchronizationContext;
    }
    
    private long GetId
    {
        get
        {
            return ++this.idGenerater;
        }
    }
    
    public WChannel Create(string address, long id)
    {
		ClientWebSocket webSocket = new ClientWebSocket();
        WChannel channel = new WChannel(id, webSocket, address, this);
        this.channels[channel.Id] = channel;
        return channel;
    }

    public override void Remove(long id)
    {
        WChannel channel;
        if (!this.channels.TryGetValue(id, out channel))
        {
            return;
        }

        this.channels.Remove(id);
        channel.Dispose();
    }

    public override bool IsDispose()
    {
        return this.ThreadSynchronizationContext == null;
    }

    protected void Get(long id, string address)
    {
        if (!this.channels.TryGetValue(id, out _))
        {
            this.Create(address, id);
        }
    }

    public override void Dispose()
    {
        this.ThreadSynchronizationContext = null;
        this.httpListener?.Close();
        this.httpListener = null;
    }

    private async ETTask StartAccept(IEnumerable<string> prefixs)
    {
        try
        {
            foreach (string prefix in prefixs)
            {
                this.httpListener.Prefixes.Add(prefix);
            }
            
            httpListener.Start();

            while (true)
            {
                try
                {
                    HttpListenerContext httpListenerContext = await this.httpListener.GetContextAsync();

                    HttpListenerWebSocketContext webSocketContext = await httpListenerContext.AcceptWebSocketAsync(null);

                    WChannel channel = new WChannel(this.GetId, webSocketContext, this);

                    this.channels[channel.Id] = channel;

                    this.OnAccept(channel.Id, channel.RemoteAddress);
                }
                catch (Exception e)
                {
                    Log.Error(e);
                }
            }
        }
        catch (HttpListenerException e)
        {
            if (e.ErrorCode == 5)
            {
                throw new Exception($"CMD管理员中输入: netsh http add urlacl url=http://*:8080/ user=Everyone", e);
            }

            Log.Error(e);
        }
        catch (Exception e)
        {
            Log.Error(e);
        }
    }
    
    protected override void Get(long id, IPEndPoint address)
    {
        // 将IPEndPoint转换为WebSocket适用的地址格式（ws://或wss://）
        string webSocketAddress = $"ws://{address.Address}:{address.Port}";

        // 检查是否已存在该通道
        if (!this.channels.TryGetValue(id, out _))
        {
            // 不存在则创建新的WebSocket连接
            this.Create(webSocketAddress, id);
        }
    }

    protected override void Send(long channelId, long actorId, MemoryStream stream)
    {
        this.channels.TryGetValue(channelId, out WChannel channel);
        if (channel == null)
        {
            return;
        }
        channel.Send(stream);
    }

    public override void Update()
    {
    }
}

ET_MCC

`using System;
using System.Collections.Generic;
using System.IO;
using System.Net.WebSockets;
using System.Threading;

namespace ET
{
public class WChannel : AChannel
{
public HttpListenerWebSocketContext WebSocketContext { get; }

    private readonly WService Service;

    private readonly WebSocket webSocket;

    private readonly Queue<byte[]> queue = new Queue<byte[]>();

    private bool isSending;

    private bool isConnected;

    private readonly MemoryStream recvStream;

    private CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();

    public WChannel(long id, HttpListenerWebSocketContext webSocketContext, WService service)
    {
        this.Id = id;
        this.Service = service;
        this.ChannelType = ChannelType.Accept;
        this.WebSocketContext = webSocketContext;
        this.webSocket = webSocketContext.WebSocket;
        this.recvStream = new MemoryStream(ushort.MaxValue);

        isConnected = true;

        this.Service.ThreadSynchronizationContext.PostNext(() =>
        {
            this.StartRecv().Coroutine();
            this.StartSend().Coroutine();
        });
    }

    public WChannel(long id, WebSocket webSocket, string connectUrl, WService service)
    {
        this.Id = id;
        this.Service = service;
        this.ChannelType = ChannelType.Connect;
        this.webSocket = webSocket;
        this.recvStream = new MemoryStream(ushort.MaxValue);

        isConnected = false;

        this.Service.ThreadSynchronizationContext.PostNext(() => this.ConnectAsync(connectUrl).Coroutine());
    }

    public override void Dispose()
    {
        if (this.IsDisposed)
        {
            return;
        }

        this.cancellationTokenSource.Cancel();
        this.cancellationTokenSource.Dispose();
        this.cancellationTokenSource = null;

        this.webSocket.Dispose();
    }

    public async ETTask ConnectAsync(string url)
    {
        try
        {
            await ((ClientWebSocket)this.webSocket).ConnectAsync(new Uri(url), cancellationTokenSource.Token);
            isConnected = true;

            this.StartRecv().Coroutine();
            this.StartSend().Coroutine();
        }
        catch (Exception e)
        {
            Log.Error(e);
            this.OnError(ErrorCore.ERR_WebsocketConnectError);
        }
    }

    public void Send(MemoryStream stream)
    {
        stream.Seek(Packet.ActorIdLength, SeekOrigin.Begin); // 外网不需要actorId
        byte[] bytes = new byte[stream.Length - stream.Position];
        Array.Copy(stream.GetBuffer(), Packet.ActorIdLength, bytes, 0, bytes.Length);
        this.queue.Enqueue(bytes);

        if (this.isConnected)
        {
            this.StartSend().Coroutine();
        }
    }

    public async ETTask StartSend()
    {
        if (this.IsDisposed)
        {
            return;
        }

        try
        {
            if (this.isSending)
            {
                return;
            }

            this.isSending = true;

            while (true)
            {
                if (this.queue.Count == 0)
                {
                    this.isSending = false;
                    return;
                }

                byte[] bytes = this.queue.Dequeue();
                try
                {
                    await this.webSocket.SendAsync(new ArraySegment<byte>(bytes, 0, bytes.Length),
                        WebSocketMessageType.Binary, true, cancellationTokenSource.Token);
                    if (this.IsDisposed)
                    {
                        return;
                    }
                }
                catch (Exception e)
                {
                    Log.Error(e);
                    this.OnError(ErrorCore.ERR_WebsocketSendError);
                    return;
                }
            }
        }
        catch (Exception e)
        {
            Log.Error(e);
        }
    }

    private byte[] cache = new byte[ushort.MaxValue];

    public async ETTask StartRecv()
    {
        if (this.IsDisposed)
        {
            return;
        }

        try
        {
            while (true)
            {
                // 检查 WebSocket 状态
                if (this.webSocket.State != WebSocketState.Open)
                {
                    this.OnError(ErrorCore.ERR_WebsocketNotConnected);
                    return;
                }

#if NOT_UNITY
ValueWebSocketReceiveResult receiveResult;
#else
WebSocketReceiveResult receiveResult;
#endif
int receiveCount = 0;

#if NOT_UNITY
receiveResult = await this.webSocket.ReceiveAsync(
new Memory<byte>(cache, receiveCount, this.cache.Length - receiveCount),
cancellationTokenSource.Token);
#else
receiveResult = await this.webSocket.ReceiveAsync(
new ArraySegment<byte>(this.cache, receiveCount, this.cache.Length - receiveCount),
cancellationTokenSource.Token);
#endif

                        if (this.IsDisposed)
                        {
                            return;
                        }

                        // 检查 WebSocket 状态
                        if (this.webSocket.State != WebSocketState.Open)
                        {
                            this.OnError(ErrorCore.ERR_WebsocketPeerReset);
                            return;
                        }

                        receiveCount += receiveResult.Count;

                        // 检查是否超出缓冲区大小
                        if (receiveCount >= this.cache.Length)
                        {
                            await this.webSocket.CloseAsync(WebSocketCloseStatus.MessageTooBig,
                                $"Message size exceeds buffer capacity: {receiveCount}",
                                cancellationTokenSource.Token);
                            this.OnError(ErrorCore.ERR_WebsocketMessageTooBig);
                            return;
                        }
                    } while (!receiveResult.EndOfMessage);

                    // 处理关闭消息
                    if (receiveResult.MessageType == WebSocketMessageType.Close)
                    {
                        try
                        {
                            await this.webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure,
                                "Closed by remote",
                                cancellationTokenSource.Token);
                        }
                        catch (Exception ex)
                        {
                            Log.Error($"Error closing WebSocket: {ex}");
                        }
                        finally
                        {
                            this.OnError(ErrorCore.ERR_WebsocketPeerReset);
                        }

                        return;
                    }

                    // 处理接收到的数据
                    if (receiveCount > 0)
                    {
                        this.recvStream.SetLength(receiveCount);
                        this.recvStream.Seek(2, SeekOrigin.Begin);
                        Array.Copy(this.cache, 0, this.recvStream.GetBuffer(), 0, receiveCount);
                        this.OnRead(this.recvStream);
                    }
                }
                catch (WebSocketException ex)
                {
                    Log.Error($"WebSocket error (Code: {ex.WebSocketErrorCode}): {ex.Message}");
                    this.OnError(ErrorCore.ERR_WebsocketRecvError);
                    return;
                }
                catch (OperationCanceledException)
                {
                    Log.Info("WebSocket receive operation was canceled");
                    return;
                }
                catch (ObjectDisposedException)
                {
                    Log.Info("WebSocket was disposed during receive operation");
                    return;
                }
                catch (Exception ex)
                {
                    Log.Error($"Unexpected error during WebSocket receive: {ex}");
                    this.OnError(ErrorCore.ERR_WebsocketRecvError);
                    return;
                }
            }
        }
        catch (Exception e)
        {
            Log.Error($"Fatal error in WebSocket receive loop: {e}");
            this.OnError(ErrorCore.ERR_WebsocketRecvError);
        }
    }

    private void OnRead(MemoryStream memoryStream)
    {
        try
        {
            long channelId = this.Id;
            this.Service.OnRead(channelId, memoryStream);
        }
        catch (Exception e)
        {
            Log.DetectionError($"{this.RemoteAddress} {memoryStream.Length} {e}");
            // 出现任何消息解析异常都要断开Session，防止客户端伪造消息
            this.OnError(ErrorCore.ERR_PacketParserError);
        }
    }

    private void OnError(int error)
    {
        Log.Debug($"WChannel error: {error} {this.RemoteAddress}");

        long channelId = this.Id;

        this.Service.Remove(channelId);

        this.Service.OnError(channelId, error);
    }
}

ET_MCC

暂时找了个人帮忙折腾~