Kioko.GNet.Testing 0.1.0

Kioko.GNET

Библиотека для сериализации/десериализации сообщений и фрейминга протокола GNET в стиле sans‑io. Содержит ядро IO (GNetReader/GNetWriter), source‑generator для сообщений и адаптеры поверх Stream.

См. подробности фрейминга в docs/Framing.md. Для тестового фреймворка используйте docs/TestingFramework.md, для Pandora dump и replay — docs/PandoraDumpReplay.md, для общего workflow regression и benchmark-ов — docs/TestingAndBenchmarks.md.

Быстрый старт

1. Подключите проекты Kioko.GNET и Kioko.GNET.SourceGeneration.
2. Опишите сообщение:

using Kioko.GNet.Protocol;
using Kioko.GNet.Protocol.Attributes;

[GNetSerializable]
[Operation(10)]
public partial record KickoutUser
{
    public int UserId { get; set; }
    public uint LocalsId { get; set; }
    public bool Cause { get; set; }
}

3. Запись/чтение поверх Stream:

using var gnet = new GNetProtocolStream(stream, resolver /* IGNetResolver */);

await gnet.WriteMessageAsync(new KickoutUser { UserId = 100, LocalsId = 200, Cause = true });
var res = await gnet.ReadMessageAsync() ?? throw new InvalidOperationException("Unknown opCode");
// если резолвер не знает opCode, ReadMessageAsync вернёт null и вызовет IGNetProtocolObserver.OnUnknownMessage

По умолчанию включена DefaultMessagePolicy (лимит тела кадра 1 МБ). Можно передать свою IMessagePolicy.

Атрибуты генератора

• [GNetSerializable] — включить тип в генерацию.
• [Operation(uint)] — присвоить operation‑code и реализовать IGNetMessage.
• [Rpc] — пометить тип как RPC (реализует IRpc). В теле RPC первым полем идёт requestId (UInt32).
• [ByteOrder(ByteOrder.LittleEndian|BigEndian)] — применить порядок байт ко всему типу; также поддерживается на уровне отдельных членов.
• [StringEncoding(StringEncoding.Unicode|Ascii)] — кодировка строковых членов.
• [ArraySize(int)] — фиксированный размер массива/словаря (заголовок длины не пишется/не читается).
• [ArraySizeType(ArraySizeType.Byte|Int16|Int32|CUInt32)] — тип заголовка длины коллекций (по умолчанию CUInt32).
• [EnumAs(EnumAs.Byte|Int16|Int32|Int64)] — сериализация enum заданным базовым типом.

Sans‑io

• Чистые API фреймера: FrameEncoder/FrameDecoder (без I/O), см. docs/Framing.md.
• Потоковый адаптер: GNetMessageReader.ReadNextPacketAsync использует FrameDecoder инкрементально, а затем считывает тело в предоставленный буфер.

Nullable

• Поведение и примеры описаны в docs/Nullable.md.
• Коротко: глобально Big Endian, но для Nullable<T> payload всегда в Little Endian (memmove‑совместимо), а префикс длины — varint.

RPC (IRpc)

• Отметьте сообщение атрибутом [Rpc] (тип реализует IRpc). В кадре RPC‑сообщения первым полем тела идёт requestId (UInt32).
• Чтение:

var result = await gnet.ReadMessageAsync();
if (result is not null && result.Value.IsRpc)
{
    var id = result.Value.Identifier; // RequestId/ResponseId
    var rpc = (MyRpc)result.Value.Message; // ваш тип RPC
}

• Запись ответа на RPC (или отправка RPC):

await gnet.WriteRpcAsync(rpcMessage /* IGNetSerializable */,
                         opCode: 0x100u,
                         responseId: myRpcId.ResponseId,
                         ct);

Контейнеры Команд

Для batched command traffic в проекте есть built-in контейнеры:

• 0x22 - C2SCommandsContainer
• 0x00 - S2CCommandsContainer

Рекомендуемый marker для сообщений, допустимых внутри контейнеров - [Command(code)]. Именно по нему built-in registry определяет, что сообщение можно упаковать в command container.

Прозрачное inbound unpacking включается через GNetMessageContainerRegistry:

using var gnet = new GNetProtocolStream(
    stream,
    resolver,
    containerRegistry: new GNetMessageContainerRegistry(),
    inboundDirection: GNetMessageContainerDirection.ClientToServer);

var messages = await gnet.ReadMessagesAsync();

В этом режиме outer container не попадает в handler как обычное сообщение - GNetProtocolStream возвращает уже inner commands.

Для server-side batching есть коробочная агрегация команд через GNetServerOptions.CommandAggregation. Поддерживаются режимы Timer, ManualTrigger и Hybrid, а flush можно вызвать вручную через GNetConnectionContext.FlushCommandsAsync() или GNetServer.FlushCommandsAsync().

Резолвер (IGNetResolver)

Резолвер сопоставляет opCode входящего сообщения с соответствующим типом и создаёт его экземпляр:

using Kioko.GNet;
using Kioko.GNet.Protocol;

public sealed class MyResolver : IGNetResolver
{
    private readonly Dictionary<uint, Func<IGNetMessage>> _factories = new()
    {
        [10u] = () => new KickoutUser(),           // см. [Operation(10)]
        [0x100u] = () => new MyRpc(),               // см. [Rpc] + [Operation]
        // добавляйте остальные сообщения тут
    };

    public bool TryResolveOperation(uint opCode, out IGNetMessage? message)
    {
        if (_factories.TryGetValue(opCode, out var f))
        {
            message = f();
            return true;
        }

        message = null;
        return false;
    }
}

Совет: держите значения [Operation(...)] и таблицу резолвера в одном месте (или генерируйте таблицу автоматически на этапе билда). Генератор Kioko.GNet.SourceGeneration уже создаёт класс Kioko.GNet.Generated.GeneratedResolver, который можно использовать сразу:

using Kioko.GNet.Generated;

var resolver = new GeneratedResolver();

Если встречаются дубликаты [Operation] — генератор сигнализирует диагностикой GNET20. Подробнее про генерацию резолвера и стратегию наложения можно почитать в docs/Resolver.md.

Политика (IMessagePolicy)

По умолчанию используется DefaultMessagePolicy с лимитом тела 1 МБ. Можно задать:

• список разрешённых/запрещённых кодов;
• индивидуальные ограничения размера для конкретных opCode.

Пример конфигурации:

var perCode = new Dictionary<uint, int>
{
    [100u] = 4 * 1024,  // для кода 100 — до 4 KB
    [200u] = 64 * 1024, // для кода 200 — до 64 KB
};

var policy = new DefaultMessagePolicy(
    maxBodySize: 1_048_576,              // глобальный лимит
    allowedCodes: new[] { 100u, 200u },  // опционально
    deniedCodes: null,
    maxBodySizePerCode: perCode);        // индивидуальные лимиты

using var gnet = new GNetProtocolStream(stream, resolver, policy: policy);

Наблюдение и логирование (IGNetProtocolObserver)

• Передайте свою реализацию IGNetProtocolObserver в конструктор GNetProtocolStream, чтобы получать уведомления о записи/чтении сообщений, политических блокировках и исключениях.
• Все callbacks происходят синхронно, поэтому в них нельзя бросать исключения и важно работать быстро.
• Пример:

public sealed class ConsoleObserver : IGNetProtocolObserver
{
    public void OnMessageWriting(uint opCode, IGNetSerializable payload, int bodySize, bool isRpc)
        => Console.WriteLine($"→ {opCode:x}: {payload.GetType().Name} ({bodySize} bytes)");

    public void OnMessageWritten(uint opCode, IGNetSerializable payload, int bodySize, bool isRpc) { }
    public void OnMessageRead(in MessageData meta, IGNetMessage message)
        => Console.WriteLine($"← {meta.Code:x}: {message.GetType().Name}");
    public void OnUnknownMessage(uint opCode, int bodySize)
        => Console.Error.WriteLine($"Unknown opCode {opCode:x} ({bodySize} bytes)");
    public void OnReadError(uint opCode, Exception exception)
        => Console.Error.WriteLine($"Read failed for {opCode:x}: {exception}");
    public void OnPolicyViolation(uint opCode, uint bodySize, MessagePolicyResult violation)
        => Console.Error.WriteLine($"Policy rejected {opCode:x} ({bodySize} bytes): {violation}");
}

await using var gnet = new GNetProtocolStream(stream, resolver, observer: new ConsoleObserver());

Кодировки строк

• ASCII/Unicode выбираются атрибутом [StringEncoding] на члене.
• Произвольные кодировки для чтения доступны через GNetReader.ReadString(string encodingName).

Мини пример (Ping/Pong)

[GNetSerializable]
[Operation(0x10)]
public partial record Ping
{
    public ulong Timestamp { get; set; }
}

[GNetSerializable, Rpc]
[Operation(0x11)]
public partial record Pong
{
    public uint RequestId { get; set; } // добавит генератор
    public ulong Timestamp { get; set; }
}

public sealed class DemoResolver : IGNetResolver
{
    public bool TryResolveOperation(uint opCode, out IGNetMessage? message)
    {
        message = opCode switch
        {
            0x10 => new Ping(),
            0x11 => new Pong(),
            _ => null
        };
        return message is not null;
    }
}

var resolver = new DemoResolver();
var policy = new DefaultMessagePolicy(
    maxBodySize: 256 * 1024,
    allowedCodes: new[] { 0x10u, 0x11u });

await using var stream = new NetworkStream(socket, ownsSocket: false);
await using var gnet = new GNetProtocolStream(stream, resolver, policy);

await gnet.WriteMessageAsync(new Ping { Timestamp = (ulong)Stopwatch.GetTimestamp() });

var result = await gnet.ReadMessageAsync();
if (result?.Message is Pong pong && result.Value.IsRpc)
{
    Console.WriteLine($"pong: {pong.Timestamp}");
}

Sans-io компоненты (FrameEncoder/FrameDecoder) позволяют заменить GNetProtocolStream собственным транспортом (UDP, WebSocket, каналы) — код выше остаётся неизменным, меняется только способ чтения/записи байтов.

TCP-сервер

Kioko.GNet.Server содержит готовый TCP-сервер поверх сокетов. Он поднимает Socket-слушатель и для каждого подключения создаёт GNetProtocolStream.

var server = new GNetServer(new GNetServerOptions
{
    EndPoint = new IPEndPoint(IPAddress.Any, 3333),
    Resolver = resolver,
    ConnectionHandler = async (ctx, ct) =>
    {
        while (!ct.IsCancellationRequested)
        {
            var result = await ctx.Protocol.ReadMessageAsync(ct);
            if (result?.Message is Ping ping)
            {
                await ctx.Protocol.WriteMessageAsync(new Pong { Timestamp = ping.Timestamp }, ct);
            }
        }
    }
});

await server.StartAsync();
Console.WriteLine($"Listening on {server.EndPoint}");

Через GNetServerOptions можно задавать политику сообщений, пул буферов и IGNetProtocolObserver. Для динамического порта указывайте IPEndPoint(IPAddress.Loopback, 0) — фактический адрес доступен через server.EndPoint.

Очередь исходящих сообщений работает автоматически: OutboundQueueCapacity ограничивает объём (по умолчанию 256 запросов), OutboundQueueHighWatermark задаёт «жёлтый» порог, а OutboundQueueHighHandler/OutboundQueueFullHandler дают возможность реагировать (например, логировать, переключать режим деградации или отключать клиентов). Текущее количество ожидающих сообщений доступно через GNetConnectionContext.PendingOutboundMessages.

State-machine API

var stateMachine = GNetStateMachineBuilder.Create()
    .SetInitialState("Handshake")
    .State("Handshake")
        .AllowInbound<LoginRequest>()
        .AllowOutbound<LoginAccepted>()
        .Transition<LoginRequest>("World")
    .State("World")
        .AllowInbound<WorldPing>()
        .AllowOutbound<WorldPong>()
    .Build();

options.StateMachine = stateMachine;
options.StateViolationHandler = (ctx, violation) =>
    Console.WriteLine($"[{ctx.RemoteEndPoint}] violation {violation.Kind} for 0x{violation.OpCode:X}");

State-machine ограничивает допустимые сообщения и переходы между состояниями, локальные обработчики задаются через State().OnViolation(...), глобальный — через GNetServerOptions.StateViolationHandler.

Throttling builder

options.Throttler = GNetThrottleBuilder.Create()
    .ForMessage<LoginRequest>(ThrottleDirection.Inbound, rule => rule
        .Limit(3)
        .Per(TimeSpan.FromSeconds(1))
        .OnExceeded((ctx, decision) =>
            ctx.SendAsync(new ThrottleWarning { Code = decision.OpCode })))
    .ForMessage<WorldPing>(ThrottleDirection.Outbound, rule => rule.Limit(20).Per(TimeSpan.FromMilliseconds(200)))
    .Build();

options.ThrottleViolationHandler = (ctx, decision) =>
    Console.WriteLine($"[{ctx.RemoteEndPoint}] throttled op=0x{decision.OpCode:X}");

Throttler ведёт отдельные fixed-window счётчики на связке (direction, opCode). В GNetConnectionContext состояние хранится автоматически; хук ThrottleViolationHandler позволяет централизованно реагировать/отправлять ошибки.

Back-pressure и метрики

GNetConnectionContext.SendAsync теперь неблокирующе ставит сообщение в очередь. Если глубина достигает OutboundQueueHighWatermark, сервер вызывает обработчик и публикует событие в GNetServerMetrics.Meter:

• gnet.server.outbound_queue.high_events (Counter<long>, теги: capacity, pending);
• gnet.server.outbound_queue.full_events;
• gnet.server.outbound_queue.depth (Histogram<double>).

using var listener = new MeterListener();
listener.InstrumentPublished = (instrument, meterListener) =>
{
    if (instrument.Meter == GNetServerMetrics.Meter)
        meterListener.EnableMeasurementEvents(instrument);
};
listener.SetMeasurementEventCallback<long>((instrument, measurement, tags, _) =>
{
    if (instrument.Name == "gnet.server.outbound_queue.full_events")
        Console.WriteLine($"Queue full: +{measurement}");
});
listener.Start();

Так же можно подписаться на события через GNetServerOptions.OutboundQueueHighHandler/OutboundQueueFullHandler и, например, отключать «тяжёлых» клиентов или переключать их в режим деградации.

Message pipeline

Для типовой маршрутизации сообщений используйте GNetMessagePipelineBuilder — он позволяет привязать обработчики к типам и решать, выполнять их inline или вынести на ThreadPool.

var pipeline = GNetMessagePipelineBuilder.Create()
    .Handle<LoginRequest, LoginAccepted>((ctx, msg, ct) =>
        new LoginAccepted { Motd = "hello" })
    .HandleInBackground<TradePacket>(async (ctx, msg, ct) =>
    {
        await tradeGateway.ForwardAsync(msg, ct);
    }, onError: (ctx, ex) => ctx.SendAsync(new TradeError { Reason = ex.Message }))
    .Build();

options.ConnectionHandler = pipeline.AsConnectionHandler(async (ctx, result, ct) =>
{
    Console.WriteLine($"Unhandled op=0x{result.Code:X}");
    if (result.Message is IDisposable disposable)
        disposable.Dispose();
    await Task.CompletedTask;
});

• Handle<T> — inline-обработка без дополнительных аллокаций (ValueTask по умолчанию). Подходит для лёгких операций и ECS, не блокирует цикл чтения. Обработчик может возвращать IGNetMessage (или Task/ValueTask с сообщением) — ответ автоматически отправится через ctx.SendAsync.
• HandleInBackground<T> — запуск обработчика на ThreadPool (принимает onError). Полезно для интеграции с БД/внешними сервисами; сбои изолируются на уровне конкретного типа сообщения.
• AsConnectionHandler конвертирует pipeline в готовый ConnectionHandler для GNetServerOptions. onUnhandled позволяет централизованно логировать/диспозить неизвестные сообщения.
• ConnectionClosedHandler в GNetServerOptions вызывается после завершения соединения. Для каждого контекста доступны ConnectionId, IsDisconnected, DisconnectException и Services (если задали ConnectionServicesFactory), поэтому можно безопасно убирать сущности из игрового мира или логировать причину разрыва. Ошибки внутри ConnectionHandler обрабатывайте через ConnectionErrorHandler, а handshake вынесите в HandshakeHandler — он выполняется один раз перед основным обработчиком и подходит для настройки RC4/MPPC.

Если хотите ещё более типизированный роутинг, пометьте класс [HandlerClass] и добавьте методы с [Handler]:

[HandlerClass]
public partial class LoginHandler
{
    [Handler]
    public LoginAccepted HandleLogin(GNetConnectionContext ctx, LoginRequest msg, CancellationToken ct)
        => new() { Motd = $"Welcome, {msg.UserName}!" };

    [Handler]
    public ValueTask<WorldPong> HandleWorldPing(GNetConnectionContext ctx, WorldPing msg, CancellationToken ct)
        => ValueTask.FromResult(new WorldPong { Tick = msg.Tick });
}

[Handler] автоматически принимает opCode из [Operation] сообщения во втором аргументе, поэтому дополнительные параметры нужны только если вы хотите переопределить привязку вручную.

Исходный генератор создаст метод InvokeAsync (так что LoginHandler автоматически реализует IGNetMessageHandler). Его можно обёрнуть в GNetHandlerStateMachine, переключая состояние через SetState, например: stateMachine.SetState(new WorldHandler());. Так вы получаете state machine без текстовых имён и без ручных switch.

RC4 и MPPC

Kioko.GNet.Security.Cryptography.RC4Stream умеет прозрачно включать/выключать шифрование после рукопожатия:

var tcp = new NetworkStream(socket, ownsSocket: false);
var rc4 = new RC4Stream(tcp, leaveOpen: false);
// ... plain traffic ...
rc4.SetWriteKey(writeKey);
rc4.SetReadKey(readKey);
using var protocol = new GNetProtocolStream(rc4, resolver);

Kioko.GNet.IO.Compression.MppcStream сжимает исходящий трафик без дополнительных аллокаций, его можно навесить поверх RC4 (или напрямую на NetworkStream). Для случаев, когда сжимается только запись (как в изначальном протоколе), чтение остаётся через оригинальный поток:

var rawStream = new NetworkStream(socket, ownsSocket: false);
var encrypted = new RC4Stream(rawStream, leaveOpen: true);
var compressed = new MppcStream(encrypted, leaveOpen: false);
using var protocol = new GNetProtocolStream(compressed, resolver);

Тесты CompressionAndCryptoTests содержат эталонные снэпшоты MPPC и RC4, так что изменение алгоритмов сразу подсветит регресс.

Пример решения смотрите в samples/GNetServerExample — он поднимает сервер со state-machine, observer и тестовым клиентом (dotnet run --project samples/GNetServerExample).

End-to-end сценарий (сервер TCP)

var listener = TcpListener.Create(port);
listener.Start();

while (true)
{
    var client = await listener.AcceptTcpClientAsync(ct);
    _ = Task.Run(async () =>
    {
        await using var stream = client.GetStream();
        using var gnet = new GNetProtocolStream(stream, resolver, observer: serverObserver);

        while (!ct.IsCancellationRequested)
        {
            var result = await gnet.ReadMessageAsync(ct);
            if (result is null)
                continue;

            switch (result.Value.Code)
            {
                case Ping.OperationCode:
                    await gnet.WriteMessageAsync(new Pong { Timestamp = ((Ping)result.Value.Message).Timestamp }, ct);
                    break;
                default:
                    // обработка остальных пакетов
                    break;
            }
        }
    }, ct);
}

Совместимость и версионирование

• Семвер: патчи содержат исправления без изменения ABI/формата, миноры добавляют новые атрибуты/типы, мажоры допускают breaking changes. До релиза 1.0 breaking возможны, но документируются в docs/Framing.md.
• Формат сообщений: новые поля добавляйте в конец и делайте опциональными (Nullable<T>), чтобы старые клиенты могли прочитать payload, игнорируя добавления.
• Operation-коды должны оставаться стабильными. Переназначайте код только в мажорных релизах или заводите новые типы.
• Byte order — Big Endian по умолчанию. Используйте [ByteOrder] адресно, иначе придётся синхронно обновлять все клиенты.
• Политика сообщений (IMessagePolicy) — внешний контракт по лимитам; меняйте дефолтные ограничители только в мажорных версиях. Для собственных политик документируйте выделенные лимиты, чтобы операторы знали, чего ожидать в проде.

Безопасность и производительность

• Безопасность сообщений. Устанавливайте жёсткие лимиты в DefaultMessagePolicy (глобальные и per-code), а также передавайте списки разрешённых opCode, чтобы отбрасывать неожиданные сообщения до десериализации. Для серверов, работающих с множеством резолверов, держите отдельные политики на RPC и на чат/геймплей пакеты.
• Пропорциональная аллокация. GNetMessageReader использует предоставленный буфер; на чтение тела выделяйте пул ArrayPool<byte>/MemoryPool<byte> и никогда не принимайте длину кадра, превышающую допустимый порог — иначе есть риск OOM/DoS.
• Transport Agnostic. Sans-io ядро позволяет внедрить инспекторы перед записью/после чтения (например, TLS record inspector) без доступа к Stream, что упрощает аудит.
• BenchmarkDotNet. Проект benchmarks/Kioko.GNet.Benchmarks покрывает varint encode/decode (CUInt32), заголовки фреймов, сериализацию сообщений и контейнерные сценарии (0x22 / 0x00, GNetProtocolStream, Pandora golden replay/import). Прогоняйте dotnet run -c Release --project benchmarks/Kioko.GNet.Benchmarks -- --filter * перед релизом и сохраняйте результаты.
• Stress тесты. Повторяйте чтение/запись FrameDecoder на шумном потоке (по 10^6 пакетов) и следите за GC (Gen0/Gen2). Документируйте результаты в docs/Framing.md, чтобы понимать границы пропускной способности.

Типичные ошибки

• Неизвестный opCode — IGNetResolver должен возвращать корректную реализацию для каждого [Operation]. Держите таблицу в одном месте и покрывайте unit-тестом.
• Превышен лимит размера — настройте DefaultMessagePolicy так, чтобы лимиты совпадали на клиенте и сервере; при расхождениях вы будете получать MessagePolicyException.
• Блокирующие обработчики — IGNetProtocolObserver и обработчики сообщений вызываются синхронно; используйте неблокирующие операции или отложенную обработку, иначе заблокируете поток чтения.
• Необработанные исключения при чтении — любые броски из msg.Read закрывают соединение. Оборачивайте пользовательский код в try/catch и возвращайте понятные ответы/ошибки на протокольном уровне.

Бенчмарки

• dotnet run -c Release --project benchmarks/Kioko.GNet.Benchmarks -- --filter * — полный набор (ShortRun).
• dotnet run -c Release --project benchmarks/Kioko.GNet.Benchmarks -- --filter *Container* — только контейнеры и Pandora golden replay/import.
• dotnet run -c Release --project benchmarks/Kioko.GNet.Benchmarks -- --filter *PandoraContainerReplayBenchmarks* — только realistic сценарий на checked-in Pandora fixture.
• CUInt32Benchmarks — varint encode/decode для характерных значений (<=0x7F, <=0x3FFF, <=0x1FFFFFFF, uint.Max).
• FrameEncoderBenchmarks — измерение FrameEncoder.GetHeaderSize/WriteHeader для тел 4 B–64 KB.
• MessageSerializationBenchmarks — запись комплексного сообщения и nullable коллекций, плюс десериализация nullable payload (LE).
• MessageContainerBenchmarks — синтетические benchmark-ы для built-in контейнеров C2SCommandsContainer / S2CCommandsContainer и transparent unpacking через GNetProtocolStream.
• PandoraContainerReplayBenchmarks — импорт checked-in Pandora fixture в payload/frame transcript и replay через InMemoryTransportHarness.

BenchmarkDotNet выводит статистику в BenchmarkDotNet.Artifacts/results. Сохраняйте результаты (например, results/*.md) в Wiki/Docs, чтобы отслеживать регрессии. Текущий benchmark-набор используется для ручного regression tracking по времени и аллокациям; perf-gate и CI fail/pass thresholds в проекте пока не вводятся.