put method

void put(Object keyExpr, String value, { Encoding? encoding, ZBytes? attachment, Timestamp? timestamp, CongestionControl? congestionControl, Priority? priority, bool? isExpress, Locality? allowedDestination, })

Publishes a string value on the given keyExpr.

Optionally set the encoding (MIME type) of the message. An optional attachment can be included; it is consumed by this call and must not be reused. An optional timestamp can be attached to the message; borrowed, not consumed.

Send options

congestionControl, priority, isExpress and allowedDestination 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.drop (a push operation; get and declareQuerier default to CongestionControl.block instead), Priority.data, isExpress: false, and Locality.any.

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

keyExpr is a String or a KeyExpr — including one obtained from declareKeyExpr, whose native handle reaches zenoh directly instead of being re-parsed from a string. A KeyExpr argument is loaned, not consumed.

Throws ArgumentError if keyExpr is neither a String nor a KeyExpr. Throws ZenohException if the key expression is invalid, the encoding is malformed, or the put fails. Throws StateError if the session has been closed, or the attachment has been disposed or already consumed.

⛔ 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

void put(
  Object keyExpr,
  String value, {
  Encoding? encoding,
  ZBytes? attachment,
  Timestamp? timestamp,
  CongestionControl? congestionControl,
  Priority? priority,
  bool? isExpress,
  Locality? allowedDestination,
}) {
  // FIRST STATEMENT, ahead of every allocation and every native call: a
  // refused congestion control must leave nothing allocated, nothing
  // consumed and nothing declared.
  requireCongestionControlSupported(congestionControl);
  final attachmentPtr = attachment != null ? attachment.nativePtr : nullptr;
  // Two INDEPENDENT length-carried channels (R-2), fed from the RAW pair
  // (R-3a) -- never the derived `schema` getter, which would split a
  // composed mimeType and change what goes on the wire.
  final (mime, schema) = encoding != null
      ? encodingWireChannels(encoding)
      : (null, null);
  final encodingBuf = allocLengthCarriedUtf8(mime);
  final schemaBuf = allocLengthCarriedUtf8(schema);
  final tsPtr = timestamp != null ? _timestampToNative(timestamp) : nullptr;
  // OUTER-FINALLY, the pattern deleteResource already uses. The release used
  // to live INSIDE the _withKeyExpr closure, which is never entered when the
  // session is closed (_withKeyExprArg's own _ensureOpen), the argument is
  // wrong-typed, or the key expression is invalid -- so both buffers leaked
  // on those paths, and on a ZBytes.fromString throw as well.
  try {
    _withKeyExprArg<void>(keyExpr, 'keyExpr', (loanedSession, loanedKe) {
      final payload = ZBytes.fromString(value);
      final rc = bindings.zd_put(
        loanedSession.cast(),
        loanedKe.cast(),
        payload.nativePtr.cast(),
        encodingBuf.ptr,
        encodingBuf.len,
        schemaBuf.ptr,
        schemaBuf.len,
        attachmentPtr.cast(),
        tsPtr.cast(),
        congestionControl?.value ?? -1,
        priority?.value ?? -1,
        isExpress == null ? -1 : (isExpress ? 1 : 0),
        allowedDestination?.value ?? -1,
      );
      // markConsumed is unconditional: z_bytes_move gravestones the owned
      // bytes regardless of the return code. The timestamp is borrowed (not
      // moved) -- it is never marked consumed.
      payload.markConsumed();
      if (attachment != null) attachment.markConsumed();
      if (rc != 0) {
        throw ZenohException('Put failed', rc);
      }
    });
  } finally {
    // allocLengthCarriedUtf8 allocates with calloc, so it is released with
    // calloc.free -- not the malloc.free the retired toNativeUtf8 site used.
    if (encodingBuf.ptr != nullptr) calloc.free(encodingBuf.ptr);
    if (schemaBuf.ptr != nullptr) calloc.free(schemaBuf.ptr);
    if (tsPtr != nullptr) calloc.free(tsPtr);
  }
}