Struct EndCheck 

pub struct EndCheck { /* private fields */ }

Decides when a bounded job on a feed has ended.

A windowed check opens a window on a read of the end marker. The window spans from its first read of the feed. Inside it the feed is read every 15 s, each read taken within END_CHECK_INTERVAL of the one before and answered within it of its request, and every read must find the feed empty. A read requested a window or more after the first that finds it empty is followed by a second read of the marker: the same run passes, another run opens a new window, and no marker restarts the check. An exact check, for a feed whose counts are exact and whose lister has finished, reads the feed once with no window or marker. Either way one receive that takes nothing follows, and then the dead-letter destination decides between Completed and Stalled.

A non-empty read, a read that fails retryably, a read taken or answered later than that bound, a marker read answered later than it included, or a restart sends a windowed check back to its marker read, 15 s after its previous read at most. A dead-letter read that fails retryably is retried 15 s later and ends nothing. A depth read that fails with a Fatal class ends the check Failed. The source calls restart whenever its reader holds any work, so every read that counts was answered while it held none.

use spate_core::feed::{Depth, DepthOf, EndAction, EndCheck, EndOutcome, MarkerRead};
use std::time::{Duration, Instant};

let start = Instant::now();
let mut check = EndCheck::windowed(Duration::from_secs(30), start);
assert_eq!(check.next(start), EndAction::ReadMarker);
check.marker_read(start, MarkerRead::Present("run-a".into()));

let empty = || Ok(Depth::new("queue", false));
for at in [0, 15, 30] {
    let now = start + Duration::from_secs(at);
    assert_eq!(check.next(now), EndAction::ReadDepth(DepthOf::Feed));
    check.depth_read(now, empty());
}
let now = start + Duration::from_secs(30);
assert_eq!(check.next(now), EndAction::ReadMarker);
check.marker_read(now, MarkerRead::Present("run-a".into()));
assert_eq!(check.next(now), EndAction::Receive);
check.received(now, false);
assert_eq!(check.next(now), EndAction::ReadDepth(DepthOf::DeadLetters));
check.depth_read(now, empty());
assert_eq!(check.next(now), EndAction::Ended(EndOutcome::Completed));

Implementations

impl EndCheck

pub fn windowed(window: Duration, now: Instant) -> EndCheck

A check that ends a job once the feed stays empty over window between two reads of the same lister run’s end marker. Its first read is due at now.

pub fn exact(now: Instant) -> EndCheck

A check for a feed whose counts are exact and whose lister has finished: one empty read of the feed passes, with no window and no marker. Its first read is due at now.

pub fn next(&mut self, now: Instant) -> EndAction

What to do at now. A read returned here is outstanding until its answer is reported; until then next returns Wait(None).

pub fn marker_read(&mut self, now: Instant, read: MarkerRead)

Report the end marker read that next asked for. An answer nothing asked for is ignored.

pub fn depth_read(&mut self, now: Instant, read: Result<Depth, FeedError>)

Report the depth read that next asked for. An answer nothing asked for is ignored.

pub fn received(&mut self, now: Instant, took: bool)

Report whether the receive that next asked for took a unit. One that did restarts the check.

pub fn restart(&mut self, now: Instant)

Send the check back to its opening read, because the reader holds work. A windowed check reads again 15 s after its last read; an exact one at once. A check that has ended stays ended.

Drop every read outstanding when calling it. The check tells answers apart only by kind, so an answer reported once the next read of its kind has started counts as that read’s.

pub fn outcome(&self) -> Option<&EndOutcome>

Whether the check has ended.

Trait Implementations

impl Debug for EndCheck

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.