zng_view_api/
ipc.rs

1//! IPC types.
2
3use std::time::Duration;
4
5use crate::{AnyResult, Event, Request, Response};
6
7use parking_lot::Mutex;
8use zng_task::channel::{self, ChannelError, IpcReceiver, IpcSender};
9use zng_txt::Txt;
10
11type AppInitMsg = (
12    channel::IpcReceiver<Request>,
13    channel::IpcSender<Response>,
14    channel::IpcSender<Event>,
15);
16
17/// Call `new`, then spawn the view-process using the `name` then call `connect`.
18pub(crate) struct AppInit {
19    init_sender: channel::NamedIpcSender<AppInitMsg>,
20}
21impl AppInit {
22    pub fn new() -> Self {
23        AppInit {
24            init_sender: channel::NamedIpcSender::new().expect("failed to create init channel"),
25        }
26    }
27
28    /// Unique name for the view-process to find this channel.
29    pub fn name(&self) -> &str {
30        self.init_sender.name()
31    }
32
33    /// Tries to connect to the view-process and receive the actual channels.
34    pub fn connect(self) -> AnyResult<(RequestSender, ResponseReceiver, EventReceiver)> {
35        let mut init_sender = self
36            .init_sender
37            .connect_deadline_blocking(std::time::Duration::from_secs(crate::view_timeout()))?;
38
39        let (req_sender, req_recv) = channel::ipc_unbounded()?;
40        let (rsp_sender, rsp_recv) = channel::ipc_unbounded()?;
41        let (evt_sender, evt_recv) = channel::ipc_unbounded()?;
42        init_sender.send_blocking((req_recv, rsp_sender, evt_sender))?;
43        Ok((
44            RequestSender(Mutex::new(req_sender)),
45            ResponseReceiver(Mutex::new(rsp_recv)),
46            EventReceiver(Mutex::new(evt_recv)),
47        ))
48    }
49}
50
51/// Start the view-process server and waits for `(request, response, event)`.
52pub fn connect_view_process(ipc_sender_name: Txt) -> Result<ViewChannels, channel::ChannelError> {
53    let _s = tracing::trace_span!("connect_view_process").entered();
54
55    let mut init_recv = channel::IpcReceiver::<AppInitMsg>::connect(ipc_sender_name)?;
56
57    let (req_recv, rsp_sender, evt_sender) = init_recv.recv_deadline_blocking(std::time::Duration::from_secs(crate::view_timeout()))?;
58
59    Ok(ViewChannels {
60        request_receiver: RequestReceiver(Mutex::new(req_recv)),
61        response_sender: ResponseSender(Mutex::new(rsp_sender)),
62        event_sender: EventSender(Mutex::new(evt_sender)),
63    })
64}
65
66/// Channels that must be used for implementing a view-process.
67pub struct ViewChannels {
68    /// View implementers must receive requests from this channel, call [`Api::respond`] and then
69    /// return the response using the `response_sender`.
70    ///
71    /// [`Api::respond`]: crate::Api::respond
72    pub request_receiver: RequestReceiver,
73
74    /// View implementers must synchronously send one response per request received in `request_receiver`.
75    pub response_sender: ResponseSender,
76
77    /// View implements must send events using this channel. Events can be asynchronous.
78    pub event_sender: EventSender,
79}
80
81type IpcResult<T> = Result<T, ChannelError>;
82
83pub(crate) struct RequestSender(Mutex<IpcSender<Request>>);
84impl RequestSender {
85    pub fn send(&mut self, req: Request) -> IpcResult<()> {
86        let r = self.0.get_mut().send_blocking(req);
87        if let Err(e) = &r {
88            tracing::error!("request sender error, {e}");
89        }
90        r
91    }
92}
93
94/// Requests channel end-point.
95///
96/// View-process implementers must receive [`Request`], call [`Api::respond`] and then use a [`ResponseSender`]
97/// to send back the response.
98///
99/// [`Api::respond`]: crate::Api::respond
100pub struct RequestReceiver(Mutex<IpcReceiver<Request>>); // Mutex for Sync
101impl RequestReceiver {
102    /// Receive one [`Request`].
103    pub fn recv(&mut self) -> IpcResult<Request> {
104        let r = self.0.get_mut().recv_blocking();
105        if let Err(e) = &r {
106            tracing::error!("request receiver error, {e}");
107        }
108        r
109    }
110}
111
112/// Responses channel entry-point.
113///
114/// View-process implementers must send [`Response`] returned by [`Api::respond`] using this sender.
115///
116/// Requests are received using [`RequestReceiver`] a response must be send for each request, synchronously.
117///
118/// [`Api::respond`]: crate::Api::respond
119pub struct ResponseSender(Mutex<IpcSender<Response>>); // Mutex for Sync
120impl ResponseSender {
121    /// Send a response.
122    ///
123    /// # Panics
124    ///
125    /// If the `rsp` is not [`must_be_send`].
126    ///
127    /// [`must_be_send`]: Response::must_be_send
128    pub fn send(&mut self, rsp: Response) -> IpcResult<()> {
129        assert!(rsp.must_be_send());
130        let r = self.0.get_mut().send_blocking(rsp);
131        if let Err(e) = &r {
132            tracing::error!("response sender error, {e}");
133        }
134        r
135    }
136}
137pub(crate) struct ResponseReceiver(Mutex<IpcReceiver<Response>>);
138impl ResponseReceiver {
139    pub fn recv(&mut self) -> IpcResult<Response> {
140        let r = self.0.get_mut().recv_blocking();
141        if let Err(e) = &r {
142            tracing::error!("response receiver error, {e}");
143        }
144        r
145    }
146}
147
148/// Event channel entry-point.
149///
150/// View-process implementers must send [`Event`] messages using this sender. The events
151/// can be asynchronous, not related to the [`Api::respond`] calls.
152///
153/// [`Api::respond`]: crate::Api::respond
154pub struct EventSender(Mutex<IpcSender<Event>>);
155impl EventSender {
156    /// Send an event notification.
157    pub fn send(&mut self, ev: Event) -> IpcResult<()> {
158        let r = self.0.get_mut().send_blocking(ev);
159        if let Err(e) = &r {
160            tracing::error!("event sender error, {e}");
161        }
162        r
163    }
164}
165pub(crate) struct EventReceiver(Mutex<IpcReceiver<Event>>);
166impl EventReceiver {
167    pub fn recv(&mut self) -> IpcResult<Event> {
168        let r = self.0.get_mut().recv_blocking();
169        if let Err(e) = &r {
170            tracing::error!("event receiver error, {e}");
171        }
172        r
173    }
174
175    pub fn recv_timeout(&mut self, duration: Duration) -> IpcResult<Event> {
176        let r = self.0.get_mut().recv_deadline_blocking(duration);
177        if let Err(e) = &r {
178            match e {
179                ChannelError::Timeout => {}
180                e => tracing::error!("event receiver error, {e}"),
181            }
182        }
183        r
184    }
185}
zng_view_api/ipc.rs

zng_view_api/
ipc.rs
