putBytes method

void putBytes(ZBytes payload, { Encoding? encoding, ZBytes? attachment, Timestamp? timestamp, })

Publishes ZBytes payload through this publisher.

The payload is consumed by this call and must not be reused. An optional attachment can be included (consumed by this call). An optional timestamp can be attached to the message; borrowed, not consumed.

Implementation

void putBytes(
  ZBytes payload, {
  Encoding? encoding,
  ZBytes? attachment,
  Timestamp? timestamp,
}) {
  _ensureOpen();
  // ALLOCATE-LAST: both handle reads run before the first allocation. The
  // attachment read used to sit AFTER the encoding string was allocated, so
  // a disposed or consumed attachment stranded it.
  final payloadPtr = payload.nativePtr;
  final attachmentPtr = attachment != null ? attachment.nativePtr : nullptr;
  final loaned = bindings.zd_publisher_loan(_ptr.cast());

  // Two INDEPENDENT length-carried channels (R-2), from the RAW pair (R-3a).
  final (mime, schema) = encoding != null
      ? encodingWireChannels(encoding)
      : (null, null);
  final encodingBuf = allocLengthCarriedUtf8(mime);
  final schemaBuf = allocLengthCarriedUtf8(schema);
  final tsPtr = timestamp != null ? _timestampToNative(timestamp) : nullptr;

  try {
    final rc = bindings.zd_publisher_put(
      loaned,
      payloadPtr.cast(),
      encodingBuf.ptr,
      encodingBuf.len,
      schemaBuf.ptr,
      schemaBuf.len,
      attachmentPtr.cast(),
      tsPtr.cast(),
    );

    payload.markConsumed();
    if (attachment != null) attachment.markConsumed();

    if (rc != 0) {
      throw ZenohException('Publisher put failed', rc);
    }
  } finally {
    // allocLengthCarriedUtf8 uses calloc, so the release is calloc.free.
    if (encodingBuf.ptr != nullptr) calloc.free(encodingBuf.ptr);
    if (schemaBuf.ptr != nullptr) calloc.free(schemaBuf.ptr);
    if (tsPtr != nullptr) calloc.free(tsPtr);
  }
}