Merge pull request #559 from cypherstack/persistence

Reuse sockets

lib/electrumx_rpc/cached_electrumx.dart

@@ -4,43 +4,23 @@ import 'package:stackwallet/db/hive/db.dart';
 import 'package:stackwallet/electrumx_rpc/electrumx.dart';
 import 'package:stackwallet/utilities/enums/coin_enum.dart';
 import 'package:stackwallet/utilities/logger.dart';
-import 'package:stackwallet/utilities/prefs.dart';
 import 'package:string_validator/string_validator.dart';
 
 class CachedElectrumX {
-  final ElectrumX? electrumXClient;
-
-  final String server;
-  final int port;
-  final bool useSSL;
-
-  final Prefs prefs;
-  final List<ElectrumXNode> failovers;
+  final ElectrumX electrumXClient;
 
   static const minCacheConfirms = 30;
 
   const CachedElectrumX({
-    required this.server,
-    required this.port,
-    required this.useSSL,
-    required this.prefs,
-    required this.failovers,
-    this.electrumXClient,
+    required this.electrumXClient,
   });
 
   factory CachedElectrumX.from({
-    required ElectrumXNode node,
-    required Prefs prefs,
-    required List<ElectrumXNode> failovers,
-    ElectrumX? electrumXClient,
+    required ElectrumX electrumXClient,
   }) =>
       CachedElectrumX(
-          server: node.address,
-          port: node.port,
-          useSSL: node.useSSL,
-          prefs: prefs,
-          failovers: failovers,
-          electrumXClient: electrumXClient);
+        electrumXClient: electrumXClient,
+      );
 
   Future<Map<String, dynamic>> getAnonymitySet({
     required String groupId,
@@ -66,16 +46,7 @@ class CachedElectrumX {
         set = Map<String, dynamic>.from(cachedSet);
       }
 
-      final client = electrumXClient ??
-          ElectrumX(
-            host: server,
-            port: port,
-            useSSL: useSSL,
-            prefs: prefs,
-            failovers: failovers,
-          );
-
-      final newSet = await client.getAnonymitySet(
+      final newSet = await electrumXClient.getAnonymitySet(
         groupId: groupId,
         blockhash: set["blockHash"] as String,
       );
@@ -152,16 +123,8 @@ class CachedElectrumX {
       final cachedTx = DB.instance.get<dynamic>(
           boxName: DB.instance.boxNameTxCache(coin: coin), key: txHash) as Map?;
       if (cachedTx == null) {
-        final client = electrumXClient ??
-            ElectrumX(
-              host: server,
-              port: port,
-              useSSL: useSSL,
-              prefs: prefs,
-              failovers: failovers,
-            );
-        final Map<String, dynamic> result =
-            await client.getTransaction(txHash: txHash, verbose: verbose);
+        final Map<String, dynamic> result = await electrumXClient
+            .getTransaction(txHash: txHash, verbose: verbose);
 
         result.remove("hex");
         result.remove("lelantusData");
@@ -202,16 +165,8 @@ class CachedElectrumX {
 
       final startNumber = cachedSerials.length;
 
-      final client = electrumXClient ??
-          ElectrumX(
-            host: server,
-            port: port,
-            useSSL: useSSL,
-            prefs: prefs,
-            failovers: failovers,
-          );
-
-      final serials = await client.getUsedCoinSerials(startNumber: startNumber);
+      final serials =
+          await electrumXClient.getUsedCoinSerials(startNumber: startNumber);
       List<String> newSerials = [];
 
       for (final element in (serials["serials"] as List)) {

lib/electrumx_rpc/electrumx.dart

@@ -132,6 +132,11 @@ class ElectrumX {
 
       final response = await _rpcClient!.request(jsonRequestString);
 
+      print("=================================================");
+      print("TYPE: ${response.runtimeType}");
+      print("RESPONSE: $response");
+      print("=================================================");
+
       if (response["error"] != null) {
         if (response["error"]
             .toString()
@@ -310,6 +315,13 @@ class ElectrumX {
         requestID: requestID,
         command: 'blockchain.headers.subscribe',
       );
+      if (response["result"] == null) {
+        Logging.instance.log(
+          "getBlockHeadTip returned null response",
+          level: LogLevel.Error,
+        );
+        throw 'getBlockHeadTip returned null response';
+      }
       return Map<String, dynamic>.from(response["result"] as Map);
     } catch (e) {
       rethrow;

lib/electrumx_rpc/rpc.dart

@@ -1,7 +1,9 @@
 import 'dart:async';
 import 'dart:convert';
 import 'dart:io';
+import 'dart:typed_data';
 
+import 'package:mutex/mutex.dart';
 import 'package:stackwallet/utilities/logger.dart';
 
 // hacky fix to receive large jsonrpc responses
@@ -12,65 +14,219 @@ class JsonRPC {
     this.useSSL = false,
     this.connectionTimeout = const Duration(seconds: 60),
   });
-  bool useSSL;
-  String host;
-  int port;
-  Duration connectionTimeout;
+  final bool useSSL;
+  final String host;
+  final int port;
+  final Duration connectionTimeout;
 
-  Future<dynamic> request(String jsonRpcRequest) async {
-    Socket? socket;
-    final completer = Completer<dynamic>();
-    final List<int> responseData = [];
+  final _requestMutex = Mutex();
+  final _JsonRPCRequestQueue _requestQueue = _JsonRPCRequestQueue();
+  Socket? _socket;
+  StreamSubscription<Uint8List>? _subscription;
 
-    void dataHandler(List<int> data) {
-      responseData.addAll(data);
+  void _dataHandler(List<int> data) {
+    if (_requestQueue.isEmpty) {
+      // probably just return although this case should never actually hit
+      return;
+    }
 
-      // 0x0A is newline
-      // https://electrumx-spesmilo.readthedocs.io/en/latest/protocol-basics.html
-      if (data.last == 0x0A) {
-        try {
-          final response = json.decode(String.fromCharCodes(responseData));
-          completer.complete(response);
-        } catch (e, s) {
-          Logging.instance
-              .log("JsonRPC json.decode: $e\n$s", level: LogLevel.Error);
-          completer.completeError(e, s);
-        } finally {
-          socket?.destroy();
+    final req = _requestQueue.next;
+    req.appendDataAndCheckIfComplete(data);
+
+    if (req.isComplete) {
+      _onReqCompleted(req);
+    }
+  }
+
+  void _errorHandler(Object error, StackTrace trace) {
+    Logging.instance.log(
+      "JsonRPC errorHandler: $error\n$trace",
+      level: LogLevel.Error,
+    );
+
+    final req = _requestQueue.next;
+    req.completer.completeError(error, trace);
+    _onReqCompleted(req);
+  }
+
+  void _doneHandler() {
+    Logging.instance.log(
+      "JsonRPC doneHandler: "
+      "connection closed to $host:$port, destroying socket",
+      level: LogLevel.Info,
+    );
+
+    if (_requestQueue.isNotEmpty) {
+      Logging.instance.log(
+        "JsonRPC doneHandler: queue not empty but connection closed, "
+        "completing pending requests with errors",
+        level: LogLevel.Error,
+      );
+
+      for (final req in _requestQueue.queue) {
+        if (!req.isComplete) {
+          try {
+            throw Exception(
+              "JsonRPC doneHandler: socket closed "
+              "before request could complete",
+            );
+          } catch (e, s) {
+            req.completer.completeError(e, s);
+          }
         }
       }
+      _requestQueue.clear();
     }
 
-    void errorHandler(Object error, StackTrace trace) {
-      Logging.instance
-          .log("JsonRPC errorHandler: $error\n$trace", level: LogLevel.Error);
-      completer.completeError(error, trace);
-      socket?.destroy();
+    disconnect();
+  }
+
+  void _onReqCompleted(_JsonRPCRequest req) {
+    _requestQueue.remove(req);
+    if (_requestQueue.isNotEmpty) {
+      _sendNextAvailableRequest();
+    }
+  }
+
+  void _sendNextAvailableRequest() {
+    if (_requestQueue.isEmpty) {
+      // TODO handle properly
+      throw Exception("JSON RPC queue empty");
     }
 
-    void doneHandler() {
-      socket?.destroy();
-    }
+    final req = _requestQueue.next;
 
-    if (useSSL) {
-      await SecureSocket.connect(host, port,
-          timeout: connectionTimeout,
-          onBadCertificate: (_) => true).then((Socket sock) {
-        socket = sock;
-        socket?.listen(dataHandler,
-            onError: errorHandler, onDone: doneHandler, cancelOnError: true);
-      });
+    _socket!.write('${req.jsonRequest}\r\n');
+
+    req.initiateTimeout(const Duration(seconds: 10));
+    // Logging.instance.log(
+    //   "JsonRPC request: wrote request ${req.jsonRequest} "
+    //   "to socket $host:$port",
+    //   level: LogLevel.Info,
+    // );
+  }
+
+  Future<dynamic> request(String jsonRpcRequest) async {
+    // todo: handle this better?
+    // Do we need to check the subscription, too?
+    await _requestMutex.protect(() async {
+      if (_socket == null) {
+        Logging.instance.log(
+          "JsonRPC request: opening socket $host:$port",
+          level: LogLevel.Info,
+        );
+        await connect();
+      }
+    });
+
+    final req = _JsonRPCRequest(
+      jsonRequest: jsonRpcRequest,
+      completer: Completer<dynamic>(),
+    );
+
+    _requestQueue.add(req);
+
+    // if this is the only/first request then send it right away
+    if (_requestQueue.length == 1) {
+      _sendNextAvailableRequest();
     } else {
-      await Socket.connect(host, port, timeout: connectionTimeout)
-          .then((Socket sock) {
-        socket = sock;
-        socket?.listen(dataHandler,
-            onError: errorHandler, onDone: doneHandler, cancelOnError: true);
-      });
+      // Logging.instance.log(
+      //   "JsonRPC request: queued request $jsonRpcRequest "
+      //   "to socket $host:$port",
+      //   level: LogLevel.Info,
+      // );
     }
 
-    socket?.write('$jsonRpcRequest\r\n');
+    return req.completer.future.onError(
+      (error, stackTrace) =>
+          Exception("return req.completer.future.onError: $error"),
+    );
+  }
 
-    return completer.future;
+  void disconnect() {
+    // TODO: maybe clear req queue here and wrap in mutex?
+    _subscription?.cancel().then((_) => _subscription = null);
+    _socket?.destroy();
+    _socket = null;
+  }
+
+  Future<void> connect() async {
+    if (useSSL) {
+      _socket ??= await SecureSocket.connect(
+        host,
+        port,
+        timeout: connectionTimeout,
+        onBadCertificate: (_) => true,
+      ); // TODO do not automatically trust bad certificates
+    } else {
+      _socket ??= await Socket.connect(
+        host,
+        port,
+        timeout: connectionTimeout,
+      );
+    }
+    await _subscription?.cancel();
+    _subscription = _socket!.listen(
+      _dataHandler,
+      onError: _errorHandler,
+      onDone: _doneHandler,
+      cancelOnError: true,
+    );
   }
 }
+
+class _JsonRPCRequestQueue {
+  final List<_JsonRPCRequest> _rq = [];
+
+  void add(_JsonRPCRequest req) => _rq.add(req);
+
+  bool remove(_JsonRPCRequest req) => _rq.remove(req);
+
+  void clear() => _rq.clear();
+
+  bool get isEmpty => _rq.isEmpty;
+  bool get isNotEmpty => _rq.isNotEmpty;
+  int get length => _rq.length;
+  _JsonRPCRequest get next => _rq.first;
+  List<_JsonRPCRequest> get queue => _rq.toList(growable: false);
+}
+
+class _JsonRPCRequest {
+  final String jsonRequest;
+  final Completer<dynamic> completer;
+  final List<int> _responseData = [];
+
+  _JsonRPCRequest({required this.jsonRequest, required this.completer});
+
+  void appendDataAndCheckIfComplete(List<int> data) {
+    _responseData.addAll(data);
+    // 0x0A is newline
+    // https://electrumx-spesmilo.readthedocs.io/en/latest/protocol-basics.html
+    if (data.last == 0x0A) {
+      try {
+        final response = json.decode(String.fromCharCodes(_responseData));
+        completer.complete(response);
+      } catch (e, s) {
+        Logging.instance.log(
+          "JsonRPC json.decode: $e\n$s",
+          level: LogLevel.Error,
+        );
+        completer.completeError(e, s);
+      }
+    }
+  }
+
+  void initiateTimeout(Duration timeout) {
+    Future<void>.delayed(timeout).then((_) {
+      if (!isComplete) {
+        try {
+          throw Exception("_JsonRPCRequest timed out: $jsonRequest");
+        } catch (e, s) {
+          completer.completeError(e, s);
+        }
+      }
+    });
+  }
+
+  bool get isComplete => completer.isCompleted;
+}

lib/services/coins/bitcoin/bitcoin_wallet.dart

@@ -1342,16 +1342,14 @@ class BitcoinWallet extends CoinServiceAPI
             ))
         .toList();
     final newNode = await getCurrentNode();
-    _cachedElectrumXClient = CachedElectrumX.from(
-      node: newNode,
-      prefs: _prefs,
-      failovers: failovers,
-    );
     _electrumXClient = ElectrumX.from(
       node: newNode,
       prefs: _prefs,
       failovers: failovers,
     );
+    _cachedElectrumXClient = CachedElectrumX.from(
+      electrumXClient: _electrumXClient,
+    );
 
     if (shouldRefresh) {
       unawaited(refresh());

lib/services/coins/bitcoincash/bitcoincash_wallet.dart

@@ -1234,16 +1234,14 @@ class BitcoinCashWallet extends CoinServiceAPI
             ))
         .toList();
     final newNode = await getCurrentNode();
-    _cachedElectrumXClient = CachedElectrumX.from(
-      node: newNode,
-      prefs: _prefs,
-      failovers: failovers,
-    );
     _electrumXClient = ElectrumX.from(
       node: newNode,
       prefs: _prefs,
       failovers: failovers,
     );
+    _cachedElectrumXClient = CachedElectrumX.from(
+      electrumXClient: _electrumXClient,
+    );
 
     if (shouldRefresh) {
       unawaited(refresh());

lib/services/coins/coin_service.dart

@@ -56,17 +56,7 @@ abstract class CoinServiceAPI {
       prefs: prefs,
     );
     final cachedClient = CachedElectrumX.from(
-      node: electrumxNode,
-      failovers: failovers
-          .map((e) => ElectrumXNode(
-                address: e.host,
-                port: e.port,
-                name: e.name,
-                id: e.id,
-                useSSL: e.useSSL,
-              ))
-          .toList(),
-      prefs: prefs,
+      electrumXClient: client,
     );
     switch (coin) {
       case Coin.firo:

lib/services/coins/dogecoin/dogecoin_wallet.dart

@@ -1194,16 +1194,14 @@ class DogecoinWallet extends CoinServiceAPI
             ))
         .toList();
     final newNode = await getCurrentNode();
-    _cachedElectrumXClient = CachedElectrumX.from(
-      node: newNode,
-      prefs: _prefs,
-      failovers: failovers,
-    );
     _electrumXClient = ElectrumX.from(
       node: newNode,
       prefs: _prefs,
       failovers: failovers,
     );
+    _cachedElectrumXClient = CachedElectrumX.from(
+      electrumXClient: _electrumXClient,
+    );
 
     if (shouldRefresh) {
       unawaited(refresh());

lib/services/coins/ecash/ecash_wallet.dart

@@ -410,16 +410,14 @@ class ECashWallet extends CoinServiceAPI
             ))
         .toList();
     final newNode = await getCurrentNode();
-    _cachedElectrumXClient = CachedElectrumX.from(
-      node: newNode,
-      prefs: _prefs,
-      failovers: failovers,
-    );
     _electrumXClient = ElectrumX.from(
       node: newNode,
       prefs: _prefs,
       failovers: failovers,
     );
+    _cachedElectrumXClient = CachedElectrumX.from(
+      electrumXClient: _electrumXClient,
+    );
 
     if (shouldRefresh) {
       unawaited(refresh());

lib/services/coins/firo/firo_wallet.dart

@@ -1840,16 +1840,14 @@ class FiroWallet extends CoinServiceAPI
         )
         .toList();
     final newNode = await _getCurrentNode();
-    _cachedElectrumXClient = CachedElectrumX.from(
-      node: newNode,
-      prefs: _prefs,
-      failovers: failovers,
-    );
     _electrumXClient = ElectrumX.from(
       node: newNode,
       prefs: _prefs,
       failovers: failovers,
     );
+    _cachedElectrumXClient = CachedElectrumX.from(
+      electrumXClient: _electrumXClient,
+    );
 
     if (shouldRefresh) {
       unawaited(refresh());

lib/services/coins/litecoin/litecoin_wallet.dart

@@ -1325,16 +1325,14 @@ class LitecoinWallet extends CoinServiceAPI
             ))
         .toList();
     final newNode = await getCurrentNode();
-    _cachedElectrumXClient = CachedElectrumX.from(
-      node: newNode,
-      prefs: _prefs,
-      failovers: failovers,
-    );
     _electrumXClient = ElectrumX.from(
       node: newNode,
       prefs: _prefs,
       failovers: failovers,
     );
+    _cachedElectrumXClient = CachedElectrumX.from(
+      electrumXClient: _electrumXClient,
+    );
 
     if (shouldRefresh) {
       unawaited(refresh());

lib/services/coins/namecoin/namecoin_wallet.dart

@@ -1314,16 +1314,14 @@ class NamecoinWallet extends CoinServiceAPI
             ))
         .toList();
     final newNode = await getCurrentNode();
-    _cachedElectrumXClient = CachedElectrumX.from(
-      node: newNode,
-      prefs: _prefs,
-      failovers: failovers,
-    );
     _electrumXClient = ElectrumX.from(
       node: newNode,
       prefs: _prefs,
       failovers: failovers,
     );
+    _cachedElectrumXClient = CachedElectrumX.from(
+      electrumXClient: _electrumXClient,
+    );
 
     if (shouldRefresh) {
       unawaited(refresh());

lib/services/coins/particl/particl_wallet.dart

@@ -1242,16 +1242,14 @@ class ParticlWallet extends CoinServiceAPI
             ))
         .toList();
     final newNode = await getCurrentNode();
-    _cachedElectrumXClient = CachedElectrumX.from(
-      node: newNode,
-      prefs: _prefs,
-      failovers: failovers,
-    );
     _electrumXClient = ElectrumX.from(
       node: newNode,
       prefs: _prefs,
       failovers: failovers,
     );
+    _cachedElectrumXClient = CachedElectrumX.from(
+      electrumXClient: _electrumXClient,
+    );
 
     if (shouldRefresh) {
       unawaited(refresh());