anyswap_sdk/
ws.rs

1//! WebSocket client for `/v1/subscribe/{user_id}`: one signal per FSM
2//! transition of any swap of this wallet. A signal only says which swap moved;
3//! read the state with [`HttpClient::get_swap`]. `HttpClient::signals` hands
4//! it to the driver as its push channel.
5
6use anyswap_core::api::SwapStateUpdate;
7use futures_util::StreamExt;
8use thiserror::Error;
9use tokio::net::TcpStream;
10use tokio_tungstenite::{MaybeTlsStream, WebSocketStream, connect_async, tungstenite};
11
12use crate::client::{ClientError, HttpClient};
13
14#[derive(Debug, Error)]
15pub enum WsError {
16    #[error(transparent)]
17    Client(#[from] ClientError),
18
19    #[error("websocket: {0}")]
20    Connect(#[from] tungstenite::Error),
21}
22
23/// An open subscription. It ends when the server closes it or the transport
24/// fails; subscribe again to reconnect.
25pub struct Subscription(WebSocketStream<MaybeTlsStream<TcpStream>>);
26
27impl HttpClient {
28    pub async fn subscribe(&self) -> Result<Subscription, WsError> {
29        let (socket, _) = connect_async(self.subscribe_url()?).await?;
30        Ok(Subscription(socket))
31    }
32}
33
34impl Subscription {
35    /// The next update, or `None` once the subscription has ended.
36    pub async fn next(&mut self) -> Option<SwapStateUpdate> {
37        while let Some(message) = self.0.next().await {
38            match message {
39                Ok(tungstenite::Message::Text(text)) => {
40                    if let Ok(update) = serde_json::from_str(&text) {
41                        return Some(update);
42                    }
43                }
44                Ok(tungstenite::Message::Close(_)) | Err(_) => return None,
45                Ok(_) => {}
46            }
47        }
48        None
49    }
50}
51
52#[cfg(feature = "driver")]
53pub use self::signals::WsSignals;
54
55#[cfg(feature = "driver")]
56mod signals {
57    use async_trait::async_trait;
58    use uuid::Uuid;
59
60    use super::Subscription;
61    use crate::{client::HttpClient, driver::Signals};
62
63    /// The subscription as the driver's push channel: opened on the first
64    /// `next`, dropped when it ends so the next call reopens it.
65    pub struct WsSignals {
66        client: HttpClient,
67        subscription: Option<Subscription>,
68    }
69
70    impl HttpClient {
71        pub fn signals(&self) -> WsSignals {
72            WsSignals {
73                client: self.clone(),
74                subscription: None,
75            }
76        }
77    }
78
79    #[async_trait]
80    impl Signals for WsSignals {
81        async fn next(&mut self) -> Option<Uuid> {
82            if self.subscription.is_none() {
83                self.subscription = Some(self.client.subscribe().await.ok()?);
84            }
85            let update = self.subscription.as_mut()?.next().await;
86            if update.is_none() {
87                self.subscription = None;
88            }
89            update.map(|update| update.swap_id)
90        }
91    }
92}
anyswap_sdk/ws.rs

anyswap_sdk/
ws.rs
