declarePullQueryable method

PullQueryable declarePullQueryable(Object keyExpr, { required ChannelKind kind, required int capacity, bool complete = false, Locality? allowedOrigin, })

Declares a queryable whose queries land in a bounded channel instead of a stream, and returns a PullQueryable to take them out of.

The channel-mode sibling of declareQueryable, carrying its identical option surface. Where declareQueryable pushes every query into a stream as it arrives and buffers without bound, this holds at most capacity queries and lets the consumer set the pace.

kind and capacity are required, deliberately: canon forces the caller to choose both, so this binding substitutes no value canon does not have.

The release is PullQueryable.close, not dispose — it undeclares, and getters stop reaching this queryable.

⚠️ Never let a same-session getter meet a full fifo channel

Measured at zenoh-c 1.8.0. On a same-session route the query delivery runs synchronously inside the getter's own z_get call, so a fifo channel that has filled up freezes that call inside the FFI boundary with no timeout escape — and that includes the existing Stream-path get on the same session, not just this handle's own consumers. Use a second session for the getter, or ChannelKind.ring.

Across two sessions the getter is never the blocked party: its call returns, and the queries queue until this channel is drained. Measured here: an ordinary Stream-path declareQueryable co-hosted on the same session keeps answering while this channel sits full, at every capacity and depth probed — so a full channel is not, on this binding's measured behaviour, a session-wide stall. Canon-direct measurement reports the opposite — inbound query delivery stalling session-wide while a fifo query channel sits full — and that did not reproduce through this stack at any capacity or depth probed, so treat the favourable behaviour above as what this binding measured rather than as a guarantee. Size the capacity for the slowest consumer you will actually run all the same: what queues has to be held somewhere.

⚠️ A ring channel drops the oldest query

Remotely that getter simply gets nothing — its stream completes with no reply at all. Lossy is the trade a ring makes to never stall the producer.

⚠️ This used to say "that is the dropped getter's timeout: it waited and got nothing", and the mechanism half is MEASURED FALSE. The dropped getter does not wait and does not time out: measured at ~1 ms, finalized with nothing, while getters still resident in the channel waited for the close. The conclusion ("got nothing") was right; the route to it was not. Pinned in test/fifo_close_window_test.dart, "the same window on a ring query channel is the control". Buffered queries are also discarded when the channel disconnects — a fifo hands its buffer over first, a ring does not.

Capacity 0

Measured on this column: at capacity 0 both kinds hand a query over under polling and the getter gets its reply. That is not the reply column's picture — see pullGet — because across two sessions the getter is never the blocked party here.

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

Implementation

PullQueryable declarePullQueryable(
  Object keyExpr, {
  required ChannelKind kind,
  required int capacity,
  bool complete = false,
  Locality? allowedOrigin,
}) {
  // BEFORE ANY NATIVE CALL: a negative is outside canon's `size_t` domain
  // entirely, and the carriage would otherwise reinterpret it as an enormous
  // unsigned capacity -- a silent transform, not a refusal. No upper bound is
  // invented.
  if (capacity < 0) {
    throw ArgumentError.value(capacity, 'capacity', 'must be non-negative');
  }
  return _withKeyExprArg(keyExpr, 'keyExpr', (loanedSession, loanedKe) {
    // ALLOCATE-LAST: both slots are claimed only after the key expression has
    // been accepted, so a rejected one cannot strand them.
    final queryableHandle = calloc<Uint8>(bindings.zd_queryable_sizeof());
    // The two handler types are distinct, so the slot is sized for the kind
    // we are about to declare -- and released through the same kind.
    final handlerHandle = calloc<Uint8>(
      bindings.zd_query_handler_sizeof(kind.value),
    );
    // The readiness channel behind `PullQueryable.recv()`: an int64 ping when
    // an armed waiter should look again, and a null sentinel from the
    // closure's drop when the producer is gone.
    final receivePort = ReceivePort();
    // Our out-cell for a SHIM-owned block. The cell is ours; the block is the
    // shim's, released through `zd_pull_tee_drop`.
    final teeOut = calloc<Pointer<Uint8>>();

    final rc = bindings.zd_declare_queryable_channel(
      queryableHandle,
      handlerHandle,
      teeOut,
      receivePort.sendPort.nativePort,
      loanedSession.cast(),
      loanedKe.cast(),
      kind.value,
      capacity,
      complete ? 1 : 0,
      allowedOrigin?.value ?? -1,
    );
    final teeValue = teeOut.value;
    calloc.free(teeOut);

    if (rc != 0) {
      receivePort.close();
      calloc
        ..free(queryableHandle)
        ..free(handlerHandle);
      // The declare channel's return space is SPLIT, mapped here, once, at
      // the single call site -- the same split the pull subscriber ships.
      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 query channel state', rc);
      }
      throw ZenohException('Failed to declare pull queryable', rc);
    }

    return PullQueryable(
      queryableHandle,
      handlerHandle,
      teeValue,
      receivePort,
      keyExprString(keyExpr, 'keyExpr'),
      kind,
    );
  });
}