connectSocket function

Future<Socket> connectSocket(ConnectSocketOptions options)

Implementation

Future<io.Socket> connectSocket(ConnectSocketOptions options) async {
  // Input validation
  if (options.apiUserName.isEmpty) throw Exception('API username required.');
  if ((options.apiKey?.isEmpty ?? true) &&
      (options.apiToken?.isEmpty ?? true)) {
    throw Exception('API key or token required.');
  }
  if (options.link.isEmpty) throw Exception('Socket link required.');

  final refusalKey = _refusalKey(options);
  final remembered = _refusedMediaCredentials[refusalKey];
  if (remembered != null) {
    if (DateTime.now().isBefore(remembered.until)) throw remembered.error;
    _refusedMediaCredentials.remove(refusalKey);
  }

  // Validate API key or token format
  bool useKey = false;
  try {
    if (options.apiKey?.length == 64 &&
        await validateApiKeyToken(options.apiKey!)) {
      useKey = true;
    } else if (options.apiToken?.length == 64 &&
        await validateApiKeyToken(options.apiToken!)) {
      useKey = false;
    } else {
      throw Exception('Invalid API key or token format.');
    }
  } catch (error) {
    throw Exception('Invalid API key or token.');
  }

  // Configure socket options based on whether apiKey or apiToken is used
  final query = useKey
      ? {'apiUserName': options.apiUserName, 'apiKey': options.apiKey}
      : {'apiUserName': options.apiUserName, 'apiToken': options.apiToken};

  final socket = io.io('${options.link}/media', {
    'transports': ['websocket'],
    'query': query,
    'autoConnect': false,
  });

  final completer = Completer<io.Socket>();
  Timer? admissionTimer;
  var settled = false;
  final startedAt = DateTime.now();
  var transportOpened = false;
  String? rejectionReason;
  String? rejectionStage;

  late void Function(dynamic) handleConnectionSuccess;
  late void Function(dynamic) handleConnectError;
  late void Function(dynamic) handleSocketError;
  late void Function(dynamic) handleDisconnectBeforeReady;
  late void Function(dynamic) handleTransportOpen;
  late void Function(dynamic) handleConnectionRejected;

  void cleanupAdmissionListeners() {
    admissionTimer?.cancel();
    socket.off('connection-success', handleConnectionSuccess);
    socket.off('connect_error', handleConnectError);
    socket.off('error', handleSocketError);
    socket.off('disconnect', handleDisconnectBeforeReady);
    socket.off('connect', handleTransportOpen);
    socket.off('connection-rejected', handleConnectionRejected);
  }

  String? describeTransportError(dynamic error) {
    if (error == null) return null;
    final text = error is Map ? (error['message'] ?? error).toString() : error.toString();
    return text.length > 200 ? text.substring(0, 200) : text;
  }

  void finishWithError(MediaSocketConnectPhase phase, [dynamic error]) {
    if (settled) return;
    settled = true;
    cleanupAdmissionListeners();
    socket.disconnect();
    final failure = MediaSocketConnectException(
      phase: rejectionReason != null ? MediaSocketConnectPhase.rejected : phase,
      reason: rejectionReason,
      stage: rejectionStage,
      elapsedMs: DateTime.now().difference(startedAt).inMilliseconds,
      online: currentOnlineState(),
      transportError: describeTransportError(error),
      link: options.link,
    );
    _rememberRefusal(refusalKey, failure);
    completer.completeError(failure);
  }

  // A raw Socket.IO connection is only transport establishment. The media
  // server emits connection-success after its credential admission finishes;
  // do not expose the socket to callers before that application-level proof.
  handleConnectionSuccess = (_) {
    if (settled) return;
    settled = true;
    cleanupAdmissionListeners();
    _refusedMediaCredentials.remove(refusalKey);
    _stopReconnectingWhenRefused(socket, refusalKey, options.link);
    completer.complete(socket);
  };

  // Handle connection error
  handleConnectError = (error) => finishWithError(
      transportOpened
          ? MediaSocketConnectPhase.afterOpen
          : MediaSocketConnectPhase.beforeOpen,
      error);
  handleSocketError = (error) => finishWithError(
      transportOpened
          ? MediaSocketConnectPhase.afterOpen
          : MediaSocketConnectPhase.beforeOpen,
      error);
  handleDisconnectBeforeReady =
      (reason) => finishWithError(MediaSocketConnectPhase.afterOpen, reason);
  handleTransportOpen = (_) => transportOpened = true;
  // Media admission v2: the node says why before it closes the handshake. A refusal is final for these credentials,
  // so it is reported as such and never retried as a network error.
  handleConnectionRejected = (payload) {
    final data = payload is Map ? payload : const {};
    rejectionReason = normalizeRejectionReason(data['reason']);
    rejectionStage = data['stage'] is String ? data['stage'] as String : null;
    finishWithError(MediaSocketConnectPhase.rejected);
  };

  socket.on('connection-success', handleConnectionSuccess);
  socket.on('connect_error', handleConnectError);
  socket.on('error', handleSocketError);
  socket.on('disconnect', handleDisconnectBeforeReady);
  socket.on('connect', handleTransportOpen);
  socket.on('connection-rejected', handleConnectionRejected);

  admissionTimer = Timer(
    const Duration(seconds: 12),
    () => finishWithError(MediaSocketConnectPhase.timeout),
  );
  socket.connect();

  return completer.future;
}