polyoxide-ws 0.39.0

The WebSocket kit the polyoxide socket clients share, one copy of each block they would otherwise each carry
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

//! The helpers the offline suites share (W15).
//!
//! The basis is `polyoxide-perps`' former `fast` and `next_event`
//! (`ws/supervised.rs`, deleted in Story 4.6), `polyoxide-binance`'s
//! (`tests/supervision.rs`) and `polyoxide-sports`' `next_item` and `drain`
//! (`tests/supervision.rs`).
//! Counting pings is [`ScriptedServer::ping_count`](super::ScriptedServer::ping_count)
//! for protocol pings and
//! [`ScriptedServer::count_frames`](super::ScriptedServer::count_frames) for
//! a protocol's own.

use std::{fmt::Debug, time::Duration};

use futures_util::{Stream, StreamExt};

use crate::Timing;

/// How long [`next_item`] and [`next_event`] wait. Generous, to stay steady
/// on a loaded machine.
pub const WINDOW: Duration = Duration::from_secs(3);

/// Test-sized timings: a ping every 50 ms, stale after 300 ms, and backoff
/// from 10 ms to 20 ms. The connect timeout and rotation are the defaults.
pub fn fast() -> Timing {
    Timing {
        ping_interval: Some(Duration::from_millis(50)),
        stale_after: Duration::from_millis(300),
        initial_backoff: Duration::from_millis(10),
        max_backoff: Duration::from_millis(20),
        ..Timing::default()
    }
}

/// The next item, which must arrive within [`WINDOW`]; panics if the stream
/// ends.
///
/// The deadline is checked first, so an item that arrives only because the
/// deadline woke the task, which is a lost wakeup, fails rather than passing
/// late.
pub async fn next_item<S: Stream + Unpin>(stream: &mut S) -> S::Item {
    let deadline = tokio::time::sleep(WINDOW);
    tokio::pin!(deadline);
    tokio::select! {
        biased;
        () = &mut deadline => panic!("no item within {WINDOW:?}"),
        item = stream.next() => item.expect("the stream ended"),
    }
}

/// The next item, which must be an `Ok` arriving within [`WINDOW`].
pub async fn next_event<S, T, E>(stream: &mut S) -> T
where
    S: Stream<Item = Result<T, E>> + Unpin,
    E: Debug,
{
    next_item(stream).await.expect("an event, not an error")
}

/// Read the stream until it ends, or until the task reading it is aborted,
/// so the server sees every reconnect.
pub async fn drain<S: Stream + Unpin>(mut stream: S) {
    while stream.next().await.is_some() {}
}