Publisher.declare constructor

Publisher.declare(Pointer<Void> loanedSession, Pointer<Void> loanedKe, { Encoding? encoding, CongestionControl? congestionControl, Priority? priority, bool? isExpress, Locality? allowedDestination, bool enableMatchingListener = false, })

Creates a publisher on the given session and key expression.

This is called internally by Session.declarePublisher.

congestionControl, priority, isExpress and allowedDestination are each optional; omitting one — or passing null — means canon decides. A publisher is a push path, so canon's congestion default here is CongestionControl.drop; the other defaults are Priority.data, isExpress: false and Locality.any.

⚠️ Read CongestionControl before selecting CongestionControl.block.

Implementation

factory Publisher.declare(
  Pointer<Void> loanedSession,
  Pointer<Void> loanedKe, {
  Encoding? encoding,
  CongestionControl? congestionControl,
  Priority? priority,
  bool? isExpress,
  Locality? allowedDestination,
  bool enableMatchingListener = false,
}) {
  final size = bindings.zd_publisher_sizeof();
  final ptr = calloc.allocate<Void>(size);

  // 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);

  try {
    final rc = bindings.zd_declare_publisher(
      loanedSession.cast(),
      ptr.cast(),
      loanedKe.cast(),
      encodingBuf.ptr,
      encodingBuf.len,
      schemaBuf.ptr,
      schemaBuf.len,
      congestionControl?.value ?? -1,
      priority?.value ?? -1,
      isExpress == null ? -1 : (isExpress ? 1 : 0),
      allowedDestination?.value ?? -1,
    );

    if (rc != 0) {
      calloc.free(ptr);
      throw ZenohException('Failed to declare publisher', 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);
  }

  ReceivePort? matchingPort;
  StreamController<bool>? matchingController;

  if (enableMatchingListener) {
    matchingPort = ReceivePort();
    matchingController = StreamController<bool>();

    matchingPort.listen((dynamic message) {
      if (message is int) {
        matchingController!.add(message != 0);
      }
    });

    final loaned = bindings.zd_publisher_loan(ptr.cast());
    final mlRc = bindings.zd_publisher_declare_background_matching_listener(
      loaned,
      matchingPort.sendPort.nativePort,
    );

    if (mlRc != 0) {
      matchingPort.close();
      unawaited(matchingController.close());
      bindings.zd_publisher_drop(ptr.cast());
      calloc.free(ptr);
      throw ZenohException('Failed to declare matching listener', mlRc);
    }
  }

  return Publisher._(ptr, matchingPort, matchingController);
}