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

mj_controller/session_manager/
client_backend.rs
