mj_controller/session_manager/
client_backend.rs

1use super::*;
2
3#[derive(Clone)]
4pub(super) struct ClientSessionHandle(pub(super) ManagedSessionHandle);
5
6impl mj_client::session::SessionHandleBackend for ClientSessionHandle {
7    fn search_prompts(
8        &self,
9        bundle_id: String,
10        scope: mj_core::storage::HistoryScope,
11        query: String,
12    ) -> mj_client::session::BoxFuture<'_, Result<Vec<mj_core::storage::PromptHistoryEntry>>> {
13        let session_id = self.0.session_id().to_owned();
14        Box::pin(async move {
15            tokio::task::spawn_blocking(move || {
16                crate::database::search_prompts(&session_id, &bundle_id, scope, &query)
17            })
18            .await
19            .context("history search task")?
20        })
21    }
22    fn review_state(
23        &self,
24    ) -> mj_client::session::BoxFuture<'_, Result<mj_client::session::ReviewState>> {
25        let session_id = self.0.session_id().to_owned();
26        Box::pin(async move {
27            tokio::task::spawn_blocking(move || {
28                Ok(mj_client::session::ReviewState {
29                    review: crate::database::active_review(&session_id)?,
30                })
31            })
32            .await
33            .context("review restoration task")?
34        })
35    }
36
37    fn resolve_review_settings(
38        &self,
39        cancelled: std::sync::Arc<std::sync::atomic::AtomicBool>,
40    ) -> mj_client::session::BoxFuture<'_, Result<mj_core::review::settings::ResolvedReviewSettings>>
41    {
42        Box::pin(crate::review_selection::resolve(
43            self.0.clone(),
44            None,
45            false,
46            cancelled,
47        ))
48    }
49
50    fn config_result(
51        &self,
52        command_id: String,
53    ) -> mj_client::session::BoxFuture<'_, Result<Option<Option<String>>>> {
54        let session_id = self.session_id().to_owned();
55        Box::pin(async move {
56            tokio::task::spawn_blocking(move || {
57                crate::database::load_config_result(&session_id, &command_id)
58            })
59            .await
60            .context("read configuration completion task")?
61        })
62    }
63
64    fn clone_box(&self) -> Box<dyn mj_client::session::SessionHandleBackend> {
65        Box::new(self.clone())
66    }
67
68    fn session_id(&self) -> &str {
69        self.0.session_id()
70    }
71
72    fn view(&self) -> ManagedSessionView {
73        self.0.view()
74    }
75
76    fn is_stopped(&self) -> bool {
77        self.0.is_stopped()
78    }
79
80    fn has_changed(&self) -> Result<bool> {
81        self.0.has_changed()
82    }
83
84    fn changed(&mut self) -> mj_client::session::BoxFuture<'_, Result<ManagedSessionView>> {
85        Box::pin(self.0.changed())
86    }
87
88    fn enqueue_submit(
89        &self,
90        command_id: String,
91        command: RelayCommand,
92    ) -> mj_client::session::BoxFuture<'_, Result<mj_client::session::PendingRelaySubmit>> {
93        Box::pin(async move {
94            let pending = self.0.enqueue_submit(command_id, command).await?;
95            Ok(mj_client::session::PendingRelaySubmit::new(Box::pin(
96                pending.wait(),
97            )))
98        })
99    }
100
101    fn enqueue_sync(
102        &self,
103    ) -> mj_client::session::BoxFuture<'_, Result<mj_client::session::PendingRelaySync>> {
104        Box::pin(async move {
105            let pending = self.0.enqueue_sync().await?;
106            Ok(mj_client::session::PendingRelaySync::new(Box::pin(
107                pending.wait(),
108            )))
109        })
110    }
111
112    fn respond_elicitation(
113        &self,
114        elicitation_id: String,
115        response: ElicitationResponse,
116    ) -> mj_client::session::BoxFuture<'_, Result<()>> {
117        Box::pin(self.0.respond_elicitation(elicitation_id, response))
118    }
119
120    fn stop_background_task(
121        &self,
122        background_task_id: String,
123    ) -> mj_client::session::BoxFuture<'_, Result<()>> {
124        Box::pin(self.0.stop_background_task(background_task_id))
125    }
126
127    fn reviewer(
128        &self,
129        role: Option<String>,
130        action: ReviewerAction,
131    ) -> mj_client::session::BoxFuture<'_, Result<ReviewerOutcome>> {
132        Box::pin(self.0.reviewer_as(role, action))
133    }
134}
135
136#[derive(Clone)]
137pub(super) struct ClientSessionControl(pub(super) SessionManagerControl);
138
139impl mj_client::session::SessionControlBackend for ClientSessionControl {
140    fn session(
141        &self,
142        session_id: String,
143    ) -> mj_client::session::BoxFuture<'_, Result<mj_client::session::SessionHandle>> {
144        Box::pin(async move { Ok(self.0.session(session_id).await?.client()) })
145    }
146}
147
148impl SessionManagerControl {
149    /// Narrow this controller-owned manager to session lookup for a control surface.
150    pub fn client(&self) -> mj_client::session::SessionControl {
151        mj_client::session::SessionControl::new(ClientSessionControl(self.clone()))
152    }
153
154    pub async fn session(&self, session_id: impl Into<String>) -> Result<ManagedSessionHandle> {
155        let session_id = session_id.into();
156        let (reply, response) = oneshot::channel();
157        self.commands
158            .send(ManagerCommand::Session {
159                session_id: session_id.clone(),
160                reply,
161            })
162            .await
163            .context("session manager stopped")?;
164        response
165            .await
166            .context("session manager stopped")?
167            .with_context(|| format!("session {session_id} is not managed"))
168    }
169
170    pub async fn wait_for_session(
171        &self,
172        session_id: &str,
173        timeout: Duration,
174    ) -> Result<ManagedSessionHandle> {
175        tokio::time::timeout(timeout, async {
176            loop {
177                match self.session(session_id.to_owned()).await {
178                    Ok(handle) => return Ok(handle),
179                    Err(error) => {
180                        tracing::trace!(session_id, "waiting for session actor: {error:#}");
181                        tokio::time::sleep(Duration::from_millis(25)).await;
182                    }
183                }
184            }
185        })
186        .await
187        .with_context(|| {
188            format!(
189                "session {session_id} did not become available within {} seconds",
190                timeout.as_secs()
191            )
192        })?
193    }
194}
mj_controller/session_manager/client_backend.rs

mj_controller/session_manager/
client_backend.rs
