AdvancedSubscriber.declare constructor

AdvancedSubscriber.declare(Pointer<Void> loanedSession, Pointer<Void> loanedKe, String keyExpr, { AdvancedSubscriberOptions options = const AdvancedSubscriberOptions(), })

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

This is called internally by Session.declareAdvancedSubscriber.

Implementation

factory AdvancedSubscriber.declare(
  Pointer<Void> loanedSession,
  Pointer<Void> loanedKe,
  String keyExpr, {
  AdvancedSubscriberOptions options = const AdvancedSubscriberOptions(),
}) {
  final size = bindings.zd_advanced_subscriber_sizeof();
  final ptr = calloc.allocate<Void>(size);

  final sampleChannel = Subscriber.createSampleChannel(
    retainPayload: options.retainPayload,
  );

  final rc = bindings.zd_declare_advanced_subscriber(
    loanedSession.cast(),
    ptr.cast(),
    loanedKe.cast(),
    sampleChannel.receivePort.sendPort.nativePort,
    options.history,
    options.detectLatePublishers,
    options.recovery,
    options.lastSampleMissDetection,
    options.periodicQueriesPeriodMs,
    options.subscriberDetection,
    options.retainPayload ? 1 : 0,
  );

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

  ReceivePort? missPort;
  StreamController<MissEvent>? missController;

  if (options.enableMissListener) {
    missPort = ReceivePort();
    missController = StreamController<MissEvent>();

    missPort.listen((dynamic message) {
      if (message is List) {
        final zidBytes = message[0] as Uint8List;
        final count = message[1] as int;
        // eid appended at index 2 by _zd_miss_callback; length-guarded so a
        // stale/short array (defensive) degrades to eid 0 rather than throwing.
        final eid = message.length > 2 ? message[2] as int : 0;
        final sourceId = EntityGlobalId(ZenohId(zidBytes), eid);
        missController!.add(MissEvent(sourceId: sourceId, count: count));
      }
    });

    final loaned = bindings.zd_advanced_subscriber_loan(ptr.cast());
    final missRc = bindings
        .zd_advanced_subscriber_declare_background_sample_miss_listener(
          loaned,
          missPort.sendPort.nativePort,
        );

    if (missRc != 0) {
      missPort.close();
      unawaited(missController.close());
      sampleChannel.abandon();
      bindings.zd_advanced_subscriber_drop(ptr.cast());
      calloc.free(ptr);
      throw ZenohException('Failed to declare miss listener', missRc);
    }
  }

  SampleChannel? detectChannel;

  final detect = options.detectPublishers;
  if (detect != null) {
    detectChannel = Subscriber.createSampleChannel(
      retainPayload: detect.retainPayload,
    );

    final loaned = bindings.zd_advanced_subscriber_loan(ptr.cast());
    final detectRc = bindings
        .zd_advanced_subscriber_detect_publishers_background(
          loaned,
          detectChannel.receivePort.sendPort.nativePort,
          // -1 = unspecified -> NULL options -> canon's own default.
          detect.history == null ? -1 : (detect.history! ? 1 : 0),
          detect.retainPayload ? 1 : 0,
        );

    if (detectRc != 0) {
      // The shipped teardown template, extended by one pair: everything
      // declared above this point comes down before the throw.
      detectChannel.abandon();
      missPort?.close();
      unawaited(missController?.close());
      sampleChannel.abandon();
      bindings.zd_advanced_subscriber_drop(ptr.cast());
      calloc.free(ptr);
      throw ZenohException('Failed to declare publisher detection', detectRc);
    }
  }

  return AdvancedSubscriber._(
    ptr,
    sampleChannel,
    missPort,
    missController,
    detectChannel,
    keyExpr,
  );
}