AdvancedPublisher.declare constructor

AdvancedPublisher.declare(Pointer<Void> loanedSession, Pointer<Void> loanedKe, String keyExpr, { bool enableCache = false, int? cacheMaxSamples, bool publisherDetection = false, bool sampleMissDetection = false, HeartbeatMode heartbeatMode = HeartbeatMode.none, int heartbeatPeriodMs = 0, bool enableMatchingListener = false, })

Creates an advanced publisher on the given session and key expression.

This is called internally by Session.declareAdvancedPublisher.

Implementation

factory AdvancedPublisher.declare(
  Pointer<Void> loanedSession,
  Pointer<Void> loanedKe,
  String keyExpr, {
  bool enableCache = false,
  int? cacheMaxSamples,
  bool publisherDetection = false,
  bool sampleMissDetection = false,
  HeartbeatMode heartbeatMode = HeartbeatMode.none,
  int heartbeatPeriodMs = 0,
  bool enableMatchingListener = false,
}) {
  // ALLOCATE-LAST: the domain check runs before anything is allocated.
  // `Session.declareAdvancedPublisher` rejects a negative bound earlier
  // still, before the key expression is even loaned; this is the second
  // Dart-side gate, and the shim's ZD_DECLARE_ECAPACITY is the structural
  // backstop at the seam itself.
  if (cacheMaxSamples != null && cacheMaxSamples < 0) {
    throw ArgumentError.value(
      cacheMaxSamples,
      'maxSamples',
      "must be >= 0; null leaves canon's own default in place",
    );
  }

  final size = bindings.zd_advanced_publisher_sizeof();
  final ptr = calloc.allocate<Void>(size);

  final rc = bindings.zd_declare_advanced_publisher(
    loanedSession.cast(),
    ptr.cast(),
    loanedKe.cast(),
    enableCache,
    // -1 is the shim's "unspecified" sentinel: canon's default is left
    // untouched. Every value >= 0 is assigned verbatim, zero included.
    cacheMaxSamples ?? -1,
    publisherDetection,
    sampleMissDetection,
    heartbeatMode.value,
    heartbeatPeriodMs,
  );

  if (rc != 0) {
    calloc.free(ptr);
    throw ZenohException('Failed to declare advanced publisher', rc);
  }

  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_advanced_publisher_loan(ptr.cast());
    final mlRc = bindings
        .zd_advanced_publisher_declare_background_matching_listener(
          loaned,
          matchingPort.sendPort.nativePort,
        );

    if (mlRc != 0) {
      // The shipped second-listener teardown template: nothing declared
      // survives the failure, and the throw comes last.
      matchingPort.close();
      unawaited(matchingController.close());
      bindings.zd_advanced_publisher_drop(ptr.cast());
      calloc.free(ptr);
      throw ZenohException('Failed to declare matching listener', mlRc);
    }
  }

  return AdvancedPublisher._(ptr, keyExpr, matchingPort, matchingController);
}