get method

Stream<Reply> get(String selector, { String? parameters, ZBytes? payload, Encoding? encoding, ZBytes? attachment, QueryTarget target = QueryTarget.bestMatching, ConsolidationMode consolidation = ConsolidationMode.auto, Duration? timeout, })

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. The timeout defaults to 10 seconds.

Optional parameters are appended to the query selector. Optional payload, encoding, and attachment attach data to the query. target controls which queryables are targeted (default: bestMatching). consolidation controls reply consolidation (default: auto).

Throws StateError if the session has been closed.

Implementation

Stream<Reply> get(
  String selector, {
  String? parameters,
  ZBytes? payload,
  Encoding? encoding,
  ZBytes? attachment,
  QueryTarget target = QueryTarget.bestMatching,
  ConsolidationMode consolidation = ConsolidationMode.auto,
  Duration? timeout,
}) {
  _ensureOpen();

  // Validate the selector keyexpr in Dart BEFORE any z_bytes_move. zd_get's
  // own z_view_keyexpr_from_str is a pre-move early-return (-1), but the
  // Dart caller cannot distinguish that rc from a post-move failure -- so we
  // validate here to keep the markConsumed discipline correct: a genuine
  // pre-move early-return throws here and does NOT mark payload/attachment
  // consumed (the caller retains ownership), mirroring put/putBytes.
  KeyExpr(selector).dispose();

  final (receivePort, controller) = _createReplyChannel();
  final timeoutMs = (timeout ?? const Duration(seconds: 10)).inMilliseconds;

  final loanedSession =
      bindings.zd_session_loan(_ptr.cast()) as Pointer<Void>;

  final selectorNative = selector.toNativeUtf8();
  Pointer<Utf8> parametersNative = nullptr;
  Pointer<Utf8> encodingNative = nullptr;

  if (parameters != null) {
    parametersNative = parameters.toNativeUtf8();
  }
  if (encoding != null) {
    encodingNative = encoding.mimeType.toNativeUtf8();
  }

  try {
    final rc = bindings.zd_get(
      loanedSession.cast(),
      selectorNative.cast(),
      receivePort.sendPort.nativePort,
      target.index,
      consolidation.value,
      payload != null ? payload.nativePtr.cast() : nullptr,
      encoding != null ? encodingNative.cast() : nullptr,
      timeoutMs,
      parameters != null ? parametersNative.cast() : nullptr,
      attachment != null ? attachment.nativePtr.cast() : nullptr,
    );

    // 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) {
      receivePort.close();
      controller.close();
      throw ZenohException('Get query failed', rc);
    }
  } finally {
    calloc.free(selectorNative);
    if (parameters != null) calloc.free(parametersNative);
    if (encoding != null) calloc.free(encodingNative);
  }

  return controller.stream;
}