stream property

Stream<Sample> get stream

A bounded, demand-gated Stream view over this handle's own recv.

Unlike Session.declareSubscriber's stream, which pushes every arrival into an unbounded StreamController seam, this one pulls: it takes one sample at a time out of the bounded native channel, and only while the subscription is demanding.

Pausing this subscription really stops the flow

Nothing is taken out of the native channel while paused, so what accumulates is the channel's own capacity and at most one already-pulled sample — never a queue that grows with traffic. What happens to the traffic that does not fit is the channel's kind, unchanged: a ChannelKind.ring drops its oldest channel entry and the publisher keeps running; a ChannelKind.fifo holds the publisher back. Those are two different promises, and kind is where you choose between them.

The "at most one" is structural rather than incidental: the first pull starts synchronously inside onListen, so at most one pull is ever in flight and the sample completing it goes to a one-slot stash. The retained amount is therefore at most capacity + 1 — a single ceiling, rather than one that depends on how the pause was reached.

⚠️ A ceiling, not an equality. On this column it is reached even under heavy overflow — measured, 1024 samples into a ring, capacity: 8 left exactly 9. On the query column, where arrivals are concurrent rather than serial, the same configuration sits one short: canon's ring makes room before it inserts.

⚠️ This getter is a MODE SWITCH

Once the returned stream has a listener, the drive loop owns this handle's recv: calling recv yourself throws StateError while a pull is in flight, and an interleaved tryRecv competes with the loop for arrivals. Pick one consumption idiom per handle.

A sample that completed a pull into a paused or cancelled subscription is stashed, not dropped: tryRecv hands it back before it touches the channel, and close releases anything still held. Measured: three samples published after a cancel() came back as three, where releasing the orphaned pull's sample instead returned two.

What the ring drops is the oldest CHANNEL entry, not the oldest sample

Because the stashed sample is not in the channel, it survives an eviction the channel makes. Measured on a ring, capacity: 4 with the subscription paused before any traffic: six samples in, five delivered — m0 (the stash) followed by m2 … m5. The entry evicted was m1.

⚠️ Capacity 0, measured per kind — this column only

On a ChannelKind.fifo at capacity 0 this stream delivers nothing while nothing else consumes it. That is recv's documented rendezvous restriction, inherited: the loop drives recv, and the parked recv is the only consumer that could release the delivery it is waiting on. ⚠️ A single interleaved tryRecv un-wedges the chain and the stream then delivers the rest — measured: the poll took the first sample and the stream then delivered the remaining three. On ChannelKind.ring at capacity 0 the stream does deliver.

Teardown, and who closes the stream in each state

With a pull still in flight — paused or not — close completes it and the loop's own terminal arm closes the stream. With the loop already exited holding a stash (a pause() taken inside onData, then one further arrival) only close's own gate step can close it. Closing the session before this handle reaches the same terminal arm. ⚠️ Closing the session first while a fifo is in overflow stalls exactly as it does for the polling handle today — pinned in test/fifo_close_deadlock_test.dart, and unchanged by this stream.

⚠️ On a liveliness carrier (Session.declarePullLivelinessSubscriber, which returns this same type, so this stream is already available there) the choice of kind carries an extra hazard — see that method's own "Think twice before choosing ring here".

Single-subscription, like every other stream in this package. Throws StateError if this subscriber has been closed, exactly as tryRecv and recv do.

Implementation

Stream<Sample> get stream {
  if (_closed) throw StateError('PullSubscriber is closed');
  return (_gate ??= DemandGate<Sample>(
    pull: recv,
    // A Sample holds no native resource, so there is nothing to release.
    release: (_) {},
  )).stream;
}