connectRelay method

Future<Tuple<bool, String>> connectRelay({ required String dirtyUrl, required ConnectionSource connectionSource, String? authPubkey, int connectTimeout = DEFAULT_WEB_SOCKET_CONNECT_TIMEOUT, })

Connects to a relay to the relay pool. Returns a tuple with the first element being a boolean indicating success \ and the second element being a string with the error message if any.

Implementation

Future<Tuple<bool, String>> connectRelay({
  required String dirtyUrl,
  required ConnectionSource connectionSource,
  String? authPubkey,
  int connectTimeout = DEFAULT_WEB_SOCKET_CONNECT_TIMEOUT,
}) async {
  String? url = cleanRelayUrl(dirtyUrl);
  if (url == null) {
    updateRelayConnectivity();
    return Tuple(false, "unclean url");
  }
  if (globalState.blockedRelays.contains(url)) {
    updateRelayConnectivity();
    return Tuple(false, "relay is blocked");
  }

  final connectionKey = authPubkey == null
      ? RelayConnectionKey.anonymous(url)
      : RelayConnectionKey.authenticated(url, authPubkey);

  if (isConnectionOpen(connectionKey)) {
    Logger.log.t(() => "relay already connected: $connectionKey");
    updateRelayConnectivity();
    return Tuple(true, "");
  }

  if (isConnectionConnecting(connectionKey)) {
    Logger.log.t(() => "relay is already connecting: $connectionKey");
    final inFlightConnect = _connectReadyCompleters[connectionKey];
    if (inFlightConnect != null) {
      final connected = await inFlightConnect.future;
      updateRelayConnectivity();
      return Tuple(
        connected,
        connected
            ? "relay finished connecting"
            : "relay failed while connecting",
      );
    }
    updateRelayConnectivity();
    return Tuple(false, "relay is still connecting");
  }
  RelayConnectivity? relayConnectivity = globalState.relays[connectionKey];
  final connectCompleter = Completer<bool>();
  _connectReadyCompleters[connectionKey] = connectCompleter;

  try {
    if (relayConnectivity == null) {
      relayConnectivity = RelayConnectivity<T>(
        key: connectionKey,
        relay: Relay(url: url, connectionSource: connectionSource),
        specificEngineData: engineAdditionalDataFactory?.call(),
      );
      globalState.relays[connectionKey] = relayConnectivity;
    }

    relayConnectivity.relay.tryingToConnect();

    /// TO BE REMOVED, ONCE WE FIND A WAY OF AVOIDING PROBLEM WHEN CONNECTING TO THIS
    if (url.startsWith("wss://brb.io")) {
      relayConnectivity.relay.failedToConnect();
      if (!connectCompleter.isCompleted) {
        connectCompleter.complete(false);
      }
      if (identical(
        _connectReadyCompleters[connectionKey],
        connectCompleter,
      )) {
        _connectReadyCompleters.remove(connectionKey);
      }
      updateRelayConnectivity();
      return Tuple(false, "bad relay");
    }

    Logger.log.i(() => "connecting to relay $dirtyUrl");

    // a fresh socket for a key we may already know: nothing the previous one
    // authenticated carries over
    _forgetAuthState(connectionKey);
    relayConnectivity.relayTransport = nostrTransportFactory(
      url,
      onReconnect: () {
        // the relay accepted our AUTH on the socket that just died, not on
        // this one; the binding survives, the authentication does not
        _forgetAuthState(connectionKey);
        reSubscribeInFlightSubscriptions(relayConnectivity!);
        updateRelayConnectivity();
      },
      onDisconnect: (code, error, reason) {
        relayConnectivity!.stats.connectionErrors++;
        // the transport reconnects under us and keeps its message stream
        // open, so this is the only notice we get that the socket the relay
        // authenticated died
        _forgetAuthState(connectionKey);
        // the requests died with that socket too; onReconnect replays them
        relayConnectivity.stats.openRequestIds.clear();
        updateRelayConnectivity();
      },
    );
    // Start listening immediately so we don't miss early frames such as
    // relay AUTH challenges that may arrive before the transport reports
    // itself fully open.
    _startListeningToSocket(relayConnectivity);
    final opened = await _waitForTransportOpen(
      relayConnectivity.relayTransport!,
      timeoutSeconds: connectTimeout,
    );
    if (!opened) {
      throw TimeoutException(
        "Future not completed",
        Duration(seconds: connectTimeout),
      );
    }

    Logger.log.i(() => "connected to relay: $url");
    relayConnectivity.relay.succeededToConnect();
    relayConnectivity.stats.connections++;
    getRelayInfo(url).then((info) {
      relayConnectivity!.relayInfo = info;
    });
    if (!connectCompleter.isCompleted) {
      connectCompleter.complete(true);
    }
    if (identical(_connectReadyCompleters[connectionKey], connectCompleter)) {
      _connectReadyCompleters.remove(connectionKey);
    }
    updateRelayConnectivity();
    return Tuple(true, "");
  } catch (e) {
    Logger.log.e(() => "!! could not connect to $url -> $e");
    await relayConnectivity!.close();
  }
  relayConnectivity.relay.failedToConnect();
  relayConnectivity.stats.connectionErrors++;
  if (!connectCompleter.isCompleted) {
    connectCompleter.complete(false);
  }
  if (identical(_connectReadyCompleters[connectionKey], connectCompleter)) {
    _connectReadyCompleters.remove(connectionKey);
  }
  updateRelayConnectivity();
  return Tuple(false, "could not connect to $url");
}