82 lines
3.0 KiB
C#
82 lines
3.0 KiB
C#
using System;
|
|
using System.IO;
|
|
using System.Net.WebSockets;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using ChannelsNS = System.Threading.Channels;
|
|
using XWorld.Framework.Protocol;
|
|
|
|
namespace XWorld.Server.Gateway
|
|
{
|
|
// 一个 WebSocket 连接：收循环解码帧→Submit；发循环排空出站队列→SendAsync（每 socket 仅一个发循环，发送串行）
|
|
public sealed class Connection : IClientConnection
|
|
{
|
|
public int PlayerId { get; }
|
|
|
|
private readonly WebSocket _socket;
|
|
private readonly ServerLoop _loop;
|
|
private readonly ChannelsNS.Channel<byte[]> _outbound = ChannelsNS.Channel.CreateUnbounded<byte[]>();
|
|
|
|
public Connection(int playerId, WebSocket socket, ServerLoop loop)
|
|
{
|
|
PlayerId = playerId; _socket = socket; _loop = loop;
|
|
}
|
|
|
|
public void Send(byte[] frame) => _outbound.Writer.TryWrite(frame);
|
|
|
|
public async Task RunAsync(CancellationToken ct)
|
|
{
|
|
var sendTask = SendLoopAsync(ct);
|
|
try { await ReceiveLoopAsync(ct).ConfigureAwait(false); }
|
|
finally
|
|
{
|
|
_outbound.Writer.TryComplete();
|
|
await sendTask.ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
private async Task ReceiveLoopAsync(CancellationToken ct)
|
|
{
|
|
var buf = new byte[8192];
|
|
using var ms = new MemoryStream();
|
|
try
|
|
{
|
|
while (_socket.State == WebSocketState.Open && !ct.IsCancellationRequested)
|
|
{
|
|
ms.SetLength(0);
|
|
WebSocketReceiveResult res;
|
|
do
|
|
{
|
|
res = await _socket.ReceiveAsync(new ArraySegment<byte>(buf), ct).ConfigureAwait(false);
|
|
if (res.MessageType == WebSocketMessageType.Close) return;
|
|
ms.Write(buf, 0, res.Count);
|
|
} while (!res.EndOfMessage);
|
|
|
|
Frame frame;
|
|
try { frame = FrameCodec.Decode(ms.ToArray()); }
|
|
catch (FormatException) { continue; } // 畸形帧：丢弃，不拖垮连接
|
|
|
|
_loop.Submit(new FrameCommand { PlayerId = PlayerId, Frame = frame });
|
|
}
|
|
}
|
|
catch (OperationCanceledException) { } // 主机关停/请求中止
|
|
catch (WebSocketException) { } // 对端突然断开（TCP 掉线）
|
|
}
|
|
|
|
private async Task SendLoopAsync(CancellationToken ct)
|
|
{
|
|
try
|
|
{
|
|
await foreach (var frame in _outbound.Reader.ReadAllAsync(ct).ConfigureAwait(false))
|
|
{
|
|
if (_socket.State != WebSocketState.Open) break;
|
|
await _socket.SendAsync(new ArraySegment<byte>(frame),
|
|
WebSocketMessageType.Binary, true, ct).ConfigureAwait(false);
|
|
}
|
|
}
|
|
catch (OperationCanceledException) { }
|
|
catch (WebSocketException) { } // 对端已断，停止发送
|
|
}
|
|
}
|
|
}
|