im_core/realtime/
service.rs

1pub struct RealtimeService<'a> {
2    client: &'a crate::core::ImClient,
3}
4
5impl<'a> RealtimeService<'a> {
6    pub(crate) fn new(client: &'a crate::core::ImClient) -> Self {
7        Self { client }
8    }
9
10    pub fn status(&self) -> crate::ImResult<super::RealtimeStatus> {
11        Ok(super::RealtimeStatus {
12            connected: false,
13            state: super::RealtimeConnectionState::Disconnected,
14            subscriptions: Vec::new(),
15            last_error: None,
16        })
17    }
18
19    #[cfg(feature = "blocking")]
20    pub fn connect(
21        &self,
22        options: super::RealtimeOptions,
23    ) -> crate::ImResult<super::RealtimeHandle> {
24        validate_options(&options)?;
25        if self.client.core_inner().sdk_config().transport_policy
26            == crate::config::MessageTransportPolicy::HttpOnly
27        {
28            return Err(crate::ImError::unsupported("realtime-runner"));
29        }
30        super::runner::spawn_default(self.client.clone(), options)
31    }
32
33    pub async fn start_async(
34        &self,
35        options: super::RealtimeOptions,
36    ) -> crate::ImResult<super::RealtimeSession> {
37        validate_options(&options)?;
38        if self.client.core_inner().sdk_config().transport_policy
39            == crate::config::MessageTransportPolicy::HttpOnly
40        {
41            return Err(crate::ImError::unsupported("realtime-runner"));
42        }
43        super::runner::spawn_default_async(self.client.clone(), options).await
44    }
45
46    #[cfg(feature = "blocking")]
47    pub fn run_until_shutdown(
48        &self,
49        options: super::RealtimeOptions,
50        shutdown: super::ShutdownSignal,
51    ) -> crate::ImResult<super::RealtimeExit> {
52        validate_options(&options)?;
53        if shutdown.is_requested() {
54            return Ok(super::RealtimeExit {
55                reason: super::RealtimeExitReason::ShutdownRequested,
56                reconnect_attempts: 0,
57                warnings: Vec::new(),
58            });
59        }
60        if self.client.core_inner().sdk_config().transport_policy
61            == crate::config::MessageTransportPolicy::HttpOnly
62        {
63            return Ok(super::RealtimeExit {
64                reason: super::RealtimeExitReason::TransportUnavailable,
65                reconnect_attempts: 0,
66                warnings: vec!["unsupported capability: realtime-runner".to_string()],
67            });
68        }
69        super::runner::run_default_until_shutdown(self.client, options, shutdown)
70    }
71
72    #[cfg(feature = "blocking")]
73    pub fn run_until_shutdown_with_event_sink<S>(
74        &self,
75        options: super::RealtimeOptions,
76        shutdown: super::ShutdownSignal,
77        event_sink: &mut S,
78    ) -> crate::ImResult<super::RealtimeExit>
79    where
80        S: super::RealtimeRunnerEventSink,
81    {
82        validate_options(&options)?;
83        if shutdown.is_requested() {
84            return Ok(super::RealtimeExit {
85                reason: super::RealtimeExitReason::ShutdownRequested,
86                reconnect_attempts: 0,
87                warnings: Vec::new(),
88            });
89        }
90        if self.client.core_inner().sdk_config().transport_policy
91            == crate::config::MessageTransportPolicy::HttpOnly
92        {
93            return Ok(super::RealtimeExit {
94                reason: super::RealtimeExitReason::TransportUnavailable,
95                reconnect_attempts: 0,
96                warnings: vec!["unsupported capability: realtime-runner".to_string()],
97            });
98        }
99        super::runner::run_default_with_event_sink_until_shutdown(
100            self.client,
101            options,
102            shutdown,
103            event_sink,
104        )
105    }
106}
107
108fn validate_options(options: &super::RealtimeOptions) -> crate::ImResult<()> {
109    if options.event_buffer == 0 {
110        return Err(crate::ImError::invalid_input(
111            Some("event_buffer".to_string()),
112            "event buffer must be greater than zero",
113        ));
114    }
115    validate_reconnect_policy(&options.reconnect)
116}
117
118fn validate_reconnect_policy(policy: &super::ReconnectPolicy) -> crate::ImResult<()> {
119    match policy {
120        super::ReconnectPolicy::Disabled => Ok(()),
121        super::ReconnectPolicy::Fixed { delay_ms, .. } if *delay_ms == 0 => {
122            Err(crate::ImError::invalid_input(
123                Some("reconnect.delay_ms".to_string()),
124                "fixed reconnect delay must be greater than zero",
125            ))
126        }
127        super::ReconnectPolicy::Fixed { .. } => Ok(()),
128        super::ReconnectPolicy::Exponential {
129            base_delay_ms,
130            max_delay_ms,
131            ..
132        } if *base_delay_ms == 0 || *max_delay_ms == 0 => Err(crate::ImError::invalid_input(
133            Some("reconnect".to_string()),
134            "exponential reconnect delays must be greater than zero",
135        )),
136        super::ReconnectPolicy::Exponential {
137            base_delay_ms,
138            max_delay_ms,
139            ..
140        } if base_delay_ms > max_delay_ms => Err(crate::ImError::invalid_input(
141            Some("reconnect".to_string()),
142            "base reconnect delay must not exceed max delay",
143        )),
144        super::ReconnectPolicy::Exponential { .. } => Ok(()),
145    }
146}
im_core/realtime/service.rs

im_core/realtime/
service.rs
