using System; using System.Collections.Concurrent; using System.IO; using System.Net.WebSockets; using System.Threading; using System.Threading.Tasks; using Vanguard.Shared.Protocol; namespace Vanguard.Battle { /// /// WebSocket-клиент матча. Кадры — MessagePack из Vanguard.Shared. /// Чтение идёт в async-цикле на главном Unity-потоке, поэтому события /// можно безопасно обрабатывать в UI. /// public sealed class GameConnection : IAsyncDisposable { private readonly ClientWebSocket _socket = new ClientWebSocket(); private readonly SemaphoreSlim _sendLock = new SemaphoreSlim(1, 1); private readonly ConcurrentDictionary> _pending = new ConcurrentDictionary>(); private CancellationTokenSource _cts; private Task _readTask; private Task _pingTask; private int _closed; public event Action HelloReceived; public event Action SnapshotReceived; public event Action EventsReceived; public event Action IntentResultReceived; public event Action ClockReceived; public event Action ErrorReceived; public event Action Closed; public int Seat { get; private set; } = -1; public long LastSeq { get; private set; } public MatchView CurrentView { get; private set; } public bool IsConnected => _socket.State == WebSocketState.Open; public async Task ConnectAsync( string wsUrl, string ticket, string clientBuild, long resumeFromSeq, CancellationToken cancellationToken) { if (_socket.State != WebSocketState.None) throw new InvalidOperationException("Соединение уже открыто или закрыто."); _cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); await _socket.ConnectAsync(new Uri(wsUrl), cancellationToken); await SendAsync(new ClientHello { ProtocolVersion = ProtocolConstants.Version, Ticket = ticket, ClientBuild = clientBuild, ResumeFromSeq = resumeFromSeq, }, cancellationToken); _readTask = ReadLoopAsync(_cts.Token); _pingTask = PingLoopAsync(_cts.Token); } public async Task SubmitAsync( IntentPayload intent, CancellationToken cancellationToken) { var intentId = Guid.NewGuid().ToString("N"); var completion = new TaskCompletionSource(); _pending[intentId] = completion; try { await SendAsync(new ClientIntent { IntentId = intentId, BasedOnSeq = LastSeq, Intent = intent, }, cancellationToken); using (var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken)) { timeout.CancelAfter(TimeSpan.FromSeconds(15)); using (timeout.Token.Register(() => completion.TrySetException(new TimeoutException( "Сервер не ответил на интент.")))) { return await completion.Task; } } } finally { _pending.TryRemove(intentId, out _); } } public Task RequestSnapshotAsync(CancellationToken cancellationToken) => SendAsync(new ClientResync { FromSeq = LastSeq + 1, FullSnapshot = true, }, cancellationToken); public Task LeaveAsync(string reason, CancellationToken cancellationToken) => SendAsync(new ClientLeave { Reason = reason }, cancellationToken); public async Task SendAsync(ClientFrame frame, CancellationToken cancellationToken) { var bytes = ProtocolSerializer.Encode(frame); await _sendLock.WaitAsync(cancellationToken); try { await _socket.SendAsync( new ArraySegment(bytes), WebSocketMessageType.Binary, endOfMessage: true, cancellationToken); } finally { _sendLock.Release(); } } private async Task ReadLoopAsync(CancellationToken cancellationToken) { var buffer = new byte[16 * 1024]; try { while (!cancellationToken.IsCancellationRequested && _socket.State == WebSocketState.Open) { using (var stream = new MemoryStream()) { WebSocketReceiveResult result; do { result = await _socket.ReceiveAsync( new ArraySegment(buffer), cancellationToken); if (result.MessageType == WebSocketMessageType.Close) { NotifyClosed("Сервер закрыл соединение."); return; } if (result.MessageType != WebSocketMessageType.Binary) { NotifyClosed("Сервер прислал не бинарный кадр."); return; } stream.Write(buffer, 0, result.Count); if (stream.Length > ProtocolConstants.MaxFrameBytes) { NotifyClosed("Кадр сервера слишком велик."); return; } } while (!result.EndOfMessage); ServerFrame frame; try { frame = ProtocolSerializer.DecodeServer(stream.ToArray()); } catch (Exception exception) { NotifyClosed("Ошибка разбора кадра: " + exception.Message); return; } HandleFrame(frame, cancellationToken); } } } catch (OperationCanceledException) { NotifyClosed("Соединение закрыто."); } catch (Exception exception) { NotifyClosed("Соединение потеряно: " + exception.Message); } } private void HandleFrame(ServerFrame frame, CancellationToken cancellationToken) { switch (frame) { case ServerHello hello: Seat = hello.PlayerIndex; LastSeq = hello.EventSeq; HelloReceived?.Invoke(hello); break; case ServerSnapshot snapshot: CurrentView = snapshot.View; LastSeq = snapshot.View.EventSeq; SnapshotReceived?.Invoke(snapshot); break; case ServerEvents events: if (events.ToSeq > LastSeq) LastSeq = events.ToSeq; EventsReceived?.Invoke(events); break; case ServerIntentResult result: IntentResultReceived?.Invoke(result); if (_pending.TryGetValue(result.IntentId, out var completion)) completion.TrySetResult(result); break; case ServerClock clock: if (clock.EventSeq > LastSeq) LastSeq = clock.EventSeq; ClockReceived?.Invoke(clock); break; case ServerPing ping: _ = SendPongAsync(ping.Nonce, cancellationToken); break; case ServerPong _: break; case ServerError error: ErrorReceived?.Invoke(error); if (error.Fatal) NotifyClosed(error.Message); break; } } private async Task SendPongAsync(long nonce, CancellationToken cancellationToken) { try { await SendAsync(new ClientPong { Nonce = nonce }, cancellationToken); } catch (Exception) { // соединение уже закрывается } } private async Task PingLoopAsync(CancellationToken cancellationToken) { try { while (!cancellationToken.IsCancellationRequested && IsConnected) { await Task.Delay(ProtocolConstants.ClientPingIntervalMs, cancellationToken); await SendAsync(new ClientPing { Nonce = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(), ClientTimeMs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(), }, cancellationToken); } } catch (OperationCanceledException) { // нормальное завершение } catch (Exception exception) { NotifyClosed("Ping не отправлен: " + exception.Message); } } private void NotifyClosed(string reason) { if (Interlocked.Exchange(ref _closed, 1) != 0) return; foreach (var pending in _pending.Values) pending.TrySetException(new InvalidOperationException(reason)); Closed?.Invoke(reason); } public async ValueTask DisposeAsync() { try { if (_cts != null) _cts.Cancel(); } catch (Exception) { // ignore } try { if (_socket.State == WebSocketState.Open) { await _socket.CloseOutputAsync( WebSocketCloseStatus.NormalClosure, "bye", CancellationToken.None); } } catch (Exception) { // ignore } try { if (_readTask != null) await _readTask; } catch (Exception) { // ignore } try { if (_pingTask != null) await _pingTask; } catch (Exception) { // ignore } _cts?.Dispose(); _sendLock.Dispose(); _socket.Dispose(); } } }