declareLivelinessSubscriber method

Subscriber declareLivelinessSubscriber(Object keyExpr, { bool history = false, bool retainPayload = false, })

Declares a liveliness subscriber on the given keyExpr.

Returns a Subscriber whose Subscriber.stream delivers Samples with SampleKind.put when a liveliness token is declared and SampleKind.delete when a token is undeclared.

If history is true, the subscriber will also receive notifications for liveliness tokens that were declared before the subscription.

⚠️ This stream is UNBOUNDED, and pausing it does not stop the flow. Arrivals are pushed into a StreamController as they land, so a paused or slow consumer accumulates them without limit — pause() throttles delivery to your listener, never the producer, and nothing in this package wires the gate callbacks that would.

Measured on the shipped package, two OS processes over TCP, 64 MiB posted into a listener paused before any traffic: resident memory grew by 154 MiB on this surface, against 3 MiB for the bounded alternative on the same load.

For a bounded consumer use declarePullLivelinessSubscriber and its PullSubscriber.stream — the same mechanism, reached through the shared return type at no extra cost.

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

Implementation

Subscriber declareLivelinessSubscriber(
  Object keyExpr, {
  bool history = false,

  /// See [declareSubscriber] for what `retainPayload` costs and promises.
  bool retainPayload = false,
}) {
  return _withKeyExprArg(keyExpr, 'keyExpr', (loanedSession, loanedKe) {
    // ALLOCATE-LAST: the slot and the channel are claimed only once the key
    // expression has been accepted.
    final ptr = calloc.allocate<Void>(bindings.zd_subscriber_sizeof());
    final channel = Subscriber.createSampleChannel(
      retainPayload: retainPayload,
    );

    final rc = bindings.zd_liveliness_declare_subscriber(
      ptr.cast(),
      loanedSession.cast(),
      loanedKe.cast(),
      channel.receivePort.sendPort.nativePort,
      history ? 1 : 0,
      retainPayload ? 1 : 0,
    );

    if (rc != 0) {
      channel.abandon();
      calloc.free(ptr);
      throw ZenohException('Failed to declare liveliness subscriber', rc);
    }

    return Subscriber.fromParts(
      ptr,
      channel,
      keyExprString(keyExpr, 'keyExpr'),
    );
  });
}