cratestack_client_rust/subscription/
open.rs

1//! Which binding opens a subscription: the configured one, or `Auto`'s
2//! chain. Each binding this build lacks answers with a `BadInput` naming the
3//! feature it needs.
4
5use serde::de::DeserializeOwned;
6
7use super::failure::Failure;
8use super::{ClientSubscription, SubscriptionTarget, SubscriptionTransport, sse};
9use crate::client::CratestackClient;
10use crate::codec::HttpClientCodec;
11use crate::error::ClientError;
12
13const WT: bool = cfg!(all(feature = "webtransport", not(target_arch = "wasm32")));
14const WS: bool = cfg!(all(feature = "ws", not(target_arch = "wasm32")));
15
16/// How long `Auto` waits for a WebTransport session before moving on: a
17/// UDP-blocked network otherwise waits for the QUIC idle timeout.
18#[cfg(all(feature = "webtransport", not(target_arch = "wasm32")))]
19const AUTO_WT_WITHIN: std::time::Duration = std::time::Duration::from_secs(3);
20
21impl<C: HttpClientCodec> CratestackClient<C> {
22    /// Opens `target`, resolving once the server says `Ready`.
23    pub async fn subscribe<T>(
24        &self,
25        target: SubscriptionTarget,
26    ) -> Result<ClientSubscription<T>, ClientError>
27    where
28        T: DeserializeOwned + Send + 'static,
29    {
30        match self.config.subscription_transport {
31            SubscriptionTransport::Sse => sse::open(self, &target).await,
32            SubscriptionTransport::WebSocket => {
33                self.open_ws(&target).await.map_err(Failure::into_error)
34            }
35            SubscriptionTransport::WebTransport => self
36                .open_wt(&target, false)
37                .await
38                .map_err(Failure::into_error),
39            SubscriptionTransport::Auto => self.open_auto(&target).await,
40        }
41    }
42
43    async fn open_auto<T: DeserializeOwned + Send + 'static>(
44        &self,
45        target: &SubscriptionTarget,
46    ) -> Result<ClientSubscription<T>, ClientError> {
47        if WT && self.webtransport_configured() {
48            match self.open_wt(target, true).await {
49                Ok(subscription) => return Ok(subscription),
50                Err(Failure::Refused(error)) => return Err(error),
51                Err(Failure::Unreachable(_)) => {}
52            }
53        }
54        if WS {
55            match self.open_ws(target).await {
56                Ok(subscription) => return Ok(subscription),
57                Err(Failure::Refused(error)) => return Err(error),
58                Err(Failure::Unreachable(_)) => {}
59            }
60        }
61        sse::open(self, target).await
62    }
63
64    #[cfg(all(feature = "webtransport", not(target_arch = "wasm32")))]
65    fn webtransport_configured(&self) -> bool {
66        self.config
67            .webtransport
68            .as_ref()
69            .is_some_and(|endpoint| !endpoint.recently_failed())
70    }
71
72    #[cfg(not(all(feature = "webtransport", not(target_arch = "wasm32"))))]
73    fn webtransport_configured(&self) -> bool {
74        false
75    }
76
77    /// `auto`: bounded by [`AUTO_WT_WITHIN`], and a failure is remembered.
78    #[cfg(all(feature = "webtransport", not(target_arch = "wasm32")))]
79    async fn open_wt<T: DeserializeOwned + Send + 'static>(
80        &self,
81        target: &SubscriptionTarget,
82        auto: bool,
83    ) -> Result<ClientSubscription<T>, Failure> {
84        let Some(endpoint) = self.config.webtransport.as_ref() else {
85            return Err(Failure::Refused(ClientError::BadInput(
86                "WebTransport subscriptions need `ClientConfig::with_webtransport`".to_owned(),
87            )));
88        };
89        let connect = super::wt::connect(self, endpoint);
90        let session = if auto {
91            match tokio::time::timeout(AUTO_WT_WITHIN, connect).await {
92                Ok(session) => session,
93                Err(_) => Err(Failure::Unreachable(ClientError::InvalidResponse(
94                    "WebTransport: no session in time".to_owned(),
95                ))),
96            }
97        } else {
98            connect.await
99        };
100        let session = session.inspect_err(|failure| {
101            if matches!(failure, Failure::Unreachable(_)) {
102                endpoint.mark_failed();
103            }
104        })?;
105        super::wt::open(session, target).await
106    }
107
108    #[cfg(not(all(feature = "webtransport", not(target_arch = "wasm32"))))]
109    async fn open_wt<T: DeserializeOwned + Send + 'static>(
110        &self,
111        _target: &SubscriptionTarget,
112        _auto: bool,
113    ) -> Result<ClientSubscription<T>, Failure> {
114        Err(Failure::Refused(ClientError::BadInput(
115            "WebTransport subscriptions need the `webtransport` feature on a native target"
116                .to_owned(),
117        )))
118    }
119
120    #[cfg(all(feature = "ws", not(target_arch = "wasm32")))]
121    async fn open_ws<T: DeserializeOwned + Send + 'static>(
122        &self,
123        target: &SubscriptionTarget,
124    ) -> Result<ClientSubscription<T>, Failure> {
125        super::ws::open(self, target).await
126    }
127
128    #[cfg(not(all(feature = "ws", not(target_arch = "wasm32"))))]
129    async fn open_ws<T: DeserializeOwned + Send + 'static>(
130        &self,
131        _target: &SubscriptionTarget,
132    ) -> Result<ClientSubscription<T>, Failure> {
133        Err(Failure::Refused(ClientError::BadInput(
134            "WebSocket subscriptions need the `ws` feature on a native target".to_owned(),
135        )))
136    }
137}
cratestack_client_rust/subscription/open.rs

cratestack_client_rust/subscription/
open.rs
