pullGet method

PullReplies pullGet(Object selector, { required ChannelKind kind, required int capacity, String? parameters, ZBytes? payload, Encoding? encoding, ZBytes? attachment, QueryTarget target = QueryTarget.bestMatching, ConsolidationMode consolidation = ConsolidationMode.auto, Duration? timeout, CongestionControl? congestionControl, Priority? priority, bool? isExpress, Locality? allowedDestination, ReplyKeyExpr? acceptReplies, bool retainPayload = false, })

Sends a query whose replies land in a bounded channel instead of a stream, and returns a PullReplies handle to take them out of.

The channel-mode sibling of get, carrying its identical option surface. Where get pushes every reply as it arrives and buffers without bound, this holds at most capacity replies and lets the consumer set the pace — which is canon's own documented default get flow (z_get + z_fifo_channel_reply_new).

kind and capacity are required, deliberately. Canon forces the caller to choose both — its channel constructors take a raw size_t and have no options-default, and the C++ binding's channel selector has no default either — so this binding substitutes no value canon does not have.

The handle's release is PullReplies.dispose, not close: the query still runs to completion natively and no peer observes the drop.

⚠️ Never poll a full fifo channel from the session that answers it

Measured at zenoh-c 1.8.0. On a same-session route the reply delivery runs synchronously inside your own z_get call, so a fifo channel that fills up freezes this call inside the FFI boundary — and timeout cannot rescue a thread stuck in that push. Use a second session for the replier, or ChannelKind.ring. The same hazard runs the other way on a channel-backed queryable.

⚠️ A ring channel must be polled while the query is in flight

A ring discards its whole buffer when the channel disconnects, and a get completes immediately after its replies — so a ring reply channel polled only after completion recovers nothing. That is canon's behaviour, measured, and it is rendered here rather than papered over.

⚠️ consolidation decides what this channel can even see

Measured. canon's default, ConsolidationMode.auto, resolves to a consolidating mode that both dedupes replies by key expression and withholds them until the query completes. Three replies on one key then reach the channel as one, and nothing at all is visible in flight — which for a ring channel means nothing at all, full stop. Pass ConsolidationMode.none when you want every reply, or when you intend to consume in flight.

Capacity 0

Measured on this column: a capacity-0 fifo is a rendezvous — full when empty — and PullReplies.tryRecv works there, because a synchronous poll is itself the concurrent consumer the rendezvous needs. PullReplies.recv does not: see its own documentation. A capacity-0 ring recovers nothing after completion, like any other ring.

Throws ArgumentError if capacity is negative, or if timeout marshals to 0 ms (see get). Throws StateError if the session has been closed.

⛔ Throws ArgumentError if congestionControl is CongestionControl.blockFirst and the loaded native was built without Z_FEATURE_UNSTABLE_API. canon declares Z_CONGESTION_CONTROL_BLOCK_FIRST only under that flag, so on such a build there is no value to send. Pass CongestionControl.block or CongestionControl.drop, or select the unstable native through your app's user_defines.

Implementation

PullReplies pullGet(
  Object selector, {
  required ChannelKind kind,
  required int capacity,
  String? parameters,
  ZBytes? payload,
  Encoding? encoding,
  ZBytes? attachment,
  QueryTarget target = QueryTarget.bestMatching,
  ConsolidationMode consolidation = ConsolidationMode.auto,
  Duration? timeout,
  CongestionControl? congestionControl,
  Priority? priority,
  bool? isExpress,
  Locality? allowedDestination,
  ReplyKeyExpr? acceptReplies,

  /// Whether each pulled OK reply carries a retained
  /// [Sample.payloadZBytes]. Off by default; the ERROR arm never does.
  bool retainPayload = false,
}) {
  // FIRST STATEMENT: the stated order on this entry point is congestion,
  // then capacity, then timeout. A negative capacity and a zero timeout are
  // both expressible correctly by choosing another number; blockFirst on a
  // stable native is not, so reporting the fixable faults first would send
  // the caller round a loop.
  requireCongestionControlSupported(congestionControl);
  // BEFORE ANY NATIVE CALL, and for the same reason as on the pull
  // subscriber: 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');
  }
  _rejectSentinelTimeout(timeout);

  return _withKeyExprArg(selector, 'selector', (loanedSession, loanedKe) {
    // ALLOCATE-LAST: claimed only after the selector has been accepted.
    final handlerHandle = calloc<Uint8>(
      bindings.zd_reply_handler_sizeof(kind.value),
    );
    // The readiness channel behind `PullReplies.recv()`. The shim posts an
    // int64 ping when an armed waiter should look again, and a null sentinel
    // from the closure's drop when the query completes.
    final receivePort = ReceivePort();
    // Our out-cell for a SHIM-owned block: the shim mallocs the tee context
    // and `zd_pull_tee_drop` releases the handle's reference to it. This cell
    // is ours, and the outer `finally` encloses every statement that can
    // throw.
    final teeOut = calloc<Pointer<Uint8>>();
    Pointer<Uint8> teeValue = nullptr;
    var started = false;
    Pointer<Char> parametersNative = nullptr;
    var parametersLen = 0;
    // The encoding joins parameters on the length-carried side of this same
    // signature -- the asymmetry inside it (parameters length-carried, the
    // encoding a bare C string four lines below) is what this seed removes.
    // Two INDEPENDENT channels, from the RAW pair (R-3a).
    Pointer<Char> encodingNative = nullptr;
    var encodingLen = 0;
    Pointer<Char> schemaNative = nullptr;
    var schemaLen = 0;

    try {
      final marshalled = allocLengthCarriedUtf8(parameters);
      parametersNative = marshalled.ptr;
      parametersLen = marshalled.len;
      final (mime, schema) = encoding != null
          ? encodingWireChannels(encoding)
          : (null, null);
      final encMarshalled = allocLengthCarriedUtf8(mime);
      encodingNative = encMarshalled.ptr;
      encodingLen = encMarshalled.len;
      final schemaMarshalled = allocLengthCarriedUtf8(schema);
      schemaNative = schemaMarshalled.ptr;
      schemaLen = schemaMarshalled.len;

      final rc = bindings.zd_get_channel(
        handlerHandle,
        teeOut,
        receivePort.sendPort.nativePort,
        loanedSession.cast(),
        loanedKe.cast(),
        kind.value,
        capacity,
        target.index,
        consolidation.value,
        payload != null ? payload.nativePtr.cast() : nullptr,
        encodingNative,
        encodingLen,
        schemaNative,
        schemaLen,
        timeout != null ? timeout.inMilliseconds : 0,
        parametersNative,
        parametersLen,
        attachment != null ? attachment.nativePtr.cast() : nullptr,
        congestionControl?.value ?? -1,
        priority?.value ?? -1,
        isExpress == null ? -1 : (isExpress ? 1 : 0),
        allowedDestination?.value ?? -1,
        acceptReplies?.value ?? -1,
      );

      // Marked UNCONDITIONALLY, exactly as on [get]: the shim has either
      // moved them into zenoh-c or dropped them itself on every return code
      // reachable from here. The one code that returns pre-move is the
      // capacity refusal, which the guard above makes unreachable.
      if (payload != null) {
        payload.markConsumed();
      }
      if (attachment != null) {
        attachment.markConsumed();
      }

      if (rc != 0) {
        // The declare channel's return space is SPLIT, and it is 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 reply channel state', rc);
        }
        throw ZenohException('Get query failed', rc);
      }
      teeValue = teeOut.value;
      started = true;
    } finally {
      if (parametersNative != nullptr) calloc.free(parametersNative);
      if (encodingNative != nullptr) calloc.free(encodingNative);
      if (schemaNative != nullptr) calloc.free(schemaNative);
      if (!started) {
        receivePort.close();
        calloc.free(handlerHandle);
      }
      calloc.free(teeOut);
    }

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