a3s_boot/websocket/
message.rs

1use crate::{validate_value, BootError, BoxFuture, Result, Validate};
2use serde::de::DeserializeOwned;
3use serde::{Deserialize, Serialize};
4use serde_json::Value;
5use std::future::Future;
6use std::sync::Arc;
7
8/// Adapter-neutral WebSocket message used by gateways and adapters.
9#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
10pub struct WebSocketMessage {
11    pub event: String,
12    #[serde(default)]
13    pub data: Value,
14}
15
16impl WebSocketMessage {
17    pub fn new(event: impl Into<String>, data: impl Into<Value>) -> Self {
18        Self {
19            event: event.into(),
20            data: data.into(),
21        }
22    }
23
24    pub fn event(&self) -> &str {
25        &self.event
26    }
27
28    pub fn data(&self) -> &Value {
29        &self.data
30    }
31
32    pub fn data_as<T>(&self) -> Result<T>
33    where
34        T: DeserializeOwned,
35    {
36        serde_json::from_value(self.data.clone())
37            .map_err(|err| BootError::BadRequest(err.to_string()))
38    }
39
40    pub fn data_field(&self, name: &str) -> Result<Option<Value>> {
41        let Value::Object(fields) = &self.data else {
42            return Err(BootError::BadRequest(
43                "expected JSON object websocket data".to_string(),
44            ));
45        };
46
47        Ok(fields.get(name).filter(|value| !value.is_null()).cloned())
48    }
49
50    pub fn data_field_as<T>(&self, name: &str) -> Result<T>
51    where
52        T: DeserializeOwned,
53    {
54        let Some(value) = self.data_field(name)? else {
55            return Err(BootError::BadRequest(format!(
56                "missing websocket data field: {name}"
57            )));
58        };
59        deserialize_data_field("websocket data field", name, value)
60    }
61
62    pub fn optional_data_field_as<T>(&self, name: &str) -> Result<Option<T>>
63    where
64        T: DeserializeOwned,
65    {
66        self.data_field(name)?
67            .map(|value| deserialize_data_field("websocket data field", name, value))
68            .transpose()
69    }
70
71    pub fn data_field_string(&self, name: &str) -> Result<String> {
72        let Some(value) = self.data_field(name)? else {
73            return Err(BootError::BadRequest(format!(
74                "missing websocket data field: {name}"
75            )));
76        };
77        data_field_value_to_string(value)
78    }
79
80    pub fn optional_data_field_string(&self, name: &str) -> Result<Option<String>> {
81        self.data_field(name)?
82            .map(data_field_value_to_string)
83            .transpose()
84    }
85
86    pub fn validated_data<T>(&self) -> Result<T>
87    where
88        T: DeserializeOwned + Validate,
89    {
90        validate_value(self.data_as::<T>()?)
91    }
92
93    pub fn text(event: impl Into<String>, data: impl Into<String>) -> Self {
94        Self::new(event, Value::String(data.into()))
95    }
96
97    pub fn json<T>(event: impl Into<String>, data: &T) -> Result<Self>
98    where
99        T: Serialize,
100    {
101        Ok(Self::new(
102            event,
103            serde_json::to_value(data).map_err(|err| BootError::Internal(err.to_string()))?,
104        ))
105    }
106}
107
108fn deserialize_data_field<T>(label: &str, name: &str, value: Value) -> Result<T>
109where
110    T: DeserializeOwned,
111{
112    serde_json::from_value(value)
113        .map_err(|error| BootError::BadRequest(format!("invalid {label} {name}: {error}")))
114}
115
116fn data_field_value_to_string(value: Value) -> Result<String> {
117    match value {
118        Value::String(value) => Ok(value),
119        Value::Bool(value) => Ok(value.to_string()),
120        Value::Number(value) => Ok(value.to_string()),
121        Value::Array(_) | Value::Object(_) => {
122            serde_json::to_string(&value).map_err(|error| BootError::BadRequest(error.to_string()))
123        }
124        Value::Null => Ok("null".to_string()),
125    }
126}
127
128/// Return value accepted by WebSocket gateway handlers.
129pub trait IntoWebSocketReply {
130    fn into_websocket_reply(self) -> Option<WebSocketMessage>;
131}
132
133impl IntoWebSocketReply for WebSocketMessage {
134    fn into_websocket_reply(self) -> Option<WebSocketMessage> {
135        Some(self)
136    }
137}
138
139impl IntoWebSocketReply for Option<WebSocketMessage> {
140    fn into_websocket_reply(self) -> Option<WebSocketMessage> {
141        self
142    }
143}
144
145impl IntoWebSocketReply for () {
146    fn into_websocket_reply(self) -> Option<WebSocketMessage> {
147        None
148    }
149}
150
151/// Outbound writer for adapter-backed WebSocket connections.
152pub trait WebSocketOutbound: Send + Sync + 'static {
153    fn send(&self, message: WebSocketMessage) -> BoxFuture<'static, Result<()>>;
154}
155
156impl<F, Fut> WebSocketOutbound for F
157where
158    F: Fn(WebSocketMessage) -> Fut + Send + Sync + 'static,
159    Fut: Future<Output = Result<()>> + Send + 'static,
160{
161    fn send(&self, message: WebSocketMessage) -> BoxFuture<'static, Result<()>> {
162        Box::pin(self(message))
163    }
164}
165
166impl WebSocketOutbound for Arc<dyn WebSocketOutbound> {
167    fn send(&self, message: WebSocketMessage) -> BoxFuture<'static, Result<()>> {
168        self.as_ref().send(message)
169    }
170}
171
172pub(crate) async fn send_to_outbounds(
173    outbounds: Vec<Arc<dyn WebSocketOutbound>>,
174    message: WebSocketMessage,
175) -> Result<usize> {
176    let mut sent = 0;
177    for outbound in outbounds {
178        outbound.send(message.clone()).await?;
179        sent += 1;
180    }
181    Ok(sent)
182}
a3s_boot/websocket/message.rs

a3s_boot/websocket/
message.rs
