linera_client/
util.rs

1// Copyright (c) Zefchain Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use 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
16/// Parses the trimmed string as JSON into a value of type `T`.
17pub fn parse_json<T: serde::de::DeserializeOwned>(s: &str) -> anyhow::Result<T> {
18    Ok(serde_json::from_str(s.trim())?)
19}
20
21/// Parses the string as a number of milliseconds into a `Duration`.
22pub fn parse_millis(s: &str) -> Result<Duration, ParseIntError> {
23    Ok(Duration::from_millis(s.parse()?))
24}
25
26/// Parses a number of milliseconds into a `Duration`, rejecting zero.
27///
28/// For a delay, zero usually means "no throttle". For anything a timer sleeps on it means
29/// "spin as fast as the CPU allows", so the flags that feed one are parsed through here.
30pub fn parse_millis_nonzero(s: &str) -> Result<Duration, anyhow::Error> {
31    let millis = s.parse::<u64>()?;
32    anyhow::ensure!(millis > 0, "must be greater than zero");
33    Ok(Duration::from_millis(millis))
34}
35
36/// Converts a `Duration` to `Option<Duration>`, treating zero as `None`.
37pub fn non_zero_duration(d: Duration) -> Option<Duration> {
38    if d.is_zero() {
39        None
40    } else {
41        Some(d)
42    }
43}
44
45/// Parses the string as a number of seconds into a `Duration`.
46pub fn parse_secs(s: &str) -> Result<Duration, ParseIntError> {
47    Ok(Duration::from_secs(s.parse()?))
48}
49
50/// Parses the string as a number of milliseconds into a `TimeDelta`.
51pub fn parse_millis_delta(s: &str) -> Result<TimeDelta, ParseIntError> {
52    Ok(TimeDelta::from_millis(s.parse()?))
53}
54
55/// Parses the JSON string as an optional number of milliseconds into an `Option<TimeDelta>`.
56pub fn parse_json_optional_millis_delta(s: &str) -> anyhow::Result<Option<TimeDelta>> {
57    Ok(parse_json::<Option<u64>>(s)?.map(TimeDelta::from_millis))
58}
59
60/// Parses a comma-separated list of chain IDs into a set.
61pub fn parse_chain_set(s: &str) -> Result<HashSet<ChainId>, CryptoError> {
62    match s.trim() {
63        "" => Ok(HashSet::new()),
64        s => s.split(",").map(ChainId::from_str).collect(),
65    }
66}
67
68/// Parses a comma-separated list of application IDs into a set.
69pub fn parse_app_set(s: &str) -> anyhow::Result<HashSet<GenericApplicationId>> {
70    s.trim()
71        .split(",")
72        .map(|app_str| {
73            GenericApplicationId::from_str(app_str)
74                .or_else(|_| Ok(ApplicationId::from_str(app_str)?.into()))
75        })
76        .collect()
77}
78
79/// Returns after the specified time or if we receive a notification that a new round has started.
80pub async fn wait_for_next_round(stream: &mut NotificationStream, timeout: RoundTimeout) {
81    let mut stream = stream.filter(|notification| match &notification.reason {
82        Reason::NewBlock { height, .. } | Reason::NewEvents { height, .. } => {
83            *height >= timeout.next_block_height
84        }
85        Reason::NewRound { round, .. } => *round > timeout.current_round,
86        Reason::NewIncomingBundle { .. } | Reason::BlockExecuted { .. } => false,
87    });
88    future::select(
89        Box::pin(stream.next()),
90        Box::pin(linera_base::time::timer::sleep(
91            timeout.timestamp.duration_since(Timestamp::now()),
92        )),
93    )
94    .await;
95}
96
97macro_rules! impl_from_infallible {
98    ($target:path) => {
99        impl From<::std::convert::Infallible> for $target {
100            fn from(infallible: ::std::convert::Infallible) -> Self {
101                match infallible {}
102            }
103        }
104    };
105}
106
107pub(crate) use impl_from_infallible;
linera_client/util.rs

linera_client/
util.rs
