hook up async dispatcher in repl setup

Author: rgbkrk

Cargo.lock

@@ -262,6 +262,7 @@ anyhow = "1.0.57"
 any_vec = "0.13"
 ashpd = "0.8.0"
 async-compression = { version = "0.4", features = ["gzip", "futures-io"] }
+async-dispatcher = { version = "0.1"}
 async-fs = "1.6"
 async-recursion = "1.0.0"
 async-tar = "0.4.2"
@@ -325,7 +326,7 @@ rand = "0.8.5"
 refineable = { path = "./crates/refineable" }
 regex = "1.5"
 repair_json = "0.1.0"
-runtimelib = "0.11.1"
+runtimelib = { path = "../../runtimed/runtimed/runtimelib", default-features = false, features = ["async-dispatcher-runtime"] }
 rusqlite = { version = "0.29.0", features = ["blob", "array", "modern_sqlite"] }
 rust-embed = { version = "8.4", features = ["include-exclude"] }
 schemars = "0.8"

Cargo.toml

@@ -15,6 +15,7 @@ doctest = false
 [dependencies]
 anyhow.workspace = true
 alacritty_terminal.workspace = true
+async-dispatcher.workspace = true
 base64.workspace = true
 collections.workspace = true
 editor.workspace = true

crates/repl/Cargo.toml

@@ -1,4 +1,5 @@
 use anyhow::{anyhow, Context as _, Result};
+use async_dispatcher::{set_dispatcher, Dispatcher, Runnable};
 use collections::{HashMap, HashSet};
 use editor::{
     display_map::{
@@ -18,10 +19,11 @@ use outputs::{ExecutionStatus, ExecutionView, LineHeight as _};
 use project::Fs;
 use runtimelib::JupyterMessageContent;
 use settings::Settings as _;
-use std::ops::Range;
-use std::sync::Arc;
+use std::{ops::Range, sync::mpsc::RecvTimeoutError, time::Instant};
+use std::{sync::Arc, time::Duration};
 use theme::{ActiveTheme, ThemeSettings};
 use ui::prelude::*;
+use util::ResultExt;
 use workspace::Workspace;
 
 mod outputs;
@@ -37,7 +39,79 @@ pub struct RuntimeManagerGlobal(Model<RuntimeManager>);
 
 impl Global for RuntimeManagerGlobal {}
 
+pub fn zed_dispatcher(cx: &mut AppContext) -> impl Dispatcher {
+    struct ZedDispatcher {
+        tx: std::sync::mpsc::Sender<(Runnable, Option<Instant>)>,
+        task: Task<()>,
+    }
+
+    let (tx, rx) = std::sync::mpsc::channel();
+
+    impl Dispatcher for ZedDispatcher {
+        fn dispatch(&self, runnable: async_dispatcher::Runnable) {
+            self.tx.send((runnable, None)).log_err();
+        }
+
+        fn dispatch_after(
+            &self,
+            duration: std::time::Duration,
+            runnable: async_dispatcher::Runnable,
+        ) {
+            self.tx
+                .send((runnable, Some(Instant::now() + duration)))
+                .log_err();
+        }
+    }
+
+    let executor = cx.background_executor();
+
+    let task = executor.spawn(async {
+        let mut timers = Vec::<(Runnable, Instant)>::new();
+        let rx = rx;
+
+        loop {
+            let timeout = timers
+                .first()
+                .map(|&(_, time)| time.saturating_duration_since(Instant::now()))
+                .unwrap_or_else(|| Duration::from_secs(u64::MAX));
+
+            match rx.recv_timeout(timeout) {
+                Ok((runnable, deadline)) => {
+                    if let Some(deadline) = deadline {
+                        let now = Instant::now();
+                        if deadline > now {
+                            let idx =
+                                match timers.binary_search_by_key(&deadline, |&(_, time)| time) {
+                                    Ok(i) => i,
+                                    Err(i) => i,
+                                };
+                            timers.insert(idx, (runnable, deadline));
+                            continue;
+                        }
+                    }
+                    runnable.run();
+                }
+
+                Err(RecvTimeoutError::Timeout) => {
+                    let now = Instant::now();
+                    while let Some((_, deadline)) = timers.first() {
+                        if *deadline > now {
+                            break;
+                        }
+                        timers.remove(0).0.run();
+                    }
+                }
+                Err(RecvTimeoutError::Disconnected) => break,
+            }
+        }
+    });
+
+    ZedDispatcher { tx, task }
+}
+
 pub fn init(fs: Arc<dyn Fs>, cx: &mut AppContext) {
+    set_dispatcher(zed_dispatcher(cx));
+
     let runtime_manager = cx.new_model(|cx| RuntimeManager::new(fs.clone(), cx));
     RuntimeManager::set_global(runtime_manager.clone(), cx);