declareBackgroundSubscriber method

Stream<Sample> declareBackgroundSubscriber(Object keyExpr, { Locality? allowedOrigin, bool retainPayload = false, })

Declares a background subscriber on the given keyExpr.

Returns a Stream of Samples. Unlike declareSubscriber, the background subscriber has no handle and cannot be explicitly closed. It lives until the session is closed, at which point the stream completes automatically.

allowedOrigin restricts whose traffic this declaration accepts. Omitting it — or passing null — means canon decides, which is Locality.any. See Locality.

⚠️ This stream is UNBOUNDED, and pausing it does not stop the flow — see declareSubscriber for the measurement. There is deliberately no bounded form of this surface: a background declaration hands back no handle, so there is nothing to close and nothing to pace it with. That is a carve-out, not an oversight. Use declarePullSubscriber when you need a bound.

Throws ZenohException if the key expression is invalid. Throws StateError if the session has been closed.

Implementation

Stream<Sample> declareBackgroundSubscriber(
  Object keyExpr, {
  Locality? allowedOrigin,

  /// See [declareSubscriber] for what `retainPayload` costs and promises.
  bool retainPayload = false,
}) {
  return _withKeyExprArg(keyExpr, 'keyExpr', (loanedSession, loanedKe) {
    // ALLOCATE-LAST: the channel is created only once the key expression has
    // been accepted, so a rejected one cannot strand an open ReceivePort.
    final channel = Subscriber.createSampleChannel(
      retainPayload: retainPayload,
    );
    final rc = bindings.zd_declare_background_subscriber(
      loanedSession.cast(),
      loanedKe.cast(),
      channel.receivePort.sendPort.nativePort,
      allowedOrigin?.value ?? -1,
      retainPayload ? 1 : 0,
    );

    if (rc != 0) {
      channel.abandon();
      throw ZenohException('Failed to declare background subscriber', rc);
    }

    return channel.stream;
  });
}