mosaik 0.3.13

A Rust runtime for building self-organizing, leaderless distributed systems.
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
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

use {
	core::{
		panic::Location,
		pin::Pin,
		task::{Context, Poll, Waker},
		time::Duration,
	},
	derive_more::Debug,
	futures::TryFutureExt,
	std::{process::Output, sync::OnceLock},
	tokio::time::Timeout,
};

/// Applies a time factor to durations for testing purposes.
/// this for example when you are developing on an airplane wifi and
/// have very high latencies.
fn time_factor() -> f32 {
	static MULTIPLIER: OnceLock<f32> = OnceLock::new();
	*MULTIPLIER.get_or_init(|| {
		std::env::var("TIME_FACTOR")
			.ok()
			.and_then(|s| s.parse().ok())
			.unwrap_or(1.0)
	})
}

pub fn secs(count: u64) -> Duration {
	let count = (count as f32 * time_factor()) * 1000.0;
	#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
	Duration::from_millis(count as u64)
}

pub fn millis(count: u64) -> Duration {
	#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
	let count = (count as f32 * time_factor()) as u64;
	Duration::from_millis(count)
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
#[error("timeout elapsed after {0:?} at {1}")]
pub struct TimeoutElapsed(pub Duration, pub &'static Location<'static>);

#[track_caller]
pub fn timeout_s<F>(
	count: u64,
	future: F,
) -> impl Future<Output = Result<F::Output, TimeoutElapsed>>
where
	F: IntoFuture,
{
	let caller = Location::caller();
	let duration = secs(count);
	tokio::time::timeout(duration, future)
		.map_err(move |_| TimeoutElapsed(duration, caller))
}

#[track_caller]
pub fn timeout_after<F>(
	duration: Duration,
	future: F,
) -> impl Future<Output = Result<F::Output, TimeoutElapsed>>
where
	F: IntoFuture,
{
	let caller = Location::caller();
	tokio::time::timeout(duration, future)
		.map_err(move |_| TimeoutElapsed(duration, caller))
}

#[track_caller]
pub fn timeout_ms<F>(
	count: u64,
	future: F,
) -> impl Future<Output = Result<F::Output, TimeoutElapsed>>
where
	F: IntoFuture,
{
	let caller = Location::caller();
	let duration = millis(count);
	tokio::time::timeout(duration, future)
		.map_err(move |_| TimeoutElapsed(duration, caller))
}

pub fn sleep_s(count: u64) -> impl Future<Output = ()> {
	tokio::time::sleep(secs(count))
}

pub fn sleep_ms(count: u64) -> impl Future<Output = ()> {
	tokio::time::sleep(millis(count))
}