synd_client/client/feed_events/
connection.rs

1use reqwest::header::{self, HeaderValue};
2use tokio::net::TcpStream;
3#[cfg(unix)]
4use tokio::net::UnixStream;
5#[cfg(unix)]
6use tokio_tungstenite::client_async;
7use tokio_tungstenite::{
8    MaybeTlsStream, connect_async_tls_with_config,
9    tungstenite::{client::IntoClientRequest as _, handshake::client::Request},
10};
11use url::Url;
12
13use super::protocol::FeedEventConnection;
14use crate::client::{Client, FeedEventTransport};
15use crate::{SyndApiError, payload::FeedEvent};
16
17const FEED_EVENTS_PATH: &str = "/graphql/ws";
18
19/// Active feed-event watch over the client's configured transport.
20pub struct FeedEventWatch {
21    connection: FeedEventWatchConnection,
22}
23
24enum FeedEventWatchConnection {
25    Tcp(Box<FeedEventConnection<MaybeTlsStream<TcpStream>>>),
26    #[cfg(unix)]
27    Unix(Box<FeedEventConnection<UnixStream>>),
28}
29
30impl FeedEventWatch {
31    pub async fn next_event(&mut self) -> Result<FeedEvent, SyndApiError> {
32        match &mut self.connection {
33            FeedEventWatchConnection::Tcp(connection) => connection.next_event().await,
34            #[cfg(unix)]
35            FeedEventWatchConnection::Unix(connection) => connection.next_event().await,
36        }
37    }
38}
39
40impl Client {
41    pub async fn watch_feed_events(&self) -> Result<FeedEventWatch, SyndApiError> {
42        let request = self.feed_event_request()?;
43        match &self.feed_event_transport {
44            FeedEventTransport::Tcp { connector } => {
45                let (socket, _) =
46                    connect_async_tls_with_config(request, None, false, connector.clone())
47                        .await
48                        .map_err(SyndApiError::WebSocket)?;
49                Ok(FeedEventWatch {
50                    connection: FeedEventWatchConnection::Tcp(Box::new(
51                        FeedEventConnection::establish(socket).await?,
52                    )),
53                })
54            }
55            #[cfg(unix)]
56            FeedEventTransport::Unix { socket_path } => {
57                let stream = UnixStream::connect(socket_path).await.map_err(|error| {
58                    SyndApiError::WebSocket(tokio_tungstenite::tungstenite::Error::Io(error))
59                })?;
60                let (socket, _) = client_async(request, stream)
61                    .await
62                    .map_err(SyndApiError::WebSocket)?;
63                Ok(FeedEventWatch {
64                    connection: FeedEventWatchConnection::Unix(Box::new(
65                        FeedEventConnection::establish(socket).await?,
66                    )),
67                })
68            }
69        }
70    }
71
72    fn feed_event_request(&self) -> Result<Request, SyndApiError> {
73        let mut request = self
74            .graphql_ws_endpoint()?
75            .as_str()
76            .into_client_request()
77            .map_err(SyndApiError::WebSocket)?;
78        request.headers_mut().insert(
79            header::SEC_WEBSOCKET_PROTOCOL,
80            HeaderValue::from_static("graphql-transport-ws"),
81        );
82        self.authentication
83            .apply_authorization_header(request.headers_mut())?;
84        Ok(request)
85    }
86
87    fn graphql_ws_endpoint(&self) -> Result<Url, SyndApiError> {
88        let mut url = self.endpoint.join(FEED_EVENTS_PATH)?;
89        let scheme = match url.scheme() {
90            "http" => "ws",
91            "https" => "wss",
92            scheme => {
93                return Err(SyndApiError::UnsupportedWebSocketScheme {
94                    scheme: scheme.to_owned(),
95                });
96            }
97        };
98        url.set_scheme(scheme)
99            .map_err(|()| SyndApiError::SetWebSocketScheme)?;
100        Ok(url)
101    }
102}
synd_client/client/feed_events/connection.rs

synd_client/client/feed_events/
connection.rs
