proto-socket/dart/lib/src/ws_protobuf_client_io.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';

abstract class WsProtobufClient extends BaseClient<WsProtobufClient> {
  final WebSocket _ws;
  late final _WebSocketTransport _transport;

  /// Plain WebSocket connection.
  static Future<WebSocket> connect(String host, int port,
          {String path = '/'}) =>
      WebSocket.connect('ws://$host:$port$path');

  /// Secure WebSocket (WSS) connection.
  static Future<WebSocket> connectSecure(String host, int port,
      {String path = '/',
      SecurityContext? context,
      bool Function(X509Certificate certificate)? onBadCertificate}) {
    final client =
        HttpClient(context: context ?? SecurityContext.defaultContext);
    if (onBadCertificate != null) {
      client.badCertificateCallback = (certificate, _, __) {
        return onBadCertificate(certificate);
      };
    }
    return WebSocket.connect('wss://$host:$port$path', customClient: client);
  }

  WsProtobufClient(this._ws, int heartbeatIntervalTime, int heartbeatWaitTime,
      Map<String, GeneratedMessage Function(List<int>)> parserMap) {
    initSelf(this);
    initHeartbeat(heartbeatIntervalTime, heartbeatWaitTime);
    _transport = _WebSocketTransport(_ws);
    isAlive = true;
    parserMap.addAll({(HeartBeat).toString(): HeartBeat.fromBuffer});
    super.initialize(parserMap, transport: _transport);
    _ws.listen(_onMessage, onError: onError, onDone: close);
    addListener(onHeartBeat);
    sendHeartBeat();
  }

  void onError(dynamic e) {}

  void _onMessage(dynamic data) {
    final bytes = data is List<int> ? data : (data as Uint8List).toList();
    final common = PacketBase.fromBuffer(bytes);
    onReceivedData(common.typeName, common.data,
        incomingNonce: common.nonce, responseNonce: common.responseNonce);
    sendHeartBeat();
  }

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

class _WebSocketTransport implements Transport {
  final WebSocket _ws;

  _WebSocketTransport(this._ws);

  @override
  Future<void> writePacket(PacketBase base) async {
    _ws.add(base.writeToBuffer());
  }

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