pitchfork-cli 2.6.0

Daemons with DX
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

//! IPC request handling and dispatch
//!
//! Handles incoming IPC requests from CLI clients and routes them to the appropriate handlers.

use super::{SUPERVISOR, Supervisor};
use crate::Result;
use crate::ipc::server::IpcServer;
use crate::ipc::{IpcRequest, IpcResponse};

impl Supervisor {
    /// Main IPC connection watch loop - reads and dispatches requests
    pub(crate) async fn conn_watch(&self, mut ipc: IpcServer) -> ! {
        loop {
            let (msg, send) = match ipc.read().await {
                Ok(msg) => msg,
                Err(e) => {
                    error!("failed to accept connection: {e:?}");
                    continue;
                }
            };
            debug!("received message: {msg:?}");
            tokio::spawn(async move {
                let rsp = SUPERVISOR
                    .handle_ipc(msg)
                    .await
                    .unwrap_or_else(|err| IpcResponse::Error(err.to_string()));
                if let Err(err) = send.send(rsp).await {
                    debug!("failed to send message: {err:?}");
                }
            });
        }
    }

    /// Handle a single IPC request and return the appropriate response
    pub(crate) async fn handle_ipc(&self, req: IpcRequest) -> Result<IpcResponse> {
        let rsp = match req {
            IpcRequest::Invalid { error } => {
                warn!("Invalid IPC request: {error}");
                return Ok(IpcResponse::Error(format!("Invalid request: {error}")));
            }
            IpcRequest::Connect => {
                debug!("received connect message");
                IpcResponse::Ok
            }
            IpcRequest::Stop { id } => {
                // id is already DaemonId, no validation needed
                self.stop(&id).await?
            }
            IpcRequest::Run(opts) => {
                // opts.id is already DaemonId, no validation needed
                self.run(opts).await?
            }
            IpcRequest::Enable { id } => {
                // id is already DaemonId, no validation needed
                if self.enable(&id).await? {
                    IpcResponse::Yes
                } else {
                    IpcResponse::No
                }
            }
            IpcRequest::Disable { id } => {
                // id is already DaemonId, no validation needed
                if self.disable(&id).await? {
                    IpcResponse::Yes
                } else {
                    IpcResponse::No
                }
            }
            IpcRequest::GetActiveDaemons => {
                let daemons = self.active_daemons().await;
                IpcResponse::ActiveDaemons(daemons)
            }
            IpcRequest::GetNotifications => {
                let notifications = self.get_notifications().await;
                IpcResponse::Notifications(notifications)
            }
            IpcRequest::UpdateShellDir { shell_pid, dir } => {
                let prev = self.get_shell_dir(shell_pid).await;
                self.set_shell_dir(shell_pid, dir.clone()).await?;
                // Cancel any pending autostops for daemons in the new directory
                self.cancel_pending_autostops_for_dir(&dir).await;
                if let Some(prev) = prev {
                    self.leave_dir(&prev).await?;
                }
                self.refresh().await?;
                IpcResponse::Ok
            }
            IpcRequest::Clean => {
                self.clean().await?;
                IpcResponse::Ok
            }
            IpcRequest::GetDisabledDaemons => {
                let disabled = self.state_file.lock().await.disabled.clone();
                IpcResponse::DisabledDaemons(disabled.into_iter().collect())
            }
        };
        Ok(rsp)
    }
}