declareSubscriber method

Subscriber declareSubscriber(Object keyExpr, { Locality? allowedOrigin, bool retainPayload = false, })

Declares a subscriber on the given keyExpr.

Returns a Subscriber whose Subscriber.stream delivers Samples. Call Subscriber.close when done to undeclare and release resources.

⚠️ This stream is UNBOUNDED, and pausing it does not stop the flow. Arrivals are pushed into a StreamController as they land, so a paused or slow consumer accumulates them without limit — pause() throttles delivery to your listener, never the producer, and nothing in this package wires the gate callbacks that would.

Measured on the shipped package, two OS processes over TCP, 64 MiB posted into a listener paused before any traffic: resident memory grew by 154 MiB on this surface, against 3 MiB for the bounded alternative on the same load.

For a bounded consumer use Session.declarePullSubscriber and its PullSubscriber.stream: it pulls one sample at a time out of a bounded native channel and only while the subscription is demanding, so what accumulates is the channel's own capacity plus at most one already-pulled sample. What happens to the traffic that does not fit is the channel's ChannelKind — ChannelKind.ring drops and keeps the publisher running, ChannelKind.fifo holds the publisher back.

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

Throws ZenohException if the key expression is invalid. Throws StateError if the session has been closed. retainPayload makes every delivered sample carry an owned Sample.payloadZBytes handle on its payload, so it can be republished without copying or asked what backs it. Off by default — nothing pays for retention that did not ask for it — and Sample.payloadBytes is unchanged either way. ⛔ A retained handle is the caller's to release.

Implementation

Subscriber declareSubscriber(
  Object keyExpr, {
  Locality? allowedOrigin,
  bool retainPayload = false,
}) {
  return _withKeyExprArg(keyExpr, 'keyExpr', (loanedSession, loanedKe) {
    return Subscriber.declare(
      loanedSession,
      loanedKe,
      keyExpr: keyExprString(keyExpr, 'keyExpr'),
      allowedOrigin: allowedOrigin,
      retainPayload: retainPayload,
    );
  });
}