cranpose_services/
host_messages.rs

1//! Messages between the application and the program embedding it.
2//!
3//! An application launched inside another program — an IDE tool window, an
4//! editor panel — shares a message line with that host. Every message names a
5//! channel and carries a text payload; which channels exist and what their
6//! payloads hold is for the host and the application to agree on (JSON is the
7//! usual choice). The newest message on each channel is kept, so a screen that
8//! starts collecting after the host spoke still learns the current value.
9//!
10//! Outside a host nothing is ever received and [`send_to_host`] reports that
11//! no one listened, so the same application runs standalone unchanged.
12
13use std::sync::{Arc, Mutex, MutexGuard, OnceLock, PoisonError};
14
15use cranpose_core::{EventStream, ProcessIds, collections::map::HashMap, rememberEventStream};
16
17/// One message on the line between the application and its host.
18#[derive(Clone, Debug, PartialEq, Eq)]
19pub struct HostMessage {
20    /// The channel the message belongs to.
21    pub channel: String,
22    /// The message body.
23    pub payload: String,
24}
25
26impl HostMessage {
27    /// A message carrying `payload` on `channel`.
28    pub fn new(channel: impl Into<String>, payload: impl Into<String>) -> Self {
29        Self {
30            channel: channel.into(),
31            payload: payload.into(),
32        }
33    }
34}
35
36type Observer = Arc<dyn Fn(String) + Send + Sync>;
37type Outbox = Arc<dyn Fn(HostMessage) + Send + Sync>;
38
39struct Mailbox {
40    observers: Vec<(u64, String, Observer)>,
41    latest: HashMap<String, String>,
42}
43
44impl Mailbox {
45    fn new() -> Self {
46        Self {
47            observers: Vec::new(),
48            latest: HashMap::default(),
49        }
50    }
51
52    fn observe(&mut self, id: u64, channel: &str, observer: Observer) -> Option<String> {
53        self.observers.push((id, channel.to_owned(), observer));
54        self.latest.get(channel).cloned()
55    }
56
57    fn publish(&mut self, message: &HostMessage) -> Vec<Observer> {
58        self.latest
59            .insert(message.channel.clone(), message.payload.clone());
60        self.observers
61            .iter()
62            .filter(|(_, channel, _)| *channel == message.channel)
63            .map(|(_, _, observer)| Arc::clone(observer))
64            .collect()
65    }
66
67    fn remove_observer(&mut self, id: u64) {
68        self.observers.retain(|(existing, _, _)| *existing != id);
69    }
70
71    fn clear(&mut self) {
72        self.latest.clear();
73    }
74}
75
76fn mailbox() -> MutexGuard<'static, Mailbox> {
77    static MAILBOX: OnceLock<Mutex<Mailbox>> = OnceLock::new();
78    MAILBOX
79        .get_or_init(|| Mutex::new(Mailbox::new()))
80        .lock()
81        .unwrap_or_else(PoisonError::into_inner)
82}
83
84fn outbox() -> MutexGuard<'static, Option<Outbox>> {
85    static OUTBOX: OnceLock<Mutex<Option<Outbox>>> = OnceLock::new();
86    OUTBOX
87        .get_or_init(|| Mutex::new(None))
88        .lock()
89        .unwrap_or_else(PoisonError::into_inner)
90}
91
92static NEXT_ID: ProcessIds = ProcessIds::starting_at(1);
93
94/// Keeps a host message observer registered until it is dropped.
95pub struct HostMessageObserver {
96    id: u64,
97}
98
99impl Drop for HostMessageObserver {
100    fn drop(&mut self) {
101        mailbox().remove_observer(self.id);
102    }
103}
104
105/// Registers `observer` for the messages the host sends on `channel`, handing
106/// it the newest one at once when the host already spoke there.
107///
108/// Applications collect [`rememberHostMessages`] instead of calling this.
109pub fn observe_host_messages(
110    channel: &str,
111    observer: impl Fn(String) + Send + Sync + 'static,
112) -> HostMessageObserver {
113    let id = NEXT_ID.next();
114    let observer: Observer = Arc::new(observer);
115    let replay = mailbox().observe(id, channel, Arc::clone(&observer));
116    if let Some(payload) = replay {
117        observer(payload);
118    }
119    HostMessageObserver { id }
120}
121
122/// Delivers a message the host sent to every observer of its channel.
123/// Callable from any thread; platform hosts call it as messages arrive.
124pub fn publish_host_message(message: HostMessage) {
125    let observers = mailbox().publish(&message);
126    for observer in observers {
127        observer(message.payload.clone());
128    }
129}
130
131/// Forgets the newest message of every channel. Used by tests and host
132/// teardown.
133pub fn clear_host_messages() {
134    mailbox().clear();
135}
136
137/// Installs where [`send_to_host`] delivers: the platform host's connection.
138pub fn install_host_outbox(outbox_fn: impl Fn(HostMessage) + Send + Sync + 'static) {
139    *outbox() = Some(Arc::new(outbox_fn));
140}
141
142/// Removes the installed outbox, after which [`send_to_host`] reports that no
143/// host listens.
144pub fn clear_host_outbox() {
145    *outbox() = None;
146}
147
148/// Sends `payload` on `channel` to the embedding host.
149///
150/// Returns `false` when the application runs without a host, so nothing was
151/// sent. Callable from any thread.
152pub fn send_to_host(channel: &str, payload: &str) -> bool {
153    let Some(outbox_fn) = outbox().clone() else {
154        log::debug!("host message on {channel} dropped: no host is attached");
155        return false;
156    };
157    outbox_fn(HostMessage::new(channel, payload));
158    true
159}
160
161/// Collects the messages the host sends on `channel` for as long as this call
162/// stays in the composition, starting with the newest one already received.
163///
164/// ```rust,no_run
165/// use cranpose_macros::composable;
166/// use cranpose_services::rememberHostMessages;
167///
168/// #[composable]
169/// fn CurrentFile() {
170///     let file =
171///         cranpose_core::collectAsState(rememberHostMessages("ide.editor"), (), String::new());
172///     log::info!("the host shows {}", file.get());
173/// }
174/// ```
175#[expect(non_snake_case, reason = "names follow Compose's casing")]
176#[track_caller]
177pub fn rememberHostMessages(channel: &str) -> EventStream<String> {
178    let channel = channel.to_owned();
179    rememberEventStream(channel.clone(), move |sender| {
180        observe_host_messages(&channel, move |payload| sender.send(payload))
181    })
182}
183
184#[cfg(test)]
185#[path = "tests/host_messages_tests.rs"]
186mod tests;
cranpose_services/host_messages.rs

cranpose_services/
host_messages.rs
