1use std::{collections::HashSet, num::ParseIntError, str::FromStr};
5
6use futures::future;
7use linera_base::{
8 crypto::CryptoError,
9 data_types::{TimeDelta, Timestamp},
10 identifiers::{ApplicationId, ChainId, GenericApplicationId},
11 time::Duration,
12};
13use linera_core::{data_types::RoundTimeout, node::NotificationStream, worker::Reason};
14use tokio_stream::StreamExt as _;
15
16pub fn non_zero_duration(d: Duration) -> Option<Duration> {
18 (d > Duration::ZERO).then_some(d)
19}
20
21pub fn parse_json<T: serde::de::DeserializeOwned>(s: &str) -> anyhow::Result<T> {
23 Ok(serde_json::from_str(s.trim())?)
24}
25
26pub fn parse_millis(s: &str) -> Result<Duration, ParseIntError> {
28 Ok(Duration::from_millis(s.parse()?))
29}
30
31pub fn parse_secs(s: &str) -> Result<Duration, ParseIntError> {
33 Ok(Duration::from_secs(s.parse()?))
34}
35
36pub fn parse_millis_delta(s: &str) -> Result<TimeDelta, ParseIntError> {
38 Ok(TimeDelta::from_millis(s.parse()?))
39}
40
41pub fn parse_json_optional_millis_delta(s: &str) -> anyhow::Result<Option<TimeDelta>> {
43 Ok(parse_json::<Option<u64>>(s)?.map(TimeDelta::from_millis))
44}
45
46pub fn parse_chain_set(s: &str) -> Result<HashSet<ChainId>, CryptoError> {
48 match s.trim() {
49 "" => Ok(HashSet::new()),
50 s => s.split(",").map(ChainId::from_str).collect(),
51 }
52}
53
54pub fn parse_app_set(s: &str) -> anyhow::Result<HashSet<GenericApplicationId>> {
56 match s.trim() {
57 "" => Ok(HashSet::new()),
60 s => s
61 .split(",")
62 .map(|app_str| {
63 GenericApplicationId::from_str(app_str)
64 .or_else(|_| Ok(ApplicationId::from_str(app_str)?.into()))
65 })
66 .collect(),
67 }
68}
69
70pub async fn wait_for_next_round(stream: &mut NotificationStream, timeout: RoundTimeout) {
72 let mut stream = stream.filter(|notification| match ¬ification.reason {
73 Reason::NewBlock { height, .. } | Reason::NewEvents { height, .. } => {
74 *height >= timeout.next_block_height
75 }
76 Reason::NewRound { round, .. } => *round > timeout.current_round,
77 Reason::NewIncomingBundle { .. } | Reason::BlockExecuted { .. } => false,
78 });
79 future::select(
80 Box::pin(stream.next()),
81 Box::pin(linera_base::time::timer::sleep(
82 timeout.timestamp.duration_since(Timestamp::now()),
83 )),
84 )
85 .await;
86}
87
88macro_rules! impl_from_infallible {
89 ($target:path) => {
90 impl From<::std::convert::Infallible> for $target {
91 fn from(infallible: ::std::convert::Infallible) -> Self {
92 match infallible {}
93 }
94 }
95 };
96}
97
98pub(crate) use impl_from_infallible;