pullGet method

PullReplies pullGet({ required ChannelKind kind, required int capacity, String? parameters, ZBytes? payload, Encoding? encoding, ZBytes? attachment, bool retainPayload = false, })

Sends a query via this querier, with replies landing in a bounded channel instead of a stream.

The channel-mode sibling of get, carrying its identical option surface. kind and capacity are required, deliberately: canon forces the caller to choose both, so this binding substitutes no value canon does not have.

The handle's release is PullReplies.dispose — local only, because the query still runs to completion natively and no peer observes the drop.

The declaration-time options this querier was created with (target, consolidation, timeout) apply here exactly as they do to get; canon's per-get options struct carries no timeout field, so a sentinel timeout is refused at Session.declareQuerier rather than here.

⚠️ The same-session freeze and the ring's discard-at-disconnect apply exactly as on Session.pullGet — see there.

Throws ArgumentError if capacity is negative. Throws StateError if the querier has been closed.

Implementation

PullReplies pullGet({
  required ChannelKind kind,
  required int capacity,
  String? parameters,
  ZBytes? payload,
  Encoding? encoding,
  ZBytes? attachment,

  /// Whether each pulled OK reply carries a retained
  /// [Sample.payloadZBytes]. Off by default; the ERROR arm never does.
  bool retainPayload = false,
}) {
  if (_closed) throw StateError('Querier is closed');
  // BEFORE ANY NATIVE CALL: a negative is outside canon's `size_t` domain,
  // and the carriage would otherwise reinterpret it as an enormous unsigned
  // capacity -- a silent transform, not a refusal.
  if (capacity < 0) {
    throw ArgumentError.value(capacity, 'capacity', 'must be non-negative');
  }

  final handlerHandle = calloc<Uint8>(
    bindings.zd_reply_handler_sizeof(kind.value),
  );
  final receivePort = ReceivePort();
  final teeOut = calloc<Pointer<Uint8>>();
  Pointer<Uint8> teeValue = nullptr;
  var started = false;
  // LENGTH-CARRIED, as on the stream sibling.
  Pointer<Char> parametersNative = nullptr;
  var parametersLen = 0;
  // The encoding joins parameters on the length-carried side: 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_querier_get_channel(
      handlerHandle,
      teeOut,
      receivePort.sendPort.nativePort,
      _ptr.cast(),
      kind.value,
      capacity,
      parametersNative,
      parametersLen,
      payload != null ? payload.nativePtr.cast() : nullptr,
      encodingNative,
      encodingLen,
      schemaNative,
      schemaLen,
      attachment != null ? attachment.nativePtr.cast() : nullptr,
    );

    // The two PRE-MOVE codes are the exception to the unconditional mark, and
    // that is why they are kept distinguishable at the seam: on 10 and 11 the
    // shim returned before touching the payload or attachment, so marking
    // them would gravestone wrappers whose native handles are still the
    // caller's. On every other code the shim has either moved them into
    // zenoh-c or dropped them itself.
    if (rc != 10 && rc != 11) {
      if (payload != null) {
        payload.markConsumed();
      }
      if (attachment != null) {
        attachment.markConsumed();
      }
    }

    if (rc != 0) {
      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('Querier get 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,
  );
}