procinsh 0.1.1

A local, read-only Linux x86-64 process inspector
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

use crate::state::AppState;
use axum::{
    Json,
    extract::State,
    http::StatusCode,
    response::{
        IntoResponse, Response, Sse,
        sse::{Event, KeepAlive},
    },
};
use serde_json::{Value, json};
use std::{convert::Infallible, sync::Arc, time::Duration};
pub async fn status(State(s): State<Arc<AppState>>) -> Json<Value> {
    Json(s.system.status.lock().unwrap().clone())
}
pub async fn snapshot(State(s): State<Arc<AppState>>) -> Json<Value> {
    Json(json!(**s.system.snapshot.read().unwrap()))
}
pub async fn events(State(s): State<Arc<AppState>>) -> Result<Response, StatusCode> {
    let viewer = s.system.viewer()?;
    let mut rx = s.system.events.subscribe();
    let stream = async_stream::stream! {
        let _viewer = viewer;
        let initial=serde_json::to_string(&**s.system.snapshot.read().unwrap()).unwrap_or_default();
        yield Ok::<_,Infallible>(Event::default().event("topology").data(initial));
        loop{if s.system.stopped(){break;}
            match tokio::time::timeout(Duration::from_secs(1),rx.recv()).await {
                Ok(Ok(message))=>{if let Ok(value)=serde_json::from_str::<Value>(&message){yield Ok(Event::default().event(value["event"].as_str().unwrap_or("message")).data(value["data"].to_string()));}},
                Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(n)))=>{yield Ok(Event::default().event("gap").data(json!({"dropped_frames":n}).to_string()));let snapshot=serde_json::to_string(&**s.system.snapshot.read().unwrap()).unwrap_or_default(); yield Ok(Event::default().event("topology").data(snapshot));},
                Ok(Err(_))=>break,Err(_)=>{}
            }
        }
    };
    Ok(Sse::new(stream)
        .keep_alive(KeepAlive::new().interval(Duration::from_secs(10)))
        .into_response())
}