start method

Future<void> start(FirehoseMessageHandler onMessage)

Runs until stop is called: connects, hands every message to onMessage, and reconnects with exponential backoff when the connection fails or closes.

A handler that throws is reported to onError and the message is skipped — one bad message must not take the consumer down. Its seq is not written, but a later successful message will advance the cursor past it, so the skipped message is not retried on restart: retry and dead-lettering belong to the handler.

Implementation

Future<void> start(final FirehoseMessageHandler onMessage) async {
  var consecutiveFailures = 0;
  // Anchor the interval here rather than treating "never flushed" as overdue,
  // which would make the first message of every run a special case.
  _lastFlushAt ??= DateTime.now();

  while (!_stopped) {
    final uptime = Stopwatch()..start();
    try {
      final cursor = _cursorDiscarded ? null : await _cursorStore.find();
      final connection = _connection = await _connect(cursor);
      try {
        if (_stopped) break;
        // Completes when the relay closes the stream; throws on a stream
        // error. Either way we fall through to the reconnect delay below.
        await _consume(connection.stream, onMessage);
      } finally {
        _connection = null;
        await _closeQuietly(connection);
      }
    } on Object catch (e, st) {
      // `Object`, not `Exception`: an `Error` escaping here (a `RangeError`
      // from a decoder edge case, say) would otherwise kill the consumer
      // permanently after a single attempt.
      await _handleConnectionFailure(e, st);
    }

    // The cursor must be durable before the next subscription asks for it,
    // otherwise the reconnect resumes from a stale position.
    await _flushCursor(force: true);
    if (_stopped) break;

    // Only a connection that actually stayed up counts as a success. A
    // socket that fails immediately must keep growing the delay.
    if (uptime.elapsed >= healthyConnectionThreshold) {
      consecutiveFailures = 0;
    }
    consecutiveFailures++;

    await _sleep(
      _withJitter(
        reconnectBackoff(
          consecutiveFailures,
          initial: initialBackoff,
          max: maxBackoff,
        ),
      ),
    );
  }

  // `break` can leave a connection open when stop() raced the connect.
  await _closeQuietly(_connection);
  _connection = null;
  await _flushCursor(force: true);
}