arora_websocket/
bridge.rs

1//! [`WsBridge`]: the Vizij WebSocket server driven as an Arora
2//! [`Bridge`](arora_bridge::Bridge) (Phase 5C).
3//!
4//! The WS server is a parallel reimplementation of `arora-bridge`. This adapter
5//! folds it onto the real thing: each incoming protocol message becomes a
6//! [`BridgeCommand`] on [`commands`](Bridge::commands), and the consumer — the
7//! Arora runtime — reacts to the ones it cares about (value-updates apply to the
8//! store, reads reply). Runtime state flows back out through
9//! [`send_data`](Bridge::send_data) as a `slot_values_changed` push.
10//!
11//! The server's existing handler API is kept and *built on top of* the command
12//! stream: the handlers registered here simply translate a message into a
13//! command. After Phase 5B the value vocabulary is `arora_types::Value`, so
14//! the translation is structural, not a conversion.
15
16use std::collections::HashMap;
17use std::sync::{Arc, Mutex};
18
19use arora_bridge::{
20    Bridge, BridgeCommand, BridgeOp, BridgeResult, CommandStream, DataRequestedStream, DeviceInfo,
21    DeviceInfoStream,
22};
23use arora_types::data::{Key, StateChange};
24use arora_types::value::Value;
25use async_trait::async_trait;
26use futures::channel::{mpsc, oneshot};
27
28use crate::messages::Outgoing;
29use crate::server::AroraWSServer;
30
31/// The Vizij WebSocket server as an Arora [`Bridge`].
32///
33/// Note: the server's `validate_paths` (on by default) checks incoming slot
34/// paths against its `Registry`, which this bridge does not populate — either
35/// mirror the runtime's keys into the registry or disable `validate_paths`
36/// when serving purely through the bridge.
37pub struct WsBridge {
38    server: Arc<AroraWSServer>,
39    /// The receiving half of the command stream, handed out once by [`commands`].
40    commands: Mutex<Option<mpsc::UnboundedReceiver<BridgeCommand>>>,
41}
42
43impl WsBridge {
44    /// Wrap a server: register handlers that turn each incoming message into a
45    /// [`BridgeCommand`] on the [`commands`](Bridge::commands) stream.
46    pub async fn new(server: Arc<AroraWSServer>) -> Self {
47        let (cmd_tx, cmd_rx) = mpsc::unbounded::<BridgeCommand>();
48
49        // SetSlotValues -> Update. The handler is synchronous, so enqueue the
50        // command and acknowledge; the runtime applies it on its next step.
51        let tx = cmd_tx.clone();
52        server
53            .set_set_slot_values_handler(move |values: HashMap<String, Value>| {
54                let mut change = StateChange::new();
55                for (path, value) in values {
56                    change.set.insert(Key::from(path), Some(value));
57                }
58                let (reply_tx, _reply_rx) = oneshot::channel();
59                tx.unbounded_send(BridgeCommand::new(BridgeOp::Update(change), reply_tx))
60                    .map_err(|_| "bridge command channel closed".to_string())
61            })
62            .await;
63
64        // GetSlotValues -> Get, awaiting the runtime's reply.
65        let tx = cmd_tx.clone();
66        server
67            .set_get_slot_values_handler(Arc::new(move |slots: Vec<String>| {
68                let tx = tx.clone();
69                Box::pin(async move {
70                    let keys: Vec<Key> = slots.iter().cloned().map(Key::from).collect();
71                    let (reply_tx, reply_rx) = oneshot::channel();
72                    if tx
73                        .unbounded_send(BridgeCommand::new(BridgeOp::Get(keys), reply_tx))
74                        .is_err()
75                    {
76                        return HashMap::new();
77                    }
78                    match reply_rx.await {
79                        Ok(Ok(result)) => values_from_get(&slots, result.ret),
80                        _ => HashMap::new(),
81                    }
82                }) as _
83            }))
84            .await;
85
86        Self {
87            server,
88            commands: Mutex::new(Some(cmd_rx)),
89        }
90    }
91}
92
93/// Decode a `Get` reply — an `ArrayValue` of `Option`s in request order — into a
94/// `path -> value` map.
95fn values_from_get(slots: &[String], ret: Value) -> HashMap<String, Value> {
96    let mut out = HashMap::new();
97    if let Value::ArrayValue(items) = ret {
98        for (slot, item) in slots.iter().zip(items) {
99            if let Value::Option(Some(value)) = item {
100                out.insert(slot.clone(), *value);
101            }
102        }
103    }
104    out
105}
106
107#[async_trait]
108impl Bridge for WsBridge {
109    async fn get_device_info(&self) -> BridgeResult<Option<DeviceInfo>> {
110        // Vizij has no device-registration concept.
111        Ok(None)
112    }
113
114    async fn device_info_updated(&self) -> BridgeResult<DeviceInfoStream> {
115        Ok(Box::pin(futures::stream::empty()))
116    }
117
118    async fn update_device_info(
119        &self,
120        info: Option<DeviceInfo>,
121    ) -> BridgeResult<Option<DeviceInfo>> {
122        Ok(info)
123    }
124
125    async fn data_requested(&self) -> DataRequestedStream {
126        // A connected editor is a data consumer; signal interest once.
127        Box::pin(futures::stream::once(async { true }))
128    }
129
130    async fn send_data(&self, data: StateChange) -> BridgeResult<()> {
131        // Push the changed slots to the connected client(s).
132        let mut values = HashMap::new();
133        for (key, value) in data.set {
134            if let Some(value) = value {
135                values.insert(key.path, value);
136            }
137        }
138        if !values.is_empty() {
139            self.server.push(Outgoing::SlotValuesChanged { values });
140        }
141        Ok(())
142    }
143
144    /// The command stream is single-use: the channel's receiving half is
145    /// handed out on the first call, and later calls get an empty stream —
146    /// with a loud warning, because a runtime silently losing its command
147    /// plane is the worst failure mode.
148    async fn commands(&self) -> CommandStream {
149        match self.commands.lock().unwrap().take() {
150            Some(rx) => Box::pin(rx),
151            None => {
152                log::warn!(
153                    "WsBridge::commands() called more than once; the command \
154                     stream was already taken, returning an empty stream"
155                );
156                Box::pin(futures::stream::empty())
157            }
158        }
159    }
160}
161
162#[cfg(test)]
163mod tests {
164    use super::*;
165    use crate::server::ServerConfig;
166
167    #[tokio::test]
168    async fn send_data_pushes_slot_values_changed() {
169        let server = Arc::new(AroraWSServer::new(ServerConfig::default()));
170        let bridge = WsBridge::new(server.clone()).await;
171        let mut rx = server.subscribe();
172
173        bridge
174            .send_data(StateChange::set("face/mouth", Value::F64(0.5)))
175            .await
176            .expect("send_data");
177
178        match rx.recv().await.expect("a push") {
179            Outgoing::SlotValuesChanged { values } => {
180                assert_eq!(values.get("face/mouth"), Some(&Value::F64(0.5)));
181            }
182            other => panic!("expected SlotValuesChanged, got {other:?}"),
183        }
184    }
185}
arora_websocket/bridge.rs

arora_websocket/
bridge.rs
