Yehowshua/ZIm
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Implement /rpc_server_snapshot endpoint

Browse source 
This returns a JSON snapshot of the state of the server
This commit is contained in:
Nathan Sobo 2022-05-25 17:42:25 -06:00
parent 6a32d55d85
commit 742dd75041
6 changed files with 68 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

14
crates/collab/src/api.rs
View file

@ -1,7 +1,7 @@
use crate::{ use crate::{
auth, auth,
db::{User, UserId}, db::{User, UserId},
rpc::ResultExt, rpc::{self, ResultExt},
AppState, Error, Result, AppState, Error, Result,
}; };
use anyhow::anyhow; use anyhow::anyhow;
@ -15,11 +15,12 @@ use axum::{
Extension, Json, Router, Extension, Json, Router,
}; };
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::sync::Arc; use std::sync::Arc;
use tower::ServiceBuilder; use tower::ServiceBuilder;
use tracing::instrument; use tracing::instrument;
pub fn routes(rpc_server: &Arc<crate::rpc::Server>, state: Arc<AppState>) -> Router<Body> { pub fn routes(rpc_server: &Arc<rpc::Server>, state: Arc<AppState>) -> Router<Body> {
Router::new() Router::new()
.route("/users", get(get_users).post(create_user)) .route("/users", get(get_users).post(create_user))
.route( .route(
@ -29,6 +30,7 @@ pub fn routes(rpc_server: &Arc<crate::rpc::Server>, state: Arc<AppState>) -> Rou
.route("/users/:id/access_tokens", post(create_access_token)) .route("/users/:id/access_tokens", post(create_access_token))
.route("/invite_codes/:code", get(get_user_for_invite_code)) .route("/invite_codes/:code", get(get_user_for_invite_code))
.route("/panic", post(trace_panic)) .route("/panic", post(trace_panic))
.route("/rpc_server_snapshot", get(get_rpc_server_snapshot))
.layer( .layer(
ServiceBuilder::new() ServiceBuilder::new()
.layer(Extension(state)) .layer(Extension(state))
@ -84,7 +86,7 @@ struct CreateUserParams {
async fn create_user( async fn create_user(
Json(params): Json<CreateUserParams>, Json(params): Json<CreateUserParams>,
Extension(app): Extension<Arc<AppState>>, Extension(app): Extension<Arc<AppState>>,
Extension(rpc_server): Extension<Arc<crate::rpc::Server>>, Extension(rpc_server): Extension<Arc<rpc::Server>>,
) -> Result<Json<User>> { ) -> Result<Json<User>> {
println!("{:?}", params); println!("{:?}", params);
@ -177,6 +179,12 @@ async fn trace_panic(panic: Json<Panic>) -> Result<()> {
Ok(()) Ok(())
} }
async fn get_rpc_server_snapshot<'a>(
Extension(rpc_server): Extension<Arc<rpc::Server>>,
) -> Result<Json<Value>> {
Ok(Json(serde_json::to_value(rpc_server.snapshot().await)?))
}
#[derive(Deserialize)] #[derive(Deserialize)]
struct CreateAccessTokenQueryParams { struct CreateAccessTokenQueryParams {
public_key: String, public_key: String,

6
crates/collab/src/main.rs
View file

@ -104,6 +104,12 @@ impl From<hyper::Error> for Error {
} }
} }
impl From<serde_json::Error> for Error {
fn from(error: serde_json::Error) -> Self {
Self::Internal(error.into())
}
}
impl IntoResponse for Error { impl IntoResponse for Error {
fn into_response(self) -> axum::response::Response { fn into_response(self) -> axum::response::Response {
match self { match self {

26
crates/collab/src/rpc.rs
View file

@ -33,6 +33,7 @@ use rpc::{
proto::{self, AnyTypedEnvelope, EntityMessage, EnvelopedMessage, RequestMessage}, proto::{self, AnyTypedEnvelope, EntityMessage, EnvelopedMessage, RequestMessage},
Connection, ConnectionId, Peer, Receipt, TypedEnvelope, Connection, ConnectionId, Peer, Receipt, TypedEnvelope,
}; };
use serde::{Serialize, Serializer};
use std::{ use std::{
any::TypeId, any::TypeId,
future::Future, future::Future,
@ -85,6 +86,7 @@ pub struct Server {
notifications: Option<mpsc::UnboundedSender<()>>, notifications: Option<mpsc::UnboundedSender<()>>,
} }
pub trait Executor: Send + Clone { pub trait Executor: Send + Clone {
type Sleep: Send + Future; type Sleep: Send + Future;
fn spawn_detached<F: 'static + Send + Future<Output = ()>>(&self, future: F); fn spawn_detached<F: 'static + Send + Future<Output = ()>>(&self, future: F);
@ -107,6 +109,23 @@ struct StoreWriteGuard<'a> {
_not_send: PhantomData<Rc<()>>, _not_send: PhantomData<Rc<()>>,
} }
#[derive(Serialize)]
pub struct ServerSnapshot<'a> {
peer: &'a Peer,
#[serde(serialize_with = "serialize_deref")]
store: RwLockReadGuard<'a, Store>,
}
pub fn serialize_deref<S, T, U>(value: &T, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
T: Deref<Target = U>,
U: Serialize
{
Serialize::serialize(value.deref(), serializer)
}
impl Server { impl Server {
pub fn new( pub fn new(
app_state: Arc<AppState>, app_state: Arc<AppState>,
@ -1469,6 +1488,13 @@ impl Server {
_not_send: PhantomData, _not_send: PhantomData,
} }
} }
pub async fn snapshot<'a>(self: &'a Arc<Self>) -> ServerSnapshot<'a> {
ServerSnapshot {
store: self.store.read().await,
peer: &self.peer
}
}
} }
impl<'a> Deref for StoreReadGuard<'a> { impl<'a> Deref for StoreReadGuard<'a> {

11
crates/collab/src/rpc/store.rs
View file

@ -2,18 +2,21 @@ use crate::db::{self, ChannelId, UserId};
use anyhow::{anyhow, Result}; use anyhow::{anyhow, Result};
use collections::{hash_map::Entry, BTreeMap, HashMap, HashSet}; use collections::{hash_map::Entry, BTreeMap, HashMap, HashSet};
use rpc::{proto, ConnectionId, Receipt}; use rpc::{proto, ConnectionId, Receipt};
use serde::Serialize;
use std::{collections::hash_map, mem, path::PathBuf}; use std::{collections::hash_map, mem, path::PathBuf};
use tracing::instrument; use tracing::instrument;
#[derive(Default)] #[derive(Default, Serialize)]
pub struct Store { pub struct Store {
connections: HashMap<ConnectionId, ConnectionState>, connections: HashMap<ConnectionId, ConnectionState>,
connections_by_user_id: HashMap<UserId, HashSet<ConnectionId>>, connections_by_user_id: HashMap<UserId, HashSet<ConnectionId>>,
projects: HashMap<u64, Project>, projects: HashMap<u64, Project>,
#[serde(skip)]
channels: HashMap<ChannelId, Channel>, channels: HashMap<ChannelId, Channel>,
next_project_id: u64, next_project_id: u64,
} }
#[derive(Serialize)]
struct ConnectionState { struct ConnectionState {
user_id: UserId, user_id: UserId,
projects: HashSet<u64>, projects: HashSet<u64>,
@ -21,21 +24,25 @@ struct ConnectionState {
channels: HashSet<ChannelId>, channels: HashSet<ChannelId>,
} }
#[derive(Serialize)]
pub struct Project { pub struct Project {
pub host_connection_id: ConnectionId, pub host_connection_id: ConnectionId,
pub host_user_id: UserId, pub host_user_id: UserId,
pub guests: HashMap<ConnectionId, (ReplicaId, UserId)>, pub guests: HashMap<ConnectionId, (ReplicaId, UserId)>,
#[serde(skip)]
pub join_requests: HashMap<UserId, Vec<Receipt<proto::JoinProject>>>, pub join_requests: HashMap<UserId, Vec<Receipt<proto::JoinProject>>>,
pub active_replica_ids: HashSet<ReplicaId>, pub active_replica_ids: HashSet<ReplicaId>,
pub worktrees: HashMap<u64, Worktree>, pub worktrees: HashMap<u64, Worktree>,
pub language_servers: Vec<proto::LanguageServer>, pub language_servers: Vec<proto::LanguageServer>,
} }
#[derive(Default)] #[derive(Default, Serialize)]
pub struct Worktree { pub struct Worktree {
pub root_name: String, pub root_name: String,
pub visible: bool, pub visible: bool,
#[serde(skip)]
pub entries: HashMap<u64, proto::Entry>, pub entries: HashMap<u64, proto::Entry>,
#[serde(skip)]
pub diagnostic_summaries: BTreeMap<PathBuf, proto::DiagnosticSummary>, pub diagnostic_summaries: BTreeMap<PathBuf, proto::DiagnosticSummary>,
pub scan_id: u64, pub scan_id: u64,
} }

18
crates/rpc/src/peer.rs
View file

@ -10,6 +10,7 @@ use futures::{
FutureExt, SinkExt, StreamExt, FutureExt, SinkExt, StreamExt,
}; };
use parking_lot::{Mutex, RwLock}; use parking_lot::{Mutex, RwLock};
use serde::{ser::SerializeStruct, Serialize};
use smol_timeout::TimeoutExt; use smol_timeout::TimeoutExt;
use std::sync::atomic::Ordering::SeqCst; use std::sync::atomic::Ordering::SeqCst;
use std::{ use std::{
@ -24,7 +25,7 @@ use std::{
}; };
use tracing::instrument; use tracing::instrument;
#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)] #[derive(Clone, Copy, PartialEq, Eq, Hash, Debug, Serialize)]
pub struct ConnectionId(pub u32); pub struct ConnectionId(pub u32);
impl fmt::Display for ConnectionId { impl fmt::Display for ConnectionId {
@ -89,10 +90,12 @@ pub struct Peer {
next_connection_id: AtomicU32, next_connection_id: AtomicU32,
} }
#[derive(Clone)] #[derive(Clone, Serialize)]
pub struct ConnectionState { pub struct ConnectionState {
#[serde(skip)]
outgoing_tx: mpsc::UnboundedSender<proto::Message>, outgoing_tx: mpsc::UnboundedSender<proto::Message>,
next_message_id: Arc<AtomicU32>, next_message_id: Arc<AtomicU32>,
#[serde(skip)]
response_channels: response_channels:
Arc<Mutex<Option<HashMap<u32, oneshot::Sender<(proto::Envelope, oneshot::Sender<()>)>>>>>, Arc<Mutex<Option<HashMap<u32, oneshot::Sender<(proto::Envelope, oneshot::Sender<()>)>>>>>,
} }
@ -471,6 +474,17 @@ impl Peer {
} }
} }
impl Serialize for Peer {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
let mut state = serializer.serialize_struct("Peer", 2)?;
state.serialize_field("connections", &*self.connections.read())?;
state.end()
}
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;

1
styles/src/styleTree/workspace.ts
View file

@ -8,7 +8,6 @@ export function workspaceBackground(theme: Theme) {
} }
export default function workspace(theme: Theme) { export default function workspace(theme: Theme) {
const tab = { const tab = {
height: 32, height: 32,
background: workspaceBackground(theme), background: workspaceBackground(theme),