get method

Stream<Reply> get(Object selector, { 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 on the given selector and returns a stream of replies.

The returned stream completes when all replies have been received or the timeout expires. When timeout is null, the query uses the default query timeout from the session's zenoh configuration (queries_default_timeout, itself 10 seconds by default) — passing wire 0, matching zenoh-c's GetOptions.timeout_ms == 0 semantics.

⚠️ A timeout that marshals to 0 ms is refused with ArgumentError. Zenoh reads wire 0 as "use the configured default", so Duration.zero — which every reasonable reading takes to mean "expire immediately" — would silently become ~10 seconds. The check is on the wire value, so a positive sub-millisecond duration such as Duration(microseconds: 500) is refused too: its inMilliseconds truncates onto the same sentinel. Pass at least 1 ms, or omit timeout to let zenoh decide explicitly. declareQuerier carries the identical rule; livelinessGet deliberately does not — see there.

Optional parameters are the selector's portion after ?. They are carried length-first, so an interior NUL is a value rather than a terminator; canon requires the string to be valid UTF-8. Omitting them and passing '' are indistinguishable at the queryable — canon collapses the two before any wire encoding. Optional payload, encoding, and attachment attach data to the query. target controls which queryables are targeted (default: bestMatching). consolidation controls reply consolidation (default: auto).

Send options

congestionControl, priority, isExpress, allowedDestination and acceptReplies are each optional, and omitting one — or passing null — means the same thing: canon decides. This binding substitutes no value of its own. On this path canon's defaults are CongestionControl.block (a request operation; put, deleteResource and declarePublisher default to CongestionControl.drop instead), Priority.data, isExpress: false, Locality.any, and ReplyKeyExpr.matchingQuery.

⚠️ Read CongestionControl before selecting CongestionControl.block: it can park the calling thread for seconds and then close the transport.

⚠️ The returned stream is UNBOUNDED, and pausing it does not stop the flow. Replies are pushed in as they arrive; pause() throttles delivery to your listener, not the responders.

pullGet is the paced alternative: it returns a PullReplies handle over a bounded channel that you poll. ⚠️ Note it offers no Stream view — the bounded-Stream mechanism is deliberately carved to the sample and query columns, so on the reply column the bounded form is the polling handle and nothing else.

⚠️ A value this binding cannot convert arrives as a STREAM ERROR, never as empty or absent data. Zenoh can hand over a payload or attachment the conversion step refuses; delivering that as a zero-length value would be indistinguishable from a legitimately empty one, and empty is a real value on this path. The error carries canon's own code.

⛔ The stream keeps running. A conversion failure is a failed call, not a dead channel — the same rule the pull family already ships — so later values still arrive and terminating on the first bad one would lose them. ⚠️ But an unhandled stream error is still an unhandled error: pass onError (or handleError), because an unhandled one can take the program down.

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

Stream<Reply> get(
  Object selector, {
  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 OK reply's sample carries a retained
  /// [Sample.payloadZBytes]. Off by default; see [declareSubscriber] for
  /// what retention costs and promises.
  /// The ERROR arm is unaffected: `ReplyError` gets no retained handle.
  bool retainPayload = false,
}) {
  // FIRST STATEMENT, ahead of the timeout guard -- the stated order,
  // congestion before timeout. It matters here for a second reason too:
  // this method returns its Stream synchronously, so a refusal raised after
  // the ReceivePort below would leave a port NO native sentinel can ever
  // close, pinning the isolate alive.
  requireCongestionControlSupported(congestionControl);
  // BEFORE any native call, including the selector validation below: a
  // refused timeout must leave nothing declared and no ReceivePort open.
  _rejectSentinelTimeout(timeout);
  // The selector is validated exactly ONCE, in the union dispatch, and its
  // temp KeyExpr is disposed before this returns. It used to be validated
  // twice -- here and again inside zd_get -- because the Dart caller cannot
  // tell a pre-move rc from a post-move one, and the markConsumed discipline
  // depends on that distinction: a rejected selector must throw WITHOUT
  // marking payload/attachment consumed, mirroring put/putBytes. Validating
  // in the dispatch, before anything is moved, preserves that exactly.
  return _withKeyExprArg(selector, 'selector', (loanedSession, loanedKe) {
    final (receivePort, controller, retention) = _createReplyChannel(
      retainPayload: retainPayload,
    );
    // Wire 0 means "use the config default query timeout"
    // (zenoh_commons.h:1039; C++ peer GetOptions.timeout_ms defaults to 0,
    // session.hxx:299). Mirror querier.dart:65 rather than substituting a
    // hardcoded default.
    final timeoutMs = timeout != null ? timeout.inMilliseconds : 0;

    // `started` gates the channel teardown in the finally. Every throw
    // between here and a successful zd_get leaves a ReceivePort that NO
    // native sentinel can ever close -- zd_get either never ran or failed --
    // and an open port pins the isolate alive. The reachable trigger is a
    // disposed or already-consumed payload/attachment: their nativePtr
    // getters throw StateError partway through building the call.
    var started = false;
    // LENGTH-CARRIED, not NUL-terminated: the parameters segment's domain
    // includes an interior NUL, so a C string would truncate it at the seam.
    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(
        loanedSession.cast(),
        loanedKe.cast(),
        receivePort.sendPort.nativePort,
        target.index,
        consolidation.value,
        payload != null ? payload.nativePtr.cast() : nullptr,
        encodingNative,
        encodingLen,
        schemaNative,
        schemaLen,
        timeoutMs,
        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,
        retainPayload ? 1 : 0,
      );

      // Mark payload + attachment ZBytes as consumed UNCONDITIONALLY:
      // zd_get moves them into zenoh-c regardless of the return code (and
      // its encoding-error early-return drops the already-moved bytes), so
      // the caller must not touch them after this call -- even on error.
      // Marking before the rc-throw prevents a later use-after-move.
      if (payload != null) {
        payload.markConsumed();
      }
      if (attachment != null) {
        attachment.markConsumed();
      }

      if (rc != 0) {
        throw ZenohException('Get query failed', rc);
      }
      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();
        unawaited(controller.close());
      }
    }

    return retention.gate(controller.stream);
  });
}