asterisk_rs_ari/
transport.rs

1//! REST transport abstraction for HTTP and WebSocket modes.
2
3use std::collections::HashMap;
4use std::sync::Arc;
5use std::sync::atomic::{AtomicU8, Ordering};
6use std::time::Duration;
7
8use tokio::sync::{oneshot, watch};
9use tokio::time::Instant;
10
11use crate::config::AriConfig;
12use crate::error::{AriError, HttpError, Result};
13use crate::event::{AriEvent, AriMessage};
14use crate::websocket::WsEventListener;
15use crate::ws_transport::WsTransport;
16use asterisk_rs_core::auth::Credentials;
17use asterisk_rs_core::event::EventBus;
18
19const REQUEST_QUEUED: u8 = 0;
20const REQUEST_WRITING: u8 = 1;
21const REQUEST_WRITTEN: u8 = 2;
22const REQUEST_CANCELLED: u8 = 3;
23static HTTP_REQUEST_COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
24pub(crate) const REST_COMMAND_CAPACITY: usize = 64;
25
26pub(crate) struct RestCommand {
27    pub request_id: String,
28    pub method: String,
29    pub uri: String,
30    pub content_type: Option<String>,
31    pub message_body: Option<String>,
32    pub deadline: Instant,
33    pub lifecycle: Arc<RequestLifecycle>,
34    pub response_tx: oneshot::Sender<Result<TransportResponse>>,
35}
36
37pub(crate) struct PendingResponse {
38    pub deadline: Instant,
39    pub request_id: String,
40    pub method: String,
41    pub uri: String,
42    pub lifecycle: Arc<RequestLifecycle>,
43    pub response_tx: oneshot::Sender<Result<TransportResponse>>,
44}
45
46pub(crate) fn fail_pending(pending: &mut HashMap<String, PendingResponse>, details: &str) {
47    for (_, response) in pending.drain() {
48        let error = write_error(
49            &response.method,
50            &response.uri,
51            &response.request_id,
52            &response.lifecycle,
53            || details.to_owned(),
54        );
55        let _ = response.response_tx.send(Err(error));
56    }
57}
58
59pub(crate) fn purge_expired(pending: &mut HashMap<String, PendingResponse>) {
60    let now = Instant::now();
61    let expired: Vec<_> = pending
62        .iter()
63        .filter(|(_, response)| response.response_tx.is_closed() || response.deadline <= now)
64        .map(|(request_id, _)| request_id.clone())
65        .collect();
66    for request_id in expired {
67        let Some(response) = pending.remove(&request_id) else {
68            continue;
69        };
70        if response.response_tx.is_closed() {
71            tracing::debug!(%request_id, "discarding cancelled REST response correlation");
72            continue;
73        }
74        let error = deadline_error(
75            &response.method,
76            &response.uri,
77            &response.request_id,
78            &response.lifecycle,
79        );
80        let _ = response.response_tx.send(Err(error));
81    }
82}
83
84pub(crate) fn route_text_message(
85    text: &str,
86    event_bus: &EventBus<AriMessage>,
87    pending: &mut HashMap<String, PendingResponse>,
88    max_response_body_bytes: usize,
89) {
90    match serde_json::from_str::<AriMessage>(text) {
91        Ok(msg) => {
92            if let AriEvent::RESTResponse {
93                ref request_id,
94                status_code,
95                ref reason_phrase,
96                ref message_body,
97                ..
98            } = msg.event
99            {
100                if let Some(response) = pending.remove(request_id) {
101                    if let Some(body) = message_body {
102                        if body.len() > max_response_body_bytes {
103                            let _ = response.response_tx.send(Err(AriError::ResponseTooLarge {
104                                limit: max_response_body_bytes,
105                                received: u64::try_from(body.len()).unwrap_or(u64::MAX),
106                            }));
107                            return;
108                        }
109                    }
110                    let result = u16::try_from(status_code)
111                        .map_err(|_| {
112                            AriError::WebSocket(format!(
113                                "invalid REST response status code: {status_code}"
114                            ))
115                        })
116                        .and_then(|status| {
117                            TransportResponse {
118                                status,
119                                body: message_body.clone(),
120                            }
121                            .require_success_with_fallback(
122                                (!reason_phrase.is_empty()).then(|| reason_phrase.clone()),
123                            )
124                        });
125                    let _ = response.response_tx.send(result);
126                }
127            } else {
128                event_bus.publish(msg);
129            }
130        }
131        Err(_) => tracing::warn!(
132            payload_bytes = text.len(),
133            "failed to deserialize ARI message"
134        ),
135    }
136}
137
138/// Observable lifecycle of the ARI event WebSocket.
139#[derive(Debug, Clone, PartialEq, Eq)]
140#[non_exhaustive]
141pub enum AriConnectionState {
142    Connecting,
143    Ready,
144    Reconnecting,
145    Terminal { details: String },
146    Disconnected,
147}
148
149pub(crate) async fn wait_until_ready(
150    state: &mut watch::Receiver<AriConnectionState>,
151    deadline: tokio::time::Instant,
152) -> Result<()> {
153    tokio::time::timeout_at(deadline, async {
154        loop {
155            match state.borrow().clone() {
156                AriConnectionState::Ready => return Ok(()),
157                AriConnectionState::Terminal { details } => {
158                    return Err(AriError::WebSocket(details));
159                }
160                AriConnectionState::Disconnected => return Err(AriError::Disconnected),
161                AriConnectionState::Connecting | AriConnectionState::Reconnecting => {}
162            }
163            state.changed().await.map_err(|_| AriError::Disconnected)?;
164        }
165    })
166    .await
167    .map_err(|_| AriError::WebSocket("initial websocket readiness timed out".to_owned()))?
168}
169
170/// shared state that makes caller timeout and actor write admission atomic
171#[derive(Debug, Default)]
172pub(crate) struct RequestLifecycle {
173    state: AtomicU8,
174}
175
176impl RequestLifecycle {
177    /// mark the exact first poll of the sink write future
178    fn begin_wire_poll(&self) -> bool {
179        self.state
180            .compare_exchange(
181                REQUEST_QUEUED,
182                REQUEST_WRITING,
183                Ordering::AcqRel,
184                Ordering::Acquire,
185            )
186            .is_ok()
187    }
188
189    pub(crate) fn mark_written(&self) {
190        self.state.store(REQUEST_WRITTEN, Ordering::Release);
191    }
192
193    /// cancel while the request is still definitely absent from the wire
194    pub(crate) fn cancel_unsent(&self) -> bool {
195        loop {
196            let state = self.state.load(Ordering::Acquire);
197            if state != REQUEST_QUEUED {
198                return false;
199            }
200            if self
201                .state
202                .compare_exchange(
203                    state,
204                    REQUEST_CANCELLED,
205                    Ordering::AcqRel,
206                    Ordering::Acquire,
207                )
208                .is_ok()
209            {
210                return true;
211            }
212        }
213    }
214
215    pub(crate) fn may_have_executed(&self) -> bool {
216        matches!(
217            self.state.load(Ordering::Acquire),
218            REQUEST_WRITING | REQUEST_WRITTEN
219        )
220    }
221}
222
223/// Poll a sink write only after atomically crossing the may-have-written boundary.
224///
225/// `None` means the caller cancelled the request before the first sink poll, so
226/// no bytes from this request can have reached the wire.
227pub(crate) async fn poll_wire_write<F>(lifecycle: &RequestLifecycle, future: F) -> Option<F::Output>
228where
229    F: std::future::Future,
230{
231    let mut future = std::pin::pin!(future);
232    let mut first_poll = true;
233    std::future::poll_fn(|cx| {
234        if first_poll {
235            first_poll = false;
236            if !lifecycle.begin_wire_poll() {
237                return std::task::Poll::Ready(None);
238            }
239        }
240        future.as_mut().poll(cx).map(Some)
241    })
242    .await
243}
244
245pub(crate) fn is_mutating(method: &str) -> bool {
246    matches!(method, "POST" | "PUT" | "DELETE" | "PATCH")
247}
248
249pub(crate) fn deadline_error(
250    method: &str,
251    uri: &str,
252    request_id: &str,
253    lifecycle: &RequestLifecycle,
254) -> AriError {
255    if lifecycle.cancel_unsent() {
256        return AriError::RequestNotSent {
257            method: method.to_owned(),
258            uri: uri.to_owned(),
259        };
260    }
261    if is_mutating(method) && lifecycle.may_have_executed() {
262        return AriError::OutcomeUnknown {
263            request_id: request_id.to_owned(),
264            method: method.to_owned(),
265            uri: uri.to_owned(),
266        };
267    }
268
269    AriError::WebSocket(format!("{method} {uri} timed out"))
270}
271
272pub(crate) fn write_error(
273    method: &str,
274    uri: &str,
275    request_id: &str,
276    lifecycle: &RequestLifecycle,
277    details: impl FnOnce() -> String,
278) -> AriError {
279    if lifecycle.cancel_unsent() {
280        return AriError::RequestNotSent {
281            method: method.to_owned(),
282            uri: uri.to_owned(),
283        };
284    }
285    if is_mutating(method) {
286        AriError::OutcomeUnknown {
287            request_id: request_id.to_owned(),
288            method: method.to_owned(),
289            uri: uri.to_owned(),
290        }
291    } else {
292        AriError::WebSocket(details())
293    }
294}
295
296pub(crate) fn outbound_message_limit_error(
297    method: &str,
298    uri: &str,
299    message_bytes: usize,
300    limit: usize,
301) -> Option<AriError> {
302    if message_bytes <= limit {
303        return None;
304    }
305
306    if is_mutating(method) {
307        Some(AriError::RequestNotSent {
308            method: method.to_owned(),
309            uri: uri.to_owned(),
310        })
311    } else {
312        Some(AriError::WebSocket(format!(
313            "serialized {method} {uri} request is {message_bytes} bytes, exceeding the websocket message limit of {limit} bytes"
314        )))
315    }
316}
317
318/// response from a transport REST operation
319pub(crate) struct TransportResponse {
320    pub status: u16,
321    pub body: Option<String>,
322}
323
324impl TransportResponse {
325    pub(crate) fn require_success(self) -> Result<Self> {
326        self.require_success_with_fallback(None)
327    }
328
329    pub(crate) fn require_success_with_fallback(self, fallback: Option<String>) -> Result<Self> {
330        if (200..300).contains(&self.status) {
331            return Ok(self);
332        }
333
334        let Self { status, body } = self;
335        let message = body
336            .or(fallback)
337            .unwrap_or_else(|| format!("HTTP {status}"));
338        Err(AriError::Api { status, message })
339    }
340}
341
342/// internal transport implementation — dispatches REST calls to either
343/// HTTP (reqwest) or a unified WebSocket connection
344pub(crate) enum TransportInner {
345    Http(HttpTransport),
346    WebSocket(WsTransport),
347}
348
349impl TransportInner {
350    pub(crate) fn connection_state(&self) -> AriConnectionState {
351        match self {
352            Self::Http(t) => t.ws_listener.connection_state(),
353            Self::WebSocket(t) => t.connection_state(),
354        }
355    }
356
357    pub(crate) fn subscribe_connection_state(&self) -> watch::Receiver<AriConnectionState> {
358        match self {
359            Self::Http(t) => t.ws_listener.subscribe_connection_state(),
360            Self::WebSocket(t) => t.subscribe_connection_state(),
361        }
362    }
363
364    pub(crate) async fn wait_ready(&self, timeout: Duration) -> Result<()> {
365        let deadline = tokio::time::Instant::now() + timeout;
366        let mut state = self.subscribe_connection_state();
367        wait_until_ready(&mut state, deadline).await
368    }
369    pub(crate) async fn request(
370        &self,
371        method: &str,
372        path: &str,
373        body: Option<String>,
374    ) -> Result<TransportResponse> {
375        match self {
376            Self::Http(t) => t.request(method, path, body).await,
377            Self::WebSocket(t) => t.request(method, path, body).await,
378        }
379    }
380
381    pub(crate) fn shutdown(&self) {
382        match self {
383            Self::Http(t) => t.ws_listener.abort(),
384            Self::WebSocket(t) => t.abort(),
385        }
386    }
387
388    pub(crate) async fn shutdown_and_wait(&self) {
389        match self {
390            Self::Http(t) => t.ws_listener.shutdown_and_wait().await,
391            Self::WebSocket(t) => t.shutdown_and_wait().await,
392        }
393    }
394}
395
396/// HTTP-based transport — uses reqwest for REST and a separate
397/// WebSocket listener for events
398pub(crate) struct HttpTransport {
399    client: reqwest::Client,
400    base_url: String,
401    credentials: Credentials,
402    ws_listener: WsEventListener,
403    max_response_body_bytes: usize,
404}
405
406impl HttpTransport {
407    pub(crate) fn new(config: &AriConfig, event_bus: EventBus<AriMessage>) -> Result<Self> {
408        let mut client_builder = reqwest::Client::builder()
409            .connect_timeout(Duration::from_secs(10))
410            .timeout(config.request_timeout())
411            .redirect(reqwest::redirect::Policy::none());
412        for root in &config.tls_trust.reqwest_roots {
413            client_builder = client_builder.add_root_certificate(root.clone());
414        }
415        let client = client_builder
416            .build()
417            .map_err(|error| AriError::Http(HttpError::new(error)))?;
418
419        let ws_listener = WsEventListener::spawn(
420            config.ws_url(),
421            event_bus,
422            config.reconnect_policy().clone(),
423            config.max_websocket_message_bytes(),
424            &config.tls_trust.rustls_roots,
425        )?;
426
427        Ok(Self {
428            client,
429            base_url: config.base_url().as_str().trim_end_matches('/').to_owned(),
430            credentials: config.credentials().clone(),
431            ws_listener,
432            max_response_body_bytes: config.max_response_body_bytes(),
433        })
434    }
435
436    pub(crate) async fn request(
437        &self,
438        method: &str,
439        path: &str,
440        body: Option<String>,
441    ) -> Result<TransportResponse> {
442        let request_id = format!(
443            "http-{}",
444            HTTP_REQUEST_COUNTER.fetch_add(1, Ordering::Relaxed)
445        );
446        let url = format!("{}/{}", self.base_url, path.trim_start_matches('/'));
447        let http_method = parse_method(method)?;
448
449        let mut req = self
450            .client
451            .request(http_method, &url)
452            .basic_auth(self.credentials.username(), Some(self.credentials.secret()));
453
454        if let Some(json_body) = body {
455            req = req
456                .header(reqwest::header::CONTENT_TYPE, "application/json")
457                .body(json_body);
458        }
459
460        let response = match req.send().await {
461            Ok(response) => response,
462            Err(error) if is_mutating(method) && error.is_connect() => {
463                return Err(AriError::RequestNotSent {
464                    method: method.to_owned(),
465                    uri: path.to_owned(),
466                });
467            }
468            Err(_error) if is_mutating(method) => {
469                return Err(AriError::OutcomeUnknown {
470                    request_id,
471                    method: method.to_owned(),
472                    uri: path.to_owned(),
473                });
474            }
475            Err(error) => return Err(AriError::Http(HttpError::new(error))),
476        };
477        let status = response.status().as_u16();
478        let body = read_response_body(response, self.max_response_body_bytes).await?;
479        TransportResponse { status, body }.require_success()
480    }
481}
482
483async fn read_response_body(
484    mut response: reqwest::Response,
485    limit: usize,
486) -> Result<Option<String>> {
487    if let Some(content_length) = response.content_length() {
488        let limit_u64 = u64::try_from(limit).unwrap_or(u64::MAX);
489        if content_length > limit_u64 {
490            return Err(AriError::ResponseTooLarge {
491                limit,
492                received: content_length,
493            });
494        }
495    }
496
497    let capacity = response
498        .content_length()
499        .and_then(|length| usize::try_from(length).ok())
500        .unwrap_or(0)
501        .min(limit);
502    let mut body = Vec::with_capacity(capacity);
503    while let Some(chunk) = response
504        .chunk()
505        .await
506        .map_err(|error| AriError::Http(HttpError::new(error)))?
507    {
508        let received = body.len().saturating_add(chunk.len());
509        if received > limit {
510            return Err(AriError::ResponseTooLarge {
511                limit,
512                received: u64::try_from(received).unwrap_or(u64::MAX),
513            });
514        }
515        body.extend_from_slice(&chunk);
516    }
517
518    if body.is_empty() {
519        Ok(None)
520    } else {
521        Ok(Some(String::from_utf8_lossy(&body).into_owned()))
522    }
523}
524
525fn parse_method(method: &str) -> Result<reqwest::Method> {
526    match method {
527        "GET" => Ok(reqwest::Method::GET),
528        "POST" => Ok(reqwest::Method::POST),
529        "PUT" => Ok(reqwest::Method::PUT),
530        "DELETE" => Ok(reqwest::Method::DELETE),
531        other => Err(AriError::WebSocket(format!(
532            "unsupported HTTP method: {other}"
533        ))),
534    }
535}
asterisk_rs_ari/transport.rs

asterisk_rs_ari/
transport.rs
