proto-socket/dart/lib/src/protobuf_client.dart

// ignore_for_file: avoid_init_to_null, prefer_final_fields

import 'dart:io';
import 'dart:async';
import 'dart:typed_data';
import 'dart:collection';

import 'package:protobuf/protobuf.dart';
import 'base_client.dart';
import 'inbound_gateway_io.dart';
import 'packets/message_common.pb.dart';
import 'transport.dart';

const int maxPacketSize = 64 << 20;

abstract class ProtobufClient extends BaseClient<ProtobufClient> {
  final int _headerSize = 4;
  final Socket _socket;
  late final _TcpSocketTransport _transport;

  /// Plain TCP connection.
  static Future<Socket> connect(String host, int port) =>
      Socket.connect(host, port);

  /// SSL/TLS connection.
  static Future<Socket> connectSecure(String host, int port,
          {SecurityContext? context,
          bool Function(X509Certificate certificate)? onBadCertificate}) =>
      SecureSocket.connect(
        host,
        port,
        context: context ?? SecurityContext.defaultContext,
        onBadCertificate: onBadCertificate,
      );

  int? _length = null;
  final _TcpFrameBuffer _frameBuffer = _TcpFrameBuffer();

  StreamSubscription<Uint8List>? _subscription;
  bool _isParsing = false;

  ProtobufClient(this._socket, int heartbeatIntervalTime, int heartbeatWaitTime,
      Map<String, GeneratedMessage Function(List<int>)> parserMap) {
    initSelf(this);
    // 작은 프레임의 fixed latency(40ms대)를 막기 위해 Nagle을 끈다. dial한
    // client socket과 server가 accept한 socket이 모두 이 생성자를 거치므로 양쪽
    // roundtrip이 함께 적용된다.
    _socket.setOption(SocketOption.tcpNoDelay, true);
    initHeartbeat(heartbeatIntervalTime, heartbeatWaitTime);
    _transport = _TcpSocketTransport(_socket, _headerSize);
    isAlive = true;
    parserMap.addAll({
      HeartBeat.getDefault().info_.qualifiedMessageName: HeartBeat.fromBuffer,
    });
    super.initialize(parserMap, transport: _transport);
    // 제품 수신 경로를 단일 gateway-backed coordinator로 통일한다. start()는
    // await하지 않아도 _ready completer를 동기 구간에서 설정하므로, spawn 완료
    // 전에 도착한 첫 프레임도 submit에서 readiness를 기다려 drop되지 않는다.
    final gateway = IsolateInboundGateway();
    attachInboundGateway(gateway);
    unawaited(gateway.start());
    _subscription = _socket.listen(onData, onError: onError);
    _subscription!.asFuture<void>().then((_) {
      close();
    });
    addListener(onHeartBeat);
    sendHeartBeat();
  }

  void onError(dynamic e) {}

  void onData(Uint8List data) async {
    try {
      _frameBuffer.add(data);
      if (_isParsing) return;
      _isParsing = true;
      _subscription?.pause();
      try {
        await _parsing();
      } finally {
        _isParsing = false;
        _subscription?.resume();
      }
    } on Exception {
      // Ignore malformed partial data and keep the socket state unchanged.
    }
  }

  Future<void> _parsing() async {
    while (true) {
      if (_length == null) {
        if (_frameBuffer.available < _headerSize) {
          return;
        }
        final length = _frameBuffer.peekInt32Be();
        if (length == null) {
          return;
        }
        if (length == 0) {
          _length = null;
          _frameBuffer.clear();
          return;
        }
        if (length < 0 || length > maxPacketSize) {
          unawaited(close());
          return;
        }
        _length = length;
      }

      if (_frameBuffer.available < _length! + _headerSize) {
        return;
      }

      _frameBuffer.takeBytes(_headerSize);
      final packetBytes = _frameBuffer.takeBytes(_length!);
      _length = null;
      // raw frame을 coordinator로 넘긴다. gateway가 envelope decode를 off-loop로
      // 처리하고 seq 순서로 onReceivedData에 dispatch한다. submit await는 gateway
      // backpressure를 socket read loop로 전파한다.
      await onReceivedFrame(packetBytes);
      sendHeartBeat();
    }
  }

  @override
  Future<void> closeTransport() async {
    await _transport.close();
  }

  @override
  bool get isSourcePaused => _subscription?.isPaused ?? false;
}

class _TcpSocketTransport implements Transport {
  final Socket _socket;
  final int _headerSize;

  _TcpSocketTransport(this._socket, this._headerSize);

  @override
  Future<void> writePacket(PacketBase base) async {
    final baseBytes = base.writeToBuffer();
    if (baseBytes.length > maxPacketSize) {
      throw StateError('packet exceeds maximum size');
    }
    final header = Uint8List(_headerSize)
      ..buffer.asByteData().setInt32(0, baseBytes.length);
    _socket.add(header);
    _socket.add(baseBytes);
    // per-packet flush()를 await하지 않는다. flush를 매 패킷마다 await하면 burst/
    // sustained에서 send가 event-loop drain에 직렬화되어 throughput이 떨어진다.
    // 순서는 add()가 보장하고, 잔여 버퍼는 close()의 socket.close()가 flush한다.
  }

  @override
  Future<void> close() async {
    try {
      await _socket.close();
      _socket.destroy();
    } catch (_) {
      // already closed
    }
  }
}

class _TcpFrameBuffer {
  final Queue<Uint8List> _chunks = Queue<Uint8List>();
  int _available = 0;

  int get available => _available;

  void add(Uint8List chunk) {
    if (chunk.isEmpty) return;
    _chunks.add(chunk);
    _available += chunk.length;
  }

  void clear() {
    _chunks.clear();
    _available = 0;
  }

  int? peekInt32Be() {
    if (_available < 4) {
      return null;
    }
    final firstChunk = _chunks.first;
    if (firstChunk.length >= 4) {
      return firstChunk.buffer
          .asByteData(firstChunk.offsetInBytes, 4)
          .getInt32(0);
    }

    final tempBytes = Uint8List(4);
    int read = 0;
    for (final chunk in _chunks) {
      final take = chunk.length < (4 - read) ? chunk.length : (4 - read);
      tempBytes.setRange(read, read + take, chunk, 0);
      read += take;
      if (read == 4) break;
    }
    return tempBytes.buffer.asByteData(tempBytes.offsetInBytes, 4).getInt32(0);
  }

  Uint8List takeBytes(int count) {
    assert(_available >= count);

    final result = Uint8List(count);
    int written = 0;

    while (written < count) {
      final chunk = _chunks.first;
      final remaining = count - written;
      if (chunk.length <= remaining) {
        result.setRange(written, written + chunk.length, chunk, 0);
        written += chunk.length;
        _chunks.removeFirst();
      } else {
        result.setRange(written, count, chunk, 0);
        _chunks.removeFirst();
        _chunks.addFirst(Uint8List.view(chunk.buffer,
            chunk.offsetInBytes + remaining, chunk.length - remaining));
        written = count;
      }
    }
    _available -= count;
    return result;
  }
}