putBytes method

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

Publishes a ZBytes payload on the given keyExpr.

The payload is consumed by this call and must not be reused. Optionally set the encoding (MIME type) of the message. An optional attachment can be included; it is also consumed by this call. 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.

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 payload or 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.

⚠️ That refusal is raised BEFORE the session-closed check, so a closed session carrying CongestionControl.blockFirst reports the argument fault rather than StateError. This is the only entry point where both can apply, and the build-configuration fault is the one nothing else in the program will ever report; a closed session is discoverable from any other call on it.

Implementation

void putBytes(
  Object keyExpr,
  ZBytes payload, {
  Encoding? encoding,
  ZBytes? attachment,
  Timestamp? timestamp,
  CongestionControl? congestionControl,
  Priority? priority,
  bool? isExpress,
  Locality? allowedDestination,
}) {
  // FIRST STATEMENT, and that means AHEAD OF `_ensureOpen()` -- this is the
  // only one of the seven entry points carrying an explicit closed-session
  // check, so this is where the ordering is a choice rather than an
  // artefact of helper placement. A closed session is discoverable from any
  // other call on that session; a build-configuration fault is discoverable
  // from nothing else in the program, so where a call carries both, the
  // undiscoverable one is the one worth reporting. The alternative -- guard
  // after whatever state check happens to exist -- makes `put` and
  // `putBytes` answer differently on identical inputs, with the
  // discriminator invisible from the API.
  requireCongestionControlSupported(congestionControl);
  _ensureOpen();
  // Validate payload state before allocating KeyExpr
  final payloadPtr = payload.nativePtr;
  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: a wrong-typed or invalid key expression throws inside
  // _withKeyExprArg before the closure runs, so the release could not sit
  // inside it.
  try {
    _withKeyExprArg<void>(keyExpr, 'keyExpr', (loanedSession, loanedKe) {
      final rc = bindings.zd_put(
        loanedSession.cast(),
        loanedKe.cast(),
        payloadPtr.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);
  }
}