canary 0.3.3

provides ergonomic abstractions for network communication and more
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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154

use crate::Result;

use crate::channel::handshake::Handshake;
use crate::err;
use crate::Channel;

use cfg_if::cfg_if;

cfg_if! {
    if #[cfg(not(target_arch = "wasm32"))] {
        use crate::io::{TcpListener, ToSocketAddrs};
        use crate::io::wss;
        use backoff::ExponentialBackoff;
    } else {
        use crate::io::Wss;
    }
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(derive_more::From, derive_more::Into)]
#[into(owned, ref, ref_mut)]
/// Websocket Provider
pub struct WebSocket(TcpListener);

#[cfg(target_arch = "wasm32")]
pub struct WebSocket;

#[cfg(not(target_arch = "wasm32"))]
impl WebSocket {
    #[inline]
    /// Bind to this address
    /// ```no_run
    /// let wss = WebSocket::bind("127.0.0.1:8080").await?;
    /// while let Ok(chan) = wss.next().await {
    ///     let mut chan = chan.encrypted().await?;
    ///     chan.send("hello!").await?;
    /// }
    /// ```
    pub async fn bind(addrs: impl ToSocketAddrs) -> Result<Self> {
        let listener = TcpListener::bind(addrs).await?;
        Ok(WebSocket(listener))
    }
    #[inline]
    /// get the next channel
    /// ```no_run
    /// while let Ok(chan) = wss.next().await {
    ///     let mut chan = chan.encrypted().await?;
    ///     chan.send("hello!").await?;
    /// }
    /// ```
    pub async fn next(&self) -> Result<Handshake> {
        let (chan, _) = self.0.accept().await?;
        let raw = wss::tokio::accept_async(chan)
            .await // this future doesn't suspend, hence why this await point is not delegated upwards.
            .map_err(|e| err!(e))?;
        let raw = Box::new(raw);
        Ok(Handshake::from(Channel::from_raw(
            raw,
            Default::default(),
            Default::default(),
        )))
    }

    /// connect to address without any backoff strategy
    pub async fn connect_no_backoff(
        addrs: impl ToSocketAddrs + std::fmt::Debug,
    ) -> Result<Handshake> {
        let addrs = tokio::net::lookup_host(&addrs)
            .await
            .map_err(|e| err!(e))?
            .next()
            .ok_or(err!("no endpoint found"))?;
        let (raw, _) = wss::tokio::connect_async(&format!("ws://{}", &addrs))
            .await
            .map_err(err!(@other))?;
        let raw = Box::new(raw);
        Ok(Handshake::from(Channel::from_raw(
            raw,
            Default::default(),
            Default::default(),
        )))
    }
    #[inline]
    /// Connect to the following address with the given id and retry in case of failure
    pub async fn connect(addrs: impl ToSocketAddrs + std::fmt::Debug) -> Result<Handshake> {
        let addrs = tokio::net::lookup_host(&addrs)
            .await
            .map_err(|e| err!(e))?
            .next()
            .ok_or(err!("no endpoint found"))?;
        let hs = backoff::future::retry(ExponentialBackoff::default(), || async {
            let (raw, _) = wss::tokio::connect_async(&format!("ws://{}", &addrs))
                .await
                .map_err(err!(@other))?;
            let raw = Box::new(raw);
            Ok(Handshake::from(Channel::from_raw(
                raw,
                Default::default(),
                Default::default(),
            )))
        })
        .await?;
        Ok(hs)
    }
}
#[cfg(target_arch = "wasm32")]
impl WebSocket {
    #[inline]
    /// connect to the following address without discovery
    pub async fn inner_connect(addrs: &str, retries: u32, time_to_retry: u64) -> Result<Wss> {
        let mut attempt = 0;
        let stream = loop {
            match reqwasm::websocket::futures::WebSocket::open(&format!("ws://{}", addrs)) {
                Ok(s) => break s,
                Err(e) => {
                    tracing::error!(
                        "connecting to address `{}` failed, attempt {} starting",
                        addrs,
                        attempt
                    );
                    async_timer::timed(
                        std::future::pending::<()>(),
                        std::time::Duration::from_millis(time_to_retry),
                    )
                    .await
                    .ok();

                    attempt += 1;
                    if attempt == retries {
                        err!((e.to_string()))?
                    }
                    continue;
                }
            }
        };
        Ok(stream)
    }
    #[inline]
    /// connect to the following address with the following id. Defaults to 3 retries.
    pub async fn connect(addrs: &str) -> Result<Handshake> {
        Self::connect_retry(addrs, 3, 10).await
    }
    #[inline]
    /// connect to the following address with the given id and retry in case of failure
    pub async fn connect_retry(addrs: &str, retries: u32, time_to_retry: u64) -> Result<Handshake> {
        let raw = Self::inner_connect(addrs, retries, time_to_retry).await?;
        let raw = Box::new(raw);
        Ok(Handshake::from(Channel::from_raw(
            raw,
            Default::default(),
            Default::default(),
        )))
    }
}