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 'package:protobuf/protobuf.dart';
import 'base_client.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}) =>
      SecureSocket.connect(host, port,
          context: context ?? SecurityContext.defaultContext);

  int? _length = null;
  late List<int> _arrivedData = [];

  ProtobufClient(this._socket, int heartbeatIntervalTime, int heartbeatWaitTime,
      Map<String, GeneratedMessage Function(List<int>)> parserMap) {
    initSelf(this);
    initHeartbeat(heartbeatIntervalTime, heartbeatWaitTime);
    _transport = _TcpSocketTransport(_socket, _headerSize);
    isAlive = true;
    parserMap.addAll({
      (HeartBeat).toString(): HeartBeat.fromBuffer,
    });
    super.initialize(parserMap, transport: _transport);
    _socket.listen(onData, onError: onError).asFuture<void>().then((_) {
      close();
    });
    addListener(onHeartBeat);
    sendHeartBeat();
  }

  void onError(dynamic e) {}

  void onData(Uint8List data) {
    try {
      _arrivedData.addAll(data);
      _parsing();
    } on Exception {
      // Ignore malformed partial data and keep the socket state unchanged.
    }
  }

  void _parsing() {
    while (true) {
      if (_length == null) {
        if (_arrivedData.length < _headerSize) {
          return;
        }
        final length = Uint8List.fromList(_arrivedData.sublist(0, _headerSize))
            .buffer
            .asByteData()
            .getInt32(0);
        if (length == 0) {
          _length = null;
          _arrivedData.clear();
          return;
        }
        if (length < 0 || length > maxPacketSize) {
          unawaited(close());
          return;
        }
        _length = length;
      }

      if (_arrivedData.length < _length! + _headerSize) {
        return;
      }

      final packetBytes = List<int>.from(
          _arrivedData.sublist(_headerSize, _headerSize + _length!));
      _arrivedData = _arrivedData.sublist(_headerSize + _length!);
      _length = null;
      final common = PacketBase.fromBuffer(packetBytes);
      onReceivedData(common.typeName, common.data,
          incomingNonce: common.nonce, responseNonce: common.responseNonce);
      sendHeartBeat();
    }
  }

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

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();
    final header = Uint8List(_headerSize)
      ..buffer.asByteData().setInt32(0, baseBytes.length);
    _socket.add([...header, ...baseBytes]);
    await _socket.flush();
  }

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