connectChat function

Implementation

ChatConnection connectChat({
  required int chatroomId,
  int? channelId,
  List<String>? channels,
  required void Function(Map<String, dynamic> event) onEvent,
  void Function(bool connected, List<String> subscribed)? onState,
  void Function(String reason)? onDrop,
  void Function(ChatDrop drop)? onDropInfo,
  void Function(Object error, StackTrace stack)? onError,
  Map<String, String>? endpoints,
  Duration timeout = const Duration(seconds: 10),
  Backoff? backoff,
}) {
  final feeds = channels ?? Realtime.chatChannels(chatroomId, channelId);
  Session? session = Session(feeds);
  final http = HttpClient()..connectionTimeout = timeout;
  final delays = backoff ?? Backoff();
  final random = Random();
  final clock = Stopwatch()..start();
  int now() => clock.elapsedMilliseconds;
  final done = Completer<void>();
  var closed = false;
  WebSocket? ws;
  StreamSubscription<dynamic>? frames;
  Timer? deadline;
  Timer? retry;
  Timer? refreshRetry;
  var clientId = '';

  // After close(), nothing is called back, onError included: what still
  // goes wrong (a callback that closed the connection, then threw) goes to
  // stderr.
  void report(Object error, StackTrace stack) {
    final handler = onError;
    if (!closed && handler != null) {
      try {
        handler(error, stack);
        return;
      } catch (e, s) {
        stderr.writeln('kicksdk: onError threw: $e\n$s');
      }
    }
    stderr.writeln('kicksdk: $error\n$stack');
  }

  // Runs a user callback. What it throws is reported, never thrown into the
  // middle of a batch of actions.
  void guard(void Function() callback) {
    try {
      callback();
    } catch (e, s) {
      report(e, s);
    }
  }

  // Lets go of the socket in `ws`, if any, at once: its frames are no longer
  // read, and its close is not waited for.
  WebSocket? detach() {
    final socket = ws;
    ws = null;
    frames?.cancel().ignore();
    frames = null;
    deadline?.cancel();
    refreshRetry?.cancel();
    if (socket != null) _discard(socket);
    return socket;
  }

  void stop() {
    if (closed) return;
    closed = true;
    retry?.cancel();
    detach();
    session?.dispose();
    session = null;
    http.close(force: true);
    done.complete();
  }

  void state() {
    final s = session;
    final handler = onState;
    if (closed || handler == null || s == null) return;
    final connected = s.isConnected;
    final subscribed = [
      for (final f in feeds)
        if (s.isSubscribed(f)) f,
    ];
    guard(() => handler(connected, subscribed));
  }

  late final void Function() refreshToken;
  late final void Function(List<Map<String, dynamic>>) apply;
  late final Future<void> Function() connect;

  void schedule() {
    deadline?.cancel();
    deadline = null;
    final s = session;
    if (closed || s == null || ws == null) return;
    final at = s.deadline;
    if (at == null) return;
    deadline = Timer(Duration(milliseconds: max(0, at - now())), () {
      final s = session;
      if (!closed && s != null) apply(s.tick(now()));
    });
  }

  // The connection in `ws` is lost, or none could be made: report it, and
  // reconnect after the backoff's wait for `retryAfterMs` (KICK.md §6.1).
  // connect() makes a new client id for every connection, which a terminal
  // close needs.
  void lost(String reason, {int? code, bool terminal = false, int retryAfterMs = 0}) {
    final hadSocket = detach() != null;
    if (closed) return;
    // The feeds it had come back unsubscribed, which reports the state.
    final lostFeeds = hadSocket ? session?.closed() ?? const [] : const <Map<String, dynamic>>[];
    final wait = delays.nextDelayFor(retryAfterMs, random.nextDouble());
    final info = ChatDrop(reason, code: code, terminal: terminal, retryIn: wait);
    final handler = onDrop;
    if (handler != null) guard(() => handler(reason));
    final infoHandler = onDropInfo;
    if (!closed && infoHandler != null) guard(() => infoHandler(info));
    if (closed) return;
    if (lostFeeds.isNotEmpty) {
      apply(lostFeeds);
    } else if (hadSocket) {
      state();
    }
    if (closed) return;
    retry?.cancel();
    retry = Timer(wait, () => connect().catchError(report));
  }

  // The socket closed by itself (no close action first): its close code
  // says whether that was terminal.
  void ended(WebSocket socket, [Object? error]) {
    if (ws != socket) return;
    final code = socket.closeCode;
    if (code == null) {
      lost(error == null ? 'socket closed' : 'socket failed: $error');
      return;
    }
    final text = socket.closeReason ?? '';
    var retryAfterMs = 0;
    try {
      retryAfterMs = Realtime.retryAfterMs(code);
    } catch (e, s) {
      report(e, s);
    }
    lost(
      'socket closed: $code${text.isEmpty ? '' : ' $text'}',
      code: code,
      terminal: retryAfterMs > 0,
      retryAfterMs: retryAfterMs,
    );
  }

  apply = (actions) {
    if (closed) return;
    var changed = false;
    try {
      batch:
      for (final a in actions) {
        if (closed) return;
        switch (a['action']) {
          case 'send':
            try {
              ws?.add(a['text'] as String);
            } on StateError {
              // The socket is closing; its onDone takes it from here.
            }
          case 'close':
            // Treated as lost at once: dart:io would hold the socket's
            // onDone for seconds on a half-open connection. What the batch
            // has after it was for that connection.
            changed = false;
            lost(
              a['reason'] as String,
              code: a['code'] as int,
              terminal: a['terminal'] == true,
              retryAfterMs: a['retry_after_ms'] as int,
            );
            break batch;
          case 'refresh_token':
            refreshToken();
          case 'subscribed':
            if (a['channel'] == feeds.first) delays.reset();
            changed = true;
          case 'connected' || 'unsubscribed' || 'subscribe_failed':
            changed = true;
          case 'event':
            guard(() => onEvent(a));
        }
      }
      if (changed) state();
    } finally {
      // Always, so that a batch cut short cannot leave the session without
      // its next tick (silence, token refresh).
      schedule();
    }
  };

  // Fetches a new token for the socket now open. A failure is reported and
  // tried again, with a backoff, while that socket lasts: if it never works,
  // Kick ends the connection when the old token runs out, and the reconnect
  // gets a new one.
  refreshToken = () {
    final socket = ws;
    if (socket == null) return;
    final id = clientId;
    final wait = Backoff();
    Future<void> attempt() async {
      try {
        final r = await _send(http, Realtime.tokenRequest(id, endpoints: endpoints), timeout);
        final token = Realtime.parseToken(r.$1, r.$2);
        final s = session;
        if (!closed && ws == socket && s != null) apply(s.refresh(token, now()));
      } catch (e, st) {
        if (closed || ws != socket) return;
        report(e, st);
        refreshRetry?.cancel();
        refreshRetry = Timer(wait.nextDelayJittered(random.nextDouble()), attempt);
      }
    }

    attempt();
  };

  connect = () async {
    if (closed) return;
    // A new client, so a new guest token, for every connection: after a
    // terminal close, Kick would refuse the old one.
    final id = clientId = _uuid();
    final String token;
    final WebSocket socket;
    try {
      final where = channelId != null
          ? Realtime.chatConnectionRequest(channelId, id, endpoints: endpoints)
          : Realtime.connectionRequest(id, endpoints: endpoints);
      final c = await _send(http, where, timeout);
      final found = Realtime.parseConnection(c.$1, c.$2).firstWhere(
        (x) => x['provider'] == 'centrifugo',
        orElse: () => throw StateError('Kick offered no Centrifugo server'),
      );
      if (closed) return;
      final t = await _send(http, Realtime.tokenRequest(id, endpoints: endpoints), timeout);
      token = Realtime.parseToken(t.$1, t.$2);
      if (closed) return;
      socket = await _openSocket(found['url'] as String, http, timeout);
    } catch (e) {
      // After close(), a request fails because the client was shut: nothing
      // to report then.
      if (closed) return;
      lost('$e');
      return;
    }
    final s = session;
    if (closed || s == null) {
      _discard(socket);
      return;
    }
    ws = socket;
    frames = socket.listen(
      (frame) {
        final s = session;
        if (ws == socket && s != null && frame is String) apply(s.receive(frame, now()));
      },
      onError: (Object e) => ended(socket, e),
      onDone: () => ended(socket),
      cancelOnError: true,
    );
    apply(s.open(token, now()));
  };

  connect().catchError(report);
  return ChatConnection._(stop, done.future);
}