declarePullLivelinessSubscriber method

PullSubscriber declarePullLivelinessSubscriber(Object keyExpr, { required ChannelKind kind, required int capacity, bool history = false, bool retainPayload = false, })

Declares a liveliness subscriber whose transitions land in a bounded channel, and returns the shipped PullSubscriber to take them out of.

The channel-mode sibling of declareLivelinessSubscriber. It delivers the alive/gone transitions as samples — SampleKind.put when a token appears, SampleKind.delete when it goes — which is a different capability from livelinessGet's snapshot of who is alive now.

This is the thinnest of the channel-mode entries: canon's liveliness declare consumes the sample closure, so the handle, the machinery and the contracts are the pull subscriber's, unchanged.

kind and capacity are required, deliberately — canon forces the caller to choose both. (declarePullSubscriber defaults them only to keep every pre-existing caller's behaviour byte-for-byte; new surface has no such debt.)

history replays tokens that were already alive when this subscriber was declared. It is canon's only option on this entry.

⚠️ Think twice before choosing ChannelKind.ring here

A ring drops the oldest entry when it fills, and on a presence feed the dropped entry may be a token-gone transition — which does not merely lose data, it inverts the consumer's world-state: you go on believing something is alive that has gone. That is a sharper edge than losing a sample on an ordinary data feed, and it is the reason to prefer ChannelKind.fifo unless you have a specific reason not to.

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

Implementation

PullSubscriber declarePullLivelinessSubscriber(
  Object keyExpr, {
  required ChannelKind kind,
  required int capacity,
  bool history = false,

  /// Whether each pulled sample carries a retained
  /// [Sample.payloadZBytes]. Off by default.
  bool retainPayload = false,
}) {
  // BEFORE ANY NATIVE CALL: a negative would otherwise be reinterpreted as an
  // enormous unsigned capacity -- a silent transform, not a refusal.
  if (capacity < 0) {
    throw ArgumentError.value(capacity, 'capacity', 'must be non-negative');
  }
  return _withKeyExprArg(keyExpr, 'keyExpr', (loanedSession, loanedKe) {
    // ALLOCATE-LAST: claimed only after the key expression has been accepted.
    final subscriberHandle = calloc<Uint8>(bindings.zd_subscriber_sizeof());
    final handlerHandle = calloc<Uint8>(
      bindings.zd_pull_handler_sizeof(kind.value),
    );
    final receivePort = ReceivePort();
    final teeOut = calloc<Pointer<Uint8>>();

    try {
      final rc = bindings.zd_declare_pull_liveliness_subscriber(
        subscriberHandle,
        handlerHandle,
        teeOut,
        loanedSession.cast(),
        loanedKe.cast(),
        kind.value,
        capacity,
        history ? 1 : 0,
        receivePort.sendPort.nativePort,
      );

      if (rc != 0) {
        receivePort.close();
        calloc
          ..free(subscriberHandle)
          ..free(handlerHandle);
        // The declare channel's SPLIT return space, mapped here once.
        if (rc == 10) {
          throw ArgumentError.value(
            capacity,
            'capacity',
            "must be non-negative and within this platform's size_t range",
          );
        }
        if (rc == 11) {
          throw ZenohException(
            'Failed to allocate pull liveliness subscriber state',
            rc,
          );
        }
        throw ZenohException(
          'Failed to declare pull liveliness subscriber',
          rc,
        );
      }

      return PullSubscriber(
        subscriberHandle,
        handlerHandle,
        teeOut.value,
        receivePort,
        keyExprString(keyExpr, 'keyExpr'),
        kind,
        retainPayload,
      );
    } finally {
      calloc.free(teeOut);
    }
  });
}