livelinessGet method

Stream<Reply> livelinessGet(Object keyExpr, { Duration? timeout, bool retainPayload = false, })

Queries liveliness tokens matching the given keyExpr.

Returns a Stream of Reply objects for each alive token. The stream completes when all replies have been received or the timeout expires. Defaults to 10 seconds if timeout is not specified — sent explicitly, not deferred to zenoh.

Unlike get and declareQuerier, a zero timeout is accepted here. The asymmetry is deliberate and measured: a liveliness query completes as soon as the reachable peers have answered (0–2 ms in every configuration probed, with or without an alive token), so the timeout never bites and a zero value changes nothing observable — an alive token's reply still arrives. There is therefore no silent substitution to refuse, and refusing anyway would invent a restriction with nothing behind it.

⚠️ The returned stream is UNBOUNDED, and pausing it does not stop the flow. Replies are pushed in as they arrive; pause() throttles delivery to your listener, not the responders.

pullLivelinessGet is the paced alternative — a bounded PullReplies handle you poll. It offers no Stream view; see get.

Throws ZenohException if the key expression is invalid or the query fails. Throws StateError if the session has been closed.

Implementation

Stream<Reply> livelinessGet(
  Object keyExpr, {
  Duration? timeout,

  /// Whether each OK reply's sample carries a retained
  /// [Sample.payloadZBytes]. Off by default.
  bool retainPayload = false,
}) {
  return _withKeyExprArg(keyExpr, 'keyExpr', (loanedSession, loanedKe) {
    // ALLOCATE-LAST: no open ReceivePort survives a rejected key expression.
    final (receivePort, controller, retention) = _createReplyChannel(
      retainPayload: retainPayload,
    );
    final timeoutMs = (timeout ?? const Duration(seconds: 10)).inMilliseconds;

    final rc = bindings.zd_liveliness_get(
      loanedSession.cast(),
      loanedKe.cast(),
      receivePort.sendPort.nativePort,
      timeoutMs,
      retainPayload ? 1 : 0,
    );

    if (rc != 0) {
      receivePort.close();
      unawaited(controller.close());
      throw ZenohException('Liveliness get failed', rc);
    }

    return retention.gate(controller.stream);
  });
}