replyBytes method

void replyBytes(Object keyExpr, ZBytes payload, { Encoding? encoding, ZBytes? attachment, Timestamp? timestamp, bool? isExpress, })

Sends a reply to this query with a ZBytes payload.

The keyExpr should match the queryable's key expression. The payload is consumed by this call (ownership transferred to zenoh). Optionally specify an encoding for the payload and a binary attachment carried alongside the reply sample. The attachment, if provided, is also consumed by this call. An optional timestamp is borrowed (not consumed) and stamped onto the reply sample.

isExpress disables batching for this reply; omitting it — or passing null — means canon decides, which is false.

isExpress is the ONLY quality-of-service option on the reply path, and that is deliberate. canon marks a reply's congestion control and priority deprecated and ignored ("Reply congestion control is not supported anymore"), so this binding does not expose them: a reply inherits the QUERY's congestion control and priority, and nothing set here could change that. Set them on the get or Querier instead.

Throws StateError if the query has been disposed. Throws ZenohException if the reply fails.

Implementation

void replyBytes(
  Object keyExpr,
  ZBytes payload, {
  Encoding? encoding,
  ZBytes? attachment,
  Timestamp? timestamp,
  bool? isExpress,
}) {
  // PRE-move guards. All of these run BEFORE any z_bytes_move, so on these
  // paths the caller retains ownership of payload/attachment and we must
  // NOT mark them consumed:
  //   (1) a disposed query throws StateError here;
  //   (2) a wrong-typed key expression throws ArgumentError in the dispatch;
  //   (3) an invalid key expression string throws ZenohException there too.
  // The dispatch encloses the FFI call, so (2) and (3) still precede the
  // move -- which is what the unconditional markConsumed below depends on.
  _ensureNotDisposed();

  withLoanedKeyExpr(keyExpr, 'keyExpr', (loanedKe) {
    // Two INDEPENDENT length-carried channels (R-2), from the RAW pair
    // (R-3a). Built BEFORE the payload move, like everything else here.
    final (mime, schema) = encoding != null
        ? encodingWireChannels(encoding)
        : (null, null);
    final encodingBuf = allocLengthCarriedUtf8(mime);
    final schemaBuf = allocLengthCarriedUtf8(schema);

    // Timestamp is BORROWED (not moved) -- allocate an 8-byte-aligned copy
    // after the pre-move guards so a pre-move throw never leaks it. Freed in
    // the finally alongside encodingNative.
    final tsPtr = timestamp != null ? _timestampToNative(timestamp) : nullptr;

    try {
      final rc = bindings.zd_query_reply(
        Pointer.fromAddress(_handle).cast(),
        loanedKe.cast(),
        payload.nativePtr.cast(),
        encodingBuf.ptr,
        encodingBuf.len,
        schemaBuf.ptr,
        schemaBuf.len,
        attachment != null ? attachment.nativePtr.cast() : nullptr,
        tsPtr.cast(),
        isExpress == null ? -1 : (isExpress ? 1 : 0),
      );

      // Mark payload + attachment ZBytes consumed UNCONDITIONALLY: once we
      // reach this FFI call the pre-move guards above have passed, so
      // zd_query_reply has moved both into zenoh-c regardless of the return
      // code (its encoding-error path drops the already-moved bytes).
      // Marking before the rc-throw prevents a later use-after-move.
      payload.markConsumed();
      attachment?.markConsumed();

      if (rc != 0) {
        throw ZenohException('Failed to reply to query', rc);
      }
    } finally {
      // allocLengthCarriedUtf8 uses calloc; ptr is nullptr exactly when the
      // value was null, so the guard is on the pointer, not on `encoding`.
      if (encodingBuf.ptr != nullptr) calloc.free(encodingBuf.ptr);
      if (schemaBuf.ptr != nullptr) calloc.free(schemaBuf.ptr);
      if (tsPtr != nullptr) {
        calloc.free(tsPtr);
      }
    }
  });
}