stream property
A stream of Samples received by this subscriber.
⚠️ 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 — ring drops and keeps the publisher
running, fifo holds the publisher back.
⚠️ A value this binding cannot convert arrives as a STREAM ERROR, never
as empty or absent data. Zenoh can hand over a payload or attachment
the conversion step refuses; delivering that as a zero-length value would
be indistinguishable from a legitimately empty one, and empty is a real
value on this path. The error carries canon's own code.
⛔ The stream keeps running. A conversion failure is a failed call,
not a dead channel — the same rule the pull family already ships — so
later values still arrive and terminating on the first bad one would
lose them. ⚠️ But an unhandled stream error is still an unhandled error:
pass onError (or handleError), because an unhandled one can take the
program down.
Implementation
Stream<Sample> get stream => _channel.stream;