proto-socket/dart/test/communicator_test.dart

import 'dart:async';
import 'dart:mirrors';

import 'package:test/test.dart';
import 'package:proto_socket/proto_socket.dart';

class _FakeCommunicator extends Communicator {
  final _FakeTransport transport = _FakeTransport();

  _FakeCommunicator({bool packageQualifiedParser = false}) {
    isAlive = true;
    final testDataType = packageQualifiedParser
        ? 'example.TestData'
        : TestData.getDefault().info_.qualifiedMessageName;
    initialize({
      testDataType: TestData.fromBuffer,
      HeartBeat.getDefault().info_.qualifiedMessageName: HeartBeat.fromBuffer,
    }, transport: transport);
  }

  void closeForTest() {
    isAlive = false;
    cancelPendingRequests();
  }

  void setNonceForTest(int value) {
    nonce = value;
  }

  void attachGatewayForTest(InboundGateway gateway) =>
      attachInboundGateway(gateway);

  Future<void> receiveFrameForTest(List<int> frame) => onReceivedFrame(frame);
}

/// Builds a raw `PacketBase` frame buffer carrying a `TestData` payload.
List<int> _frameBytes(int index,
    {int incomingNonce = 0, int responseNonce = 0}) {
  return (PacketBase()
        ..typeName = TestData.getDefault().info_.qualifiedMessageName
        ..nonce = incomingNonce
        ..responseNonce = responseNonce
        ..data = (TestData()..index = index).writeToBuffer())
      .writeToBuffer();
}

class _FakeTransport implements Transport {
  final sentPackets = <PacketBase>[];
  Object? error;

  @override
  Future<void> writePacket(PacketBase base) async {
    final error = this.error;
    if (error != null) {
      throw error;
    }
    sentPackets.add(base);
  }

  @override
  Future<void> close() async {}
}

Map<int, Object?> _pendingRequestsOf(Communicator communicator) {
  final library = reflectClass(Communicator).owner as LibraryMirror;
  final symbol = MirrorSystem.getSymbol('_pendingRequests', library);
  return reflect(communicator).getField(symbol).reflectee as Map<int, Object?>;
}

void main() {
  group('Communicator protocol guards', () {
    test('response typeName mismatch completes sendRequest with error',
        () async {
      final communicator = _FakeCommunicator();
      final future = communicator.sendRequest<TestData, TestData>(
        TestData()
          ..index = 1
          ..message = 'hello',
      );

      await Future<void>.delayed(Duration.zero);
      final requestNonce = communicator.transport.sentPackets.single.nonce;
      communicator.onReceivedData(
        HeartBeat.getDefault().info_.qualifiedMessageName,
        HeartBeat().writeToBuffer(),
        responseNonce: requestNonce,
      );

      await expectLater(
        future,
        throwsA(
          isA<StateError>().having((error) => error.toString(), 'message',
              contains('Response type mismatch')),
        ),
      );
    });

    test('close 후 sendRequest는 StateError로 완료된다', () async {
      final communicator = _FakeCommunicator();
      final future = communicator.sendRequest<TestData, TestData>(
        TestData()..index = 1,
      );

      await Future<void>.delayed(Duration.zero);
      communicator.closeForTest();

      await expectLater(
        future,
        throwsA(
          isA<StateError>().having(
            (error) => error.message,
            'message',
            contains('connection closed'),
          ),
        ),
      );
    });

    test('sendRequest가 timeout 내 응답 없으면 TimeoutException을 던진다', () async {
      final communicator = _FakeCommunicator();

      final future = communicator.sendRequest<TestData, TestData>(
        TestData()
          ..index = 1
          ..message = 'wait',
        timeout: const Duration(milliseconds: 50),
      );

      await Future<void>.delayed(Duration.zero);
      expect(communicator.transport.sentPackets, hasLength(1));

      await expectLater(
        future,
        throwsA(isA<TimeoutException>()),
      );
      expect(_pendingRequestsOf(communicator), isEmpty);
    });

    test('nonce wraps after int32 max without emitting zero', () async {
      final communicator = _FakeCommunicator();
      communicator.setNonceForTest(Communicator.maxNonce - 1);

      await communicator.send(TestData()..index = 1);
      await communicator.send(TestData()..index = 2);

      expect(
        communicator.transport.sentPackets.map((packet) => packet.nonce),
        [Communicator.maxNonce, 1],
      );
      expect(
        communicator.transport.sentPackets.map((packet) => packet.nonce),
        isNot(contains(0)),
      );
    });

    test('sendRequest matches response at nonce wrap boundary', () async {
      final communicator = _FakeCommunicator();
      communicator.setNonceForTest(Communicator.maxNonce - 1);

      final maxNonceFuture = communicator.sendRequest<TestData, TestData>(
        TestData()
          ..index = 1
          ..message = 'max',
      );
      await Future<void>.delayed(Duration.zero);
      final maxNonceRequest = communicator.transport.sentPackets.single;
      expect(maxNonceRequest.nonce, Communicator.maxNonce);
      expect(maxNonceRequest.nonce, isNot(0));
      communicator.onReceivedData(
        TestData.getDefault().info_.qualifiedMessageName,
        (TestData()
              ..index = 2
              ..message = 'max response')
            .writeToBuffer(),
        responseNonce: maxNonceRequest.nonce,
      );
      expect((await maxNonceFuture).message, 'max response');

      final wrappedFuture = communicator.sendRequest<TestData, TestData>(
        TestData()
          ..index = 3
          ..message = 'wrapped',
      );
      await Future<void>.delayed(Duration.zero);
      final wrappedRequest = communicator.transport.sentPackets.last;
      expect(wrappedRequest.nonce, 1);
      expect(wrappedRequest.nonce, isNot(0));
      communicator.onReceivedData(
        TestData.getDefault().info_.qualifiedMessageName,
        (TestData()
              ..index = 4
              ..message = 'wrapped response')
            .writeToBuffer(),
        responseNonce: wrappedRequest.nonce,
      );
      expect((await wrappedFuture).message, 'wrapped response');
    });

    test('sendRequest accepts package-qualified response typeName', () async {
      final communicator = _FakeCommunicator(packageQualifiedParser: true);
      final future = communicator.sendRequest<TestData, TestData>(
        TestData()
          ..index = 1
          ..message = 'request',
      );

      await Future<void>.delayed(Duration.zero);
      final requestNonce = communicator.transport.sentPackets.single.nonce;
      communicator.onReceivedData(
        'example.TestData',
        (TestData()
              ..index = 2
              ..message = 'qualified response')
            .writeToBuffer(),
        responseNonce: requestNonce,
      );

      expect((await future).message, 'qualified response');
    });

    test('addListener accepts package-qualified message typeName', () async {
      final communicator = _FakeCommunicator(packageQualifiedParser: true);
      final messages = <TestData>[];
      communicator.addListener<TestData>(messages.add);

      communicator.onReceivedData(
        'example.TestData',
        (TestData()
              ..index = 3
              ..message = 'qualified event')
            .writeToBuffer(),
      );

      // inbound Future chain이 처리되도록 이벤트 루프를 양보한다
      await Future<void>.delayed(Duration.zero);

      expect(messages, hasLength(1));
      expect(messages.single.message, 'qualified event');
    });

    test('addRequestListener accepts package-qualified request typeName',
        () async {
      final communicator = _FakeCommunicator(packageQualifiedParser: true);
      communicator.addRequestListener<TestData, TestData>((req) async {
        return TestData()
          ..index = req.index + 1
          ..message = 'echo: ${req.message}';
      });

      communicator.onReceivedData(
        'example.TestData',
        (TestData()
              ..index = 4
              ..message = 'qualified request')
            .writeToBuffer(),
        incomingNonce: 12,
      );
      await Future<void>.delayed(Duration.zero);

      final response = communicator.transport.sentPackets.single;
      expect(response.responseNonce, 12);
      final data = TestData.fromBuffer(response.data);
      expect(data.index, 5);
      expect(data.message, 'echo: qualified request');
    });

    test(
        'cannot register addRequestListener for a type already using addListener',
        () {
      final communicator = _FakeCommunicator();

      communicator.addListener<TestData>((_) {});

      expect(
        () => communicator.addRequestListener<TestData, TestData>(
          (req) async => TestData()..index = req.index,
        ),
        throwsA(isA<StateError>()),
      );
    });

    test(
        'cannot register addListener for a type already using addRequestListener',
        () {
      final communicator = _FakeCommunicator();

      communicator.addRequestListener<TestData, TestData>(
        (req) async => TestData()..index = req.index,
      );

      expect(
        () => communicator.addListener<TestData>((_) {}),
        throwsA(isA<StateError>()),
      );
    });

    test('queuePacket uses injected transport', () async {
      final communicator = _FakeCommunicator();
      final packet = PacketBase()
        ..typeName = TestData.getDefault().info_.qualifiedMessageName
        ..nonce = 1
        ..data = (TestData()..index = 7).writeToBuffer();

      await communicator.queuePacket(packet);

      expect(communicator.transport.sentPackets, hasLength(1));
      expect(communicator.transport.sentPackets.single, same(packet));

      final error = StateError('write failed');
      communicator.transport.error = error;
      await expectLater(
        communicator.queuePacket(PacketBase()..typeName = 'TestData'),
        throwsA(same(error)),
      );
    });

    test('inbound queue FIFO: 여러 메시지를 onReceivedData 호출 순서대로 받는다',
        () async {
      final communicator = _FakeCommunicator();
      final received = <int>[];

      communicator.addListener<TestData>((data) => received.add(data.index));

      for (var i = 1; i <= 5; i++) {
        final data = (TestData()..index = i).writeToBuffer();
        communicator.onReceivedData(
          TestData.getDefault().info_.qualifiedMessageName,
          data,
          incomingNonce: i,
        );
      }

      // inbound Future chain이 처리되도록 이벤트 루프 양보
      for (var i = 0; i < 20; i++) {
        if (received.length == 5) break;
        await Future<void>.delayed(Duration.zero);
      }

      expect(received, [1, 2, 3, 4, 5]);
    });

    test('inbound queue full 시 backpressure가 가동한다', () async {
      final communicator = _FakeCommunicator();
      final data = (TestData()..index = 1..message = 'heavy').writeToBuffer();

      final completer = Completer<TestData>();

      // 지연 request 핸들러 등록
      communicator.addRequestListener<TestData, TestData>((req) async {
        return completer.future;
      });

      // 1번째 호출: 핸들러가 completer.future에 의해 블로킹됨
      await communicator.onReceivedData(
        TestData.getDefault().info_.qualifiedMessageName,
        data,
        incomingNonce: 1,
      );

      // 2번째부터 64번째(총 63개)를 enqueue: 첫번째 아이템은 디스패치 중이므로 _inboundQueueLength에 포함됨
      // 2 ~ 64번째(총 63개) 추가 시 큐가 꽉 차게 됨 (_inboundQueueLength = 64)
      for (var i = 2; i <= 64; i++) {
        await communicator.onReceivedData(
          TestData.getDefault().info_.qualifiedMessageName,
          data,
          incomingNonce: i,
        );
      }

      // 65번째 item 추가 시도: 비동기로 대기(블로킹)해야 함
      var enqueueDone = false;
      final enqueueFuture = communicator.onReceivedData(
        TestData.getDefault().info_.qualifiedMessageName,
        data,
        incomingNonce: 65,
      ).then((_) {
        enqueueDone = true;
      });

      await Future<void>.delayed(const Duration(milliseconds: 20));
      expect(enqueueDone, isFalse);

      // 핸들러를 완료시켜 큐를 하나 소모하게 함
      completer.complete(TestData()..index = 100);

      await enqueueFuture;
      expect(enqueueDone, isTrue);

      communicator.closeForTest();
    });

    test('느린 request handler 자동 응답이 FIFO를 보존한다', () async {
      final communicator = _FakeCommunicator();

      communicator.addRequestListener<TestData, TestData>((req) async {
        if (req.index == 1) {
          // 첫 번째 요청 50ms 지연
          await Future<void>.delayed(const Duration(milliseconds: 50));
        }
        return TestData()
          ..index = req.index + 100
          ..message = 'echo';
      });

      communicator.onReceivedData(
        TestData.getDefault().info_.qualifiedMessageName,
        (TestData()..index = 1).writeToBuffer(),
        incomingNonce: 10,
      );
      communicator.onReceivedData(
        TestData.getDefault().info_.qualifiedMessageName,
        (TestData()..index = 2).writeToBuffer(),
        incomingNonce: 20,
      );

      // 두 응답이 전송될 때까지 대기 (최대 2초)
      for (var i = 0; i < 400; i++) {
        if (communicator.transport.sentPackets.length >= 2) break;
        await Future<void>.delayed(const Duration(milliseconds: 5));
      }

      expect(communicator.transport.sentPackets.length, greaterThanOrEqualTo(2));
      // FIFO: 첫 번째 응답이 responseNonce=10, 두 번째가 responseNonce=20
      expect(communicator.transport.sentPackets[0].responseNonce, 10);
      expect(communicator.transport.sentPackets[1].responseNonce, 20);
    });

    test('close 시 pending sendRequest가 StateError로 완료된다 (inbound queue 정리)',
        () async {
      final communicator = _FakeCommunicator();

      final future = communicator.sendRequest<TestData, TestData>(
        TestData()..index = 99,
        timeout: const Duration(seconds: 5),
      );

      await Future<void>.delayed(Duration.zero);
      communicator.closeForTest();

      await expectLater(
        future,
        throwsA(
          isA<StateError>().having(
            (error) => error.message,
            'message',
            contains('connection closed'),
          ),
        ),
      );
    });

    test('graceful close 시 in-flight request handler가 완료되고 자동 응답 순서가 유지된다',
        () async {
      final communicator = _FakeCommunicator();
      var handlerStarted = false;
      var handlerFinished = false;

      communicator.addRequestListener<TestData, TestData>((req) async {
        handlerStarted = true;
        await Future<void>.delayed(const Duration(milliseconds: 50));
        handlerFinished = true;
        return TestData()
          ..index = req.index + 100
          ..message = 'graceful_response';
      });

      unawaited(communicator.onReceivedData(
        TestData.getDefault().info_.qualifiedMessageName,
        (TestData()..index = 5).writeToBuffer(),
        incomingNonce: 42,
      ));

      for (var i = 0; i < 10; i++) {
        if (handlerStarted) break;
        await Future<void>.delayed(const Duration(milliseconds: 5));
      }
      expect(handlerStarted, isTrue);
      expect(handlerFinished, isFalse);

      await communicator.close();

      expect(handlerFinished, isTrue);
      expect(communicator.isAlive, isFalse);

      expect(communicator.transport.sentPackets, hasLength(1));
      final response = communicator.transport.sentPackets.single;
      expect(response.responseNonce, 42);
      final responseData = TestData.fromBuffer(response.data);
      expect(responseData.index, 105);
      expect(responseData.message, 'graceful_response');
    });

    test('worker gateway: out-of-order seq를 reorder한 뒤 입력 순서로 dispatch한다',
        () async {
      final communicator = _FakeCommunicator();
      final received = <int>[];

      communicator.addListener<TestData>((data) => received.add(data.index));

      // Fake gateway: out-of-order(seq=2, seq=1)로 전달된 아이템
      final items = <({int seq, int index, int incomingNonce})>[
        (seq: 2, index: 2, incomingNonce: 2),
        (seq: 1, index: 1, incomingNonce: 1),
      ];

      // coordinator로 밀어넣기 전에 seq 오름차순으로 reorder
      items.sort((a, b) => a.seq.compareTo(b.seq));

      for (final item in items) {
        communicator.onReceivedData(
          TestData.getDefault().info_.qualifiedMessageName,
          (TestData()..index = item.index).writeToBuffer(),
          incomingNonce: item.incomingNonce,
        );
      }

      // inbound Future chain이 처리되도록 이벤트 루프 양보
      for (var i = 0; i < 20; i++) {
        if (received.length == 2) break;
        await Future<void>.delayed(Duration.zero);
      }

      expect(received, [1, 2]);
    });

    test('FrameReorderBuffer: out-of-order seq를 입력 순서로 release한다', () {
      final buffer = FrameReorderBuffer();
      DecodedFrame frame(int seq) => DecodedFrame(
            seq: seq,
            typeName: 'T',
            data: const [],
            incomingNonce: seq,
            responseNonce: 0,
          );

      // seq=2가 먼저 도착하면 아직 release되지 않는다.
      expect(buffer.release(frame(2)), isEmpty);
      // seq=1 도착 시 1, 2가 순서대로 release된다.
      expect(buffer.release(frame(1)).map((f) => f.seq), [1, 2]);
      // 이어지는 seq=3은 즉시 release된다.
      expect(buffer.release(frame(3)).map((f) => f.seq), [3]);
      // 이미 release된 seq는 중복 emit하지 않는다.
      expect(buffer.release(frame(1)), isEmpty);
    });

    test('SyncInboundGateway: onReceivedFrame이 디코드 후 입력 순서로 dispatch한다',
        () async {
      final communicator = _FakeCommunicator();
      final received = <int>[];
      communicator.addListener<TestData>((data) => received.add(data.index));
      communicator.attachGatewayForTest(SyncInboundGateway());

      for (var i = 1; i <= 5; i++) {
        await communicator.receiveFrameForTest(_frameBytes(i, incomingNonce: i));
      }

      for (var i = 0; i < 40; i++) {
        if (received.length == 5) break;
        await Future<void>.delayed(Duration.zero);
      }

      expect(received, [1, 2, 3, 4, 5]);
      await communicator.close();
    });

    test('SyncInboundGateway: responseNonce 프레임이 pending sendRequest를 완료한다',
        () async {
      final communicator = _FakeCommunicator();
      communicator.attachGatewayForTest(SyncInboundGateway());

      final future = communicator.sendRequest<TestData, TestData>(
        TestData()..index = 1,
      );
      await Future<void>.delayed(Duration.zero);
      final requestNonce = communicator.transport.sentPackets.single.nonce;

      await communicator.receiveFrameForTest(
        _frameBytes(2, incomingNonce: 999, responseNonce: requestNonce),
      );

      expect((await future).index, 2);
      await communicator.close();
    });

    test('IsolateInboundGateway: 상시 isolate 디코드 후 수신/응답 순서가 유지된다',
        () async {
      final communicator = _FakeCommunicator();
      final received = <int>[];
      communicator.addListener<TestData>((data) => received.add(data.index));

      final gateway = IsolateInboundGateway();
      await gateway.start();
      communicator.attachGatewayForTest(gateway);

      for (var i = 1; i <= 5; i++) {
        await communicator.receiveFrameForTest(_frameBytes(i, incomingNonce: i));
      }

      for (var i = 0; i < 200; i++) {
        if (received.length == 5) break;
        await Future<void>.delayed(const Duration(milliseconds: 5));
      }

      expect(received, [1, 2, 3, 4, 5]);
      await communicator.close();
    });

    test(
        'IsolateInboundGateway submit waits for ready and preserves all frames under capacity pressure',
        () async {
      final gateway = IsolateInboundGateway(capacity: 4);
      final received = <int>[];
      final sub = gateway.results.listen((frame) => received.add(frame.seq));

      // start()를 await하지 않고 곧바로 submit해 ready 대기와 drop 없음을 검증한다.
      final startFuture = gateway.start();

      const total = 50;
      for (var seq = 1; seq <= total; seq++) {
        await gateway.submit(InboundFrame(seq: seq, bytes: _frameBytes(seq)));
        // capacity(4)를 초과하는 in-flight backlog가 쌓이지 않는다.
        expect(gateway.inFlightCount, lessThanOrEqualTo(4));
      }
      await startFuture;

      for (var i = 0; i < 400; i++) {
        if (received.length == total) break;
        await Future<void>.delayed(const Duration(milliseconds: 5));
      }

      expect(received, List<int>.generate(total, (i) => i + 1));
      await sub.cancel();
      await gateway.close();
    });

    test('SyncInboundGateway: malformed frame이 ordered decode error를 emit한다',
        () async {
      final gateway = SyncInboundGateway();
      final errors = <int>[];
      final dispatched = <int>[];
      final sub = gateway.results.listen((frame) {
        if (frame.isError) {
          errors.add(frame.seq);
        } else {
          dispatched.add(frame.incomingNonce);
        }
      });

      // seq=2가 malformed여도 seq=1은 정상 dispatch되고 seq=2는 error로 release되어
      // 이후 seq가 reorder buffer에 영구 잔류하지 않는다.
      await gateway.submit(InboundFrame(seq: 1, bytes: _frameBytes(1, incomingNonce: 1)));
      await gateway.submit(const InboundFrame(seq: 2, bytes: [0xFF, 0x01]));
      await gateway.submit(InboundFrame(seq: 3, bytes: _frameBytes(3, incomingNonce: 3)));

      await Future<void>.delayed(Duration.zero);

      expect(dispatched, [1, 3]);
      expect(errors, [2]);
      expect(gateway.queuedCount, 0);
      await sub.cancel();
      await gateway.close();
    });

    test(
        'IsolateInboundGateway: malformed frame이 isolate를 죽이지 않고 connection close + pending cleanup으로 이어진다',
        () async {
      final communicator = _FakeCommunicator();
      final gateway = IsolateInboundGateway();
      await gateway.start();
      communicator.attachGatewayForTest(gateway);

      final future = communicator.sendRequest<TestData, TestData>(
        TestData()..index = 1,
        timeout: const Duration(seconds: 5),
      );
      await Future<void>.delayed(Duration.zero);

      // malformed PacketBase bytes: invalid wire type → worker decode 실패.
      await communicator.receiveFrameForTest(const [0xFF, 0x01]);

      await expectLater(
        future,
        throwsA(
          isA<StateError>().having(
            (error) => error.message,
            'message',
            contains('connection closed'),
          ),
        ),
      );
      expect(communicator.isAlive, isFalse);
    });

    test('FrameReorderBuffer: pendingCount와 clear가 backlog를 노출하고 초기화한다', () {
      final buffer = FrameReorderBuffer();
      DecodedFrame frame(int seq) => DecodedFrame(
            seq: seq,
            typeName: 'T',
            data: const [],
            incomingNonce: seq,
            responseNonce: 0,
          );

      // seq=2,3은 seq=1을 기다리며 buffer에 잔류한다.
      buffer.release(frame(2));
      buffer.release(frame(3));
      expect(buffer.pendingCount, 2);

      // clear는 backlog를 비우고 nextSeq를 1로 초기화한다.
      buffer.clear();
      expect(buffer.pendingCount, 0);
      expect(buffer.release(frame(1)).map((f) => f.seq), [1]);
    });

    test(
        'IsolateInboundGateway: 반복 start/submit/close가 residual backlog를 남기지 않는다',
        () async {
      for (var round = 0; round < 20; round++) {
        final gateway = IsolateInboundGateway(capacity: 8);
        await gateway.start();
        final received = <int>[];
        final sub = gateway.results.listen((frame) => received.add(frame.seq));

        for (var seq = 1; seq <= 10; seq++) {
          await gateway.submit(InboundFrame(seq: seq, bytes: _frameBytes(seq)));
        }
        for (var i = 0; i < 200; i++) {
          if (received.length == 10) break;
          await Future<void>.delayed(const Duration(milliseconds: 5));
        }
        expect(received, List<int>.generate(10, (i) => i + 1));

        await sub.cancel();
        await gateway.close();

        // close 후 in-flight/queued backlog가 모두 0이다.
        expect(gateway.inFlightCount, 0);
        expect(gateway.queuedCount, 0);
      }
    });
  });
}