perspective_server/local_session.rs
1// ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
2// ┃ ██████ ██████ ██████ █ █ █ █ █ █▄ ▀███ █ ┃
3// ┃ ▄▄▄▄▄█ █▄▄▄▄▄ ▄▄▄▄▄█ ▀▀▀▀▀█▀▀▀▀▀ █ ▀▀▀▀▀█ ████████▌▐███ ███▄ ▀█ █ ▀▀▀▀▀ ┃
4// ┃ █▀▀▀▀▀ █▀▀▀▀▀ █▀██▀▀ ▄▄▄▄▄ █ ▄▄▄▄▄█ ▄▄▄▄▄█ ████████▌▐███ █████▄ █ ▄▄▄▄▄ ┃
5// ┃ █ ██████ █ ▀█▄ █ ██████ █ ███▌▐███ ███████▄ █ ┃
6// ┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┫
7// ┃ Copyright (c) 2017, the Perspective Authors. ┃
8// ┃ ╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌ ┃
9// ┃ This file is part of the Perspective library, distributed under the terms ┃
10// ┃ of the [Apache License 2.0](https://www.apache.org/licenses/LICENSE-2.0). ┃
11// ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛
12
13use perspective_client::Session;
14
15use crate::ffi;
16use crate::server::{Server, ServerError};
17
18/// A struct for implementing [`perspective_client::Session`] against an
19/// same-process [`Server`] instance.
20///
21/// See also [`perspective_client::ProxySession`] for implement the trait
22/// against an arbitrary remote transport.
23#[derive(Debug)]
24pub struct LocalSession {
25 pub(crate) id: u32,
26 pub(crate) server: Server,
27 pub(crate) closed: bool,
28}
29
30impl Drop for LocalSession {
31 fn drop(&mut self) {
32 if !self.closed {
33 tracing::error!("`Session` dropped without `Session::close`");
34 }
35 }
36}
37
38impl Session<ServerError> for LocalSession {
39 async fn handle_request(&self, request: &[u8]) -> Result<(), ServerError> {
40 let request = ffi::Request::from(request);
41 let responses = self.server.server.handle_request(self.id, &request);
42 let mut results = Vec::with_capacity(responses.size());
43 for response in responses.iter_responses() {
44 let cb = self
45 .server
46 .callbacks
47 .read()
48 .await
49 .get(&response.client_id())
50 .cloned();
51
52 if let Some(f) = cb {
53 results.push(f(response.msg()).await);
54 }
55 }
56
57 if let Some(cb) = &self.server.on_poll_request {
58 cb(&self.server).await?
59 } else {
60 results.push(self.server.poll().await);
61 }
62
63 results.into_iter().collect()
64 }
65
66 async fn close(mut self) {
67 self.closed = true;
68 self.server.server.close_session(self.id);
69 self.server
70 .callbacks
71 .write()
72 .await
73 .remove(&self.id)
74 .expect("Already closed");
75 }
76}