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

// ignore_for_file: avoid_init_to_null, prefer_final_fields, deprecated_member_use

import 'dart:async';
import 'dart:html' as html;
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 html.WebSocket _ws;
  late final _WebSocketTransport _transport;

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

  /// Secure WebSocket (WSS) connection.
  ///
  /// Browser handles TLS via the user agent's trust store, so no
  /// `SecurityContext` is accepted on web.
  static Future<html.WebSocket> connectSecure(String host, int port,
          {String path = '/'}) =>
      _open('wss://$host:$port$path');

  static Future<html.WebSocket> _open(String url) {
    final ws = html.WebSocket(url);
    ws.binaryType = 'arraybuffer';
    if (ws.readyState == html.WebSocket.OPEN) {
      return Future.value(ws);
    }
    final completer = Completer<html.WebSocket>();
    late StreamSubscription<html.Event> openSub;
    late StreamSubscription<html.Event> errorSub;
    void cleanup() {
      openSub.cancel();
      errorSub.cancel();
    }

    openSub = ws.onOpen.listen((_) {
      if (!completer.isCompleted) {
        cleanup();
        completer.complete(ws);
      }
    });
    errorSub = ws.onError.listen((event) {
      if (!completer.isCompleted) {
        cleanup();
        completer.completeError(
            StateError('WebSocket failed to open: $url'), StackTrace.current);
      }
    });
    return completer.future;
  }

  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.onMessage.listen(_onMessage, onError: onError);
    _ws.onClose.listen((_) => close());
    addListener(onHeartBeat);
    sendHeartBeat();
  }

  void onError(dynamic e) {}

  void _onMessage(html.MessageEvent event) {
    final data = event.data;
    if (data is ByteBuffer) {
      _dispatch(data.asUint8List());
    } else if (data is Uint8List) {
      _dispatch(data);
    } else if (data is List<int>) {
      _dispatch(data);
    } else if (data is html.Blob) {
      final reader = html.FileReader();
      reader.readAsArrayBuffer(data);
      reader.onLoad.first.then((_) {
        final result = reader.result;
        if (result is ByteBuffer) {
          _dispatch(result.asUint8List());
        }
      });
    }
  }

  void _dispatch(List<int> bytes) {
    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 html.WebSocket _ws;

  _WebSocketTransport(this._ws);

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

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