handle results

juleswritescode · juleswritescode · commit cb72d8472eb7 · 2024-11-04T11:57:54.000+01:00
diff --git a/crates/pg_base_db/src/path.rs b/crates/pg_base_db/src/path.rs
@@ -20,4 +20,3 @@ impl PgLspPath {
         }
     }
 }
-
diff --git a/crates/pg_lsp/src/client/client_config_opts.rs b/crates/pg_lsp/src/client/client_config_opts.rs
@@ -3,5 +3,5 @@ use serde::Deserialize;
 // TODO: Check that the Opts are correct (existed in server.rs)
 #[derive(Deserialize)]
 pub struct ClientConfigurationOptions {
-    pub db_connection_string: Option<String>
-} 
+    pub db_connection_string: Option<String>,
+}
diff --git a/crates/pg_lsp/src/client/mod.rs b/crates/pg_lsp/src/client/mod.rs
@@ -1,2 +1,2 @@
+pub mod client_config_opts;
 pub mod client_flags;
-pub mod client_config_opts;
diff --git a/crates/pg_lsp/src/debouncer.rs b/crates/pg_lsp/src/debouncer.rs
@@ -1,27 +1,27 @@
-pub(crate) struct SimpleTokioDebouncer<Args> {
+use std::{future::Future, pin::Pin};
+
+type AsyncBlock = Pin<Box<dyn Future<Output = ()> + 'static + Send>>;
+
+pub(crate) struct SimpleTokioDebouncer {
     handle: tokio::task::JoinHandle<()>,
-    tx: tokio::sync::mpsc::Sender<Args>,
+    tx: tokio::sync::mpsc::Sender<AsyncBlock>,
 }
 
-impl<Args> SimpleTokioDebouncer<Args> {
-    pub fn new<F>(timeout: std::time::Duration, mut callback: F) -> Self
-    where
-        F: FnMut(Args) + Send + 'static,
-        Args: Send + 'static,
-    {
+impl SimpleTokioDebouncer {
+    pub fn new(timeout: std::time::Duration) -> Self {
         let (tx, mut rx) = tokio::sync::mpsc::channel(100);
 
         let handle = tokio::spawn(async move {
-            let mut maybe_args: Option<Args> = None;
+            let mut maybe_args: Option<AsyncBlock> = None;
             let mut instant = tokio::time::Instant::now() + timeout;
 
             loop {
                 tokio::select! {
                     // If the timeout is reached, execute and reset the last received action
                     _ = tokio::time::sleep_until(instant) => {
                         match maybe_args {
-                            Some(args) => {
-                                callback(args);
+                            Some(block) => {
+                                block.await;
                                 maybe_args = None;
                             }
                             None => continue,
@@ -45,9 +45,8 @@ impl<Args> SimpleTokioDebouncer<Args> {
         Self { handle, tx }
     }
 
-    pub async fn debounce(&self, args: Args)
-    {
-        self.tx.send(args).await.unwrap();
+    pub async fn debounce(&self, block: AsyncBlock) {
+        self.tx.send(block).await.unwrap();
     }
 
     pub async fn shutdown(self) {
diff --git a/crates/pg_lsp/src/lib.rs b/crates/pg_lsp/src/lib.rs
@@ -1,6 +1,6 @@
 mod client;
 mod db_connection;
 mod debouncer;
-mod utils;
-mod session;
 pub mod server;
+mod session;
+mod utils;
diff --git a/crates/pg_lsp/src/main.rs b/crates/pg_lsp/src/main.rs
@@ -3,10 +3,7 @@ use tower_lsp::LspService;
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
-    let (server, client_socket) = LspService::build(|client| {
-        Server::new(client)
-    }).finish();
-
+    let (server, client_socket) = LspService::build(|client| Server::new(client)).finish();
 
     Ok(())
 }
diff --git a/crates/pg_lsp/src/server.rs b/crates/pg_lsp/src/server.rs
@@ -17,52 +17,21 @@ pub struct Server {
     client: Client,
     session: Session,
     client_capabilities: RwLock<Option<ClientFlags>>,
-    debouncer: SimpleTokioDebouncer<Url>,
+    debouncer: SimpleTokioDebouncer,
 }
 
 impl Server {
     pub fn new(client: Client) -> Self {
-        let session = Session::new();
-
-        let cloned_session = session.clone();
-        let cloned_client = client.clone();
-
-        let debouncer =
-            SimpleTokioDebouncer::new(std::time::Duration::from_millis(500), move |mut uri| {
-                normalize_uri(&mut uri);
-                let url = file_path(&uri);
-
-                let diagnostics = cloned_session.get_diagnostics_sync(url);
-
-                let diagnostics: Vec<Diagnostic> = diagnostics
-                    .into_iter()
-                    .map(|(d, r)| to_proto::diagnostic(d, r))
-                    .collect();
-
-                cloned_client.send_notification::<ShowMessage>(ShowMessageParams {
-                    typ: MessageType::INFO,
-                    message: format!("diagnostics {}", diagnostics.len()),
-                });
-
-                let params = PublishDiagnosticsParams {
-                    uri,
-                    diagnostics,
-                    version: None,
-                };
-
-                cloned_client.send_notification::<notification::PublishDiagnostics>(params);
-            });
-
         Self {
             client,
             session: Session::new(),
             client_capabilities: RwLock::new(None),
-            debouncer,
+            debouncer: SimpleTokioDebouncer::new(std::time::Duration::from_millis(500)),
         }
     }
 
     /// When the client sends a didChangeConfiguration notification, we need to parse the received JSON.
-    fn parse_options_from_client(
+    async fn parse_options_from_client(
         &self,
         mut value: serde_json::Value,
     ) -> Option<ClientConfigurationOptions> {
@@ -79,7 +48,8 @@ impl Server {
                 );
                 let typ = MessageType::WARNING;
                 self.client
-                    .send_notification::<ShowMessage>(ShowMessageParams { message, typ });
+                    .send_notification::<ShowMessage>(ShowMessageParams { message, typ })
+                    .await;
                 None
             }
         }
@@ -106,17 +76,15 @@ impl Server {
                     .next()
                     .expect("workspace/configuration request did not yield expected response.");
 
-                let opts = self.parse_options_from_client(relevant);
-
-                opts
+                self.parse_options_from_client(relevant).await
             }
             Err(why) => {
                 let message = format!(
                     "Unable to pull client options via workspace/configuration request: {}",
                     why
                 );
                 println!("{}", message);
-                self.client.log_message(MessageType::ERROR, message);
+                self.client.log_message(MessageType::ERROR, message).await;
                 None
             }
         }
@@ -137,7 +105,8 @@ impl Server {
             .send_notification::<ShowMessage>(ShowMessageParams {
                 typ: MessageType::INFO,
                 message: format!("diagnostics {}", diagnostics.len()),
-            });
+            })
+            .await;
 
         let params = PublishDiagnosticsParams {
             uri,
@@ -146,11 +115,44 @@ impl Server {
         };
 
         self.client
-            .send_notification::<notification::PublishDiagnostics>(params);
+            .send_notification::<notification::PublishDiagnostics>(params)
+            .await;
     }
 
-    async fn publish_diagnostics_debounced(&self, uri: Url) {
-        self.debouncer.debounce(uri);
+    async fn publish_diagnostics_debounced(&self, mut uri: Url) {
+        let session = self.session.clone();
+        let client = self.client.clone();
+
+        self.debouncer
+            .debounce(Box::pin(async move {
+                normalize_uri(&mut uri);
+                let url = file_path(&uri);
+
+                let diagnostics = session.get_diagnostics_sync(url);
+
+                let diagnostics: Vec<Diagnostic> = diagnostics
+                    .into_iter()
+                    .map(|(d, r)| to_proto::diagnostic(d, r))
+                    .collect();
+
+                client
+                    .send_notification::<ShowMessage>(ShowMessageParams {
+                        typ: MessageType::INFO,
+                        message: format!("diagnostics {}", diagnostics.len()),
+                    })
+                    .await;
+
+                let params = PublishDiagnosticsParams {
+                    uri,
+                    diagnostics,
+                    version: None,
+                };
+
+                client
+                    .send_notification::<notification::PublishDiagnostics>(params)
+                    .await;
+            }))
+            .await;
     }
 }
 
@@ -197,6 +199,8 @@ impl LanguageServer for Server {
     }
 
     async fn shutdown(&self) -> jsonrpc::Result<()> {
+        // TODO: Shutdown stuff.
+
         self.client
             .log_message(MessageType::INFO, "Postgres LSP terminated.")
             .await;
@@ -213,21 +217,41 @@ impl LanguageServer for Server {
                 .is_some_and(|o| o.db_connection_string.is_some())
             {
                 let conn_str = opts.unwrap().db_connection_string.unwrap();
-                self.session.change_db(conn_str).await;
+                match self.session.change_db(conn_str).await {
+                    Ok(_) => {}
+                    Err(err) => {
+                        self.client
+                            .show_message(
+                                MessageType::ERROR,
+                                format!("Pulled Client Options but failed to set them: {}", err),
+                            )
+                            .await
+                    }
+                }
                 return;
             }
         }
 
         // if we couldn't pull settings from the client,
         // we'll try parsing the passed in params.
-        let opts = self.parse_options_from_client(params.settings);
+        let opts = self.parse_options_from_client(params.settings).await;
 
         if opts
             .as_ref()
             .is_some_and(|o| o.db_connection_string.is_some())
         {
             let conn_str = opts.unwrap().db_connection_string.unwrap();
-            self.session.change_db(conn_str).await;
+            match self.session.change_db(conn_str).await {
+                Ok(_) => {}
+                Err(err) => {
+                    self.client
+                        .show_message(
+                            MessageType::ERROR,
+                            format!("Used Client Options from params but failed to set them: {}", err),
+                        )
+                        .await
+                }
+            }
         }
     }
 
@@ -268,7 +292,7 @@ impl LanguageServer for Server {
         let mut uri = params.text_document.uri;
         normalize_uri(&mut uri);
 
-        self.debouncer.debounce(uri).await
+        self.publish_diagnostics_debounced(uri).await;
     }
 
     async fn did_close(&self, params: DidCloseTextDocumentParams) {
diff --git a/crates/pg_lsp/src/session.rs b/crates/pg_lsp/src/session.rs
@@ -1,4 +1,4 @@
-use std::{collections::HashSet,sync::Arc};
+use std::{collections::HashSet, sync::Arc};
 
 use pg_base_db::{Change, DocumentChange, PgLspPath};
 use pg_commands::{Command, ExecuteStatementCommand};
@@ -9,7 +9,8 @@ use pg_workspace::Workspace;
 use text_size::TextSize;
 use tokio::sync::RwLock;
 use tower_lsp::lsp_types::{
-    CodeActionOrCommand, CompletionItem, CompletionItemKind, CompletionList, Hover, HoverContents, InlayHint, InlayHintKind, InlayHintLabel, MarkedString, Position, Range
+    CodeActionOrCommand, CompletionItem, CompletionItemKind, CompletionList, Hover, HoverContents,
+    InlayHint, InlayHintKind, InlayHintLabel, MarkedString, Position, Range,
 };
 
 use crate::{db_connection::DbConnection, utils::line_index_ext::LineIndexExt};
@@ -29,6 +30,10 @@ impl Session {
         }
     }
 
+    async fn shutdown(&mut self) {
+        // TODO
+    }
+
     /// `update_db_connection` will update `Self`'s database connection.
     /// If the passed-in connection string is the same that we're already connected to, it's a noop.
     /// Otherwise, it'll first open a new connection, replace `Self`'s connection, and then close
@@ -50,7 +55,8 @@ impl Session {
         let ide = self.ide.clone();
         db.listen_for_schema_updates(move |schema| {
             let _guard = ide.blocking_write().set_schema_cache(schema);
-        });
+        })
+        .await?;
 
         let mut current_db = self.db.blocking_write();
         let old_db = current_db.replace(db);
@@ -160,10 +166,7 @@ impl Session {
 
         let changed_files = ide.compute(pool);
 
-        changed_files
-            .into_iter()
-            .map(|p| p.document_url)
-            .collect()
+        changed_files.into_iter().map(|p| p.document_url).collect()
     }
 
     pub async fn get_available_code_actions_or_commands(

Original file line number	Diff line number	Diff line change
`@@ -20,4 +20,3 @@ impl PgLspPath {`
`20`	`20`	`}`
`21`	`21`	`}`
`22`	`22`	`}`
`23`		`-`
Original file line number	Diff line number	Diff line change
`@@ -1,2 +1,2 @@`
	`1`	`+pub mod client_config_opts;`
`1`	`2`	`pub mod client_flags;`
`2`		`-pub mod client_config_opts;`