pullLivelinessGet method

PullReplies pullLivelinessGet(Object keyExpr, { required ChannelKind kind, required int capacity, Duration? timeout, bool retainPayload = false, })

Queries liveliness tokens, with replies landing in a bounded channel instead of a stream.

The channel-mode sibling of livelinessGet, carrying its identical option surface. kind and capacity are required, deliberately — canon forces the caller to choose both.

The handle's release is PullReplies.dispose: local only, because the query still runs to completion natively.

A zero timeout is accepted here for the same measured reason it is on livelinessGet — see there.

⚠️ A ring channel recovers nothing once the query has completed, and a liveliness get completes almost immediately, so poll a ring one in flight or use ChannelKind.fifo.

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

Implementation

PullReplies pullLivelinessGet(
  Object keyExpr, {
  required ChannelKind kind,
  required int capacity,
  Duration? timeout,

  /// Whether each pulled OK reply carries a retained
  /// [Sample.payloadZBytes]. Off by default; the ERROR arm never does.
  bool retainPayload = false,
}) {
  // BEFORE ANY NATIVE CALL, for the same reason as everywhere else on this
  // axis: a negative would be reinterpreted as an enormous unsigned capacity.
  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 handlerHandle = calloc<Uint8>(
      bindings.zd_reply_handler_sizeof(kind.value),
    );
    final receivePort = ReceivePort();
    final teeOut = calloc<Pointer<Uint8>>();

    final rc = bindings.zd_liveliness_get_channel(
      handlerHandle,
      teeOut,
      receivePort.sendPort.nativePort,
      loanedSession.cast(),
      loanedKe.cast(),
      kind.value,
      capacity,
      (timeout ?? const Duration(seconds: 10)).inMilliseconds,
    );
    final teeValue = teeOut.value;
    calloc.free(teeOut);

    if (rc != 0) {
      receivePort.close();
      calloc.free(handlerHandle);
      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 reply channel state', rc);
      }
      throw ZenohException('Liveliness get failed', rc);
    }

    return PullReplies(
      handlerHandle,
      teeValue,
      receivePort,
      kind,
      retainPayload,
    );
  });
}