Struct Subscription 

pub struct Subscription<T, D> { /* private fields */ }

r[impl jetstream.subscription.surface] The caller’s end of a subscription, in the types the caller declared.

A Stream of Item<T, D>, so consuming one is consuming a stream — and dropping one cancels it, because dropping the RpcStream inside does. Nothing in the type says whether the subscription got a lane of its own or a tag on a shared one, per r[jetstream.subscription.realisation.opaque].

Implementations

impl<T, D> Subscription<T, D>

where T: Send + 'static, D: Send + 'static,

pub fn from_frames<P, F>(stream: RpcStream<P>, decode: F) -> Subscription<T, D>

where P: Protocol + 'static, <P as Protocol>::Response: 'static, F: FnMut(Frame<<P as Protocol>::Response>) -> Result<Option<Item<T, D>>, Error> + Send + 'static,

Build one from the frames of a streaming call. decode is the generated protocol’s: only it knows which response variant is an item and which is the terminator. decode returns None for a frame that ends the subscription without an item — the terminator of a subscription cut short, which frees the tag and carries no result, because a producer stopped by cancellation never produced one. Fabricating a default there would report a result that did not happen.

pub fn opening<F>(open: F) -> Subscription<T, D>

where F: Future<Output = Subscription<T, D>> + Send + 'static,

Build one that opens when it is first polled.

r[impl jetstream.subscription.surface.rust] Opening a subscription needs a tag and a request on the wire, both of which are asynchronous — and the surface the specification shows is not: fn events(&self, ..) -> Subscription<Event, Closed>. Deferring the open to the first poll is what reconciles them, and it is what a stream does anyway: nothing happens until something reads.

pub async fn establish(&mut self)

Put the request on the wire now, rather than at the first read.

r[impl jetstream.subscription.surface.rust] Opening lazily is what keeps the method signature a plain fn, and it makes “subscribe, then act, then read” a race: the act happens before the subscription exists. Awaiting this first removes that race on one lane, where r[jetstream.lane.delivery-order] then orders the request ahead of whatever follows it.

It does not order a subscription against a call on a different lane — r[jetstream.lane.no-cross-lane-order] — and nothing can. A subscription that must not miss anything says where to start, and its producer replays from there; that is what the cursor in r[jetstream.subscription.resume] is for. Cancel-safe: the opening future stays in the subscription while it is awaited.

It used to be moved out and Source::Taken left in its place, so dropping this future — a timeout, a losing select! branch — dropped the only opening future with it and left the subscription permanently Taken: a second establish() did nothing at all, and the next poll hit an unreachable!. Awaiting in place means an abandoned establishment leaves the subscription exactly as it was, ready to be established or polled again.

pub fn served<F, S>(produce: F) -> Subscription<T, D>

where F: FnOnce(CancellationToken) -> S + Send + 'static, S: Stream<Item = Result<Item<T, D>, Error>> + Send + 'static,

r[impl jetstream.subscription.surface.producer] Build the producer’s side: a subscription that is handed its cancellation when the dispatcher serves it.

pub fn producing<F, Fut>(capacity: usize, produce: F) -> Subscription<T, D>

where F: FnOnce(Producer<T, D>) -> Fut + Send + 'static, Fut: Future<Output = ()> + Send + 'static,

r[impl jetstream.subscription.surface.producer] The ergonomic producer: a task holding a Producer, which can send and learn that its subscriber has gone.

r[jetstream.subscription.detached.state] is the reason this takes a closure rather than a sender: what the closure captures is the producer’s whole state, and a handler that can only be written as “hold this sender” cannot be evicted and resumed.

pub fn serve( self, cancel: CancellationToken, ) -> Pin<Box<dyn Stream<Item = Result<Item<T, D>, Error>> + Send>>

Take the item sequence, giving the producer its cancellation.

r[impl jetstream.subscription.cancel] This is the dispatcher’s call, and it is where cancellation reaches the work. A subscription that is already a sequence — a caller’s — ignores the token, which is right: nothing on that side produces.

pub fn from_items<S>(items: S) -> Subscription<T, D>

where S: Stream<Item = Result<Item<T, D>, Error>> + Send + 'static,

Build one from any stream of already-decoded items — the shape a test, an in-process implementation, or a replay uses.

pub fn labelled<K>(self, key: K) -> Labelled<K, T, D>

where K: Clone,

r[impl jetstream.subscription.surface.composition] Tag this subscription so a merge can say which one spoke. The terminator survives the merge on its own — it is an Item — but which subscription ended does not, and that is the other half of what the rule requires.

Trait Implementations

impl<T, D> Stream for Subscription<T, D>

type Item = Result<Item<T, D>, Error>

Values yielded by the stream.

fn poll_next( self: Pin<&mut Subscription<T, D>>, cx: &mut Context<'_>, ) -> Poll<Option<<Subscription<T, D> as Stream>::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.