connect method

@override Future connect()

Open websocket channel

Implementation

@override
Future connect() async {
  if (this.socket != null) {
    throw Exception('Already connected');
  }

  _sequence = 0;

  final WebSocket socket = WebSocket(url);

  socket.connection.listen((state) {
    if (state is Reconnected) {
      for (final query in queries.values) {
        socket.send(query.message);
      }
    }
  });

  final jsonStream = socket.messages.map((response) {
    if (response == null) {
      throw Exception('Invalid jsonrpc field in decoded object');
    }

    final Map<String, dynamic> json;
    try {
      json = jsonDecode(response);
    } catch (e) {
      throw Exception('The response is not a valid json object');
    }

    if (!json.containsKey('jsonrpc') || json['jsonrpc'] != '2.0') {
      throw Exception('Invalid jsonrpc field in decoded object');
    }

    return json;
  }).asBroadcastStream();

  // Commands
  jsonStream.where((message) => message.containsKey('id')).map((message) {
    final id = message['id'] as int;
    final result = message.containsKey('result') ? message['result'] : null;
    final error = message.containsKey('error') ? message['error'] : null;
    return RpcResponse(id: id, result: result, error: error);
  }).listen((message) {
    queries.remove(message.id)!.completer.complete(message);
  });

  // Subscriptions
  jsonStream
      .where((message) =>
          !message.containsKey('id') &&
          message.containsKey('params') &&
          (message['params'] as Map<String, dynamic>).containsKey('subscription'))
      .map((message) {
    final method = message['method'] as String;
    final params = message['params'] as Map<String, dynamic>;
    final subscription = params['subscription'] as String;
    final result = params.containsKey('result') ? params['result'] : null;

    return SubscriptionMessage(method: method, subscription: subscription, result: result);
  }).listen((message) {
    final StreamController? controller = getSubscriptionController(message.subscription);
    controller?.add(message);
  });

  this.socket = socket;
  await isReady();
}