stop trying to use R_Is_Running (#176)

* use more evocative name

* switch to crossbeam

* use bus for broadcast channel

* refactor

Author: kevinushey

extensions/positron-r/amalthea/Cargo.lock

@@ -22,6 +22,7 @@ uuid = { version = "0.8.2", features = ["v4"] }
 zmq = { version = "0.9.2", features = ["vendored"] }
 strum = "0.24"
 strum_macros = "0.24"
+crossbeam = { version = "0.8.2", features = ["crossbeam-channel"] }
 
 [dev-dependencies]
 rand = "0.8.5"

extensions/positron-r/amalthea/crates/amalthea/Cargo.toml

@@ -5,6 +5,10 @@
  *
  */
 
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::thread;
+
 use crate::connection_file::ConnectionFile;
 use crate::error::Error;
 use crate::language::control_handler::ControlHandler;
@@ -19,10 +23,10 @@ use crate::socket::shell::Shell;
 use crate::socket::socket::Socket;
 use crate::socket::stdin::Stdin;
 use crate::stream_capture::StreamCapture;
-use std::sync::mpsc::sync_channel;
-use std::sync::mpsc::{Receiver, SyncSender};
-use std::sync::{Arc, Mutex};
-use std::thread;
+
+use crossbeam::channel::Receiver;
+use crossbeam::channel::Sender;
+use crossbeam::channel::bounded;
 use log::{warn, info};
 
 /// A Kernel represents a unique Jupyter kernel session and is the host for all
@@ -35,7 +39,7 @@ pub struct Kernel {
     session: Session,
 
     /// Sends messages to the IOPub socket
-    iopub_sender: SyncSender<IOPubMessage>,
+    iopub_sender: Sender<IOPubMessage>,
 
     /// Receives message sent to the IOPub socket
     iopub_receiver: Option<Receiver<IOPubMessage>>,
@@ -55,12 +59,12 @@ impl Kernel {
     pub fn new(file: ConnectionFile) -> Result<Kernel, Error> {
         let key = file.key.clone();
 
-        let (iopub_sender, iopub_receiver) = sync_channel::<IOPubMessage>(10);
+        let (iopub_sender, iopub_receiver) = bounded::<IOPubMessage>(10);
 
         Ok(Self {
             connection: file,
             session: Session::create(key)?,
-            iopub_sender,
+            iopub_sender: iopub_sender,
             iopub_receiver: Some(iopub_receiver),
         })
     }
@@ -164,7 +168,7 @@ impl Kernel {
     }
 
     /// Returns a copy of the IOPub sending channel.
-    pub fn create_iopub_sender(&self) -> SyncSender<IOPubMessage> {
+    pub fn create_iopub_sender(&self) -> Sender<IOPubMessage> {
         self.iopub_sender.clone()
     }
 
@@ -177,7 +181,7 @@ impl Kernel {
     /// Starts the shell thread.
     fn shell_thread(
         socket: Socket,
-        iopub_sender: SyncSender<IOPubMessage>,
+        iopub_sender: Sender<IOPubMessage>,
         shell_handler: Arc<Mutex<dyn ShellHandler>>,
     ) -> Result<(), Error> {
         let mut shell = Shell::new(socket, iopub_sender.clone(), shell_handler);
@@ -210,9 +214,7 @@ impl Kernel {
     }
 
     /// Starts the output capture thread.
-    fn output_capture_thread(
-        iopub_sender: SyncSender<IOPubMessage>,
-    ) -> Result<(), Error> {
+    fn output_capture_thread(iopub_sender: Sender<IOPubMessage>) -> Result<(), Error> {
         let output_capture = StreamCapture::new(iopub_sender);
         output_capture.listen();
         Ok(())

extensions/positron-r/amalthea/crates/amalthea/src/kernel.rs

@@ -15,5 +15,5 @@ use crate::error::Error;
 #[async_trait]
 pub trait LspHandler: Send {
     /// Starts the LSP server and binds it to the given TCP address.
-    fn start(&self, tcp_address: String) -> Result<(), Error>;
+    fn start(&mut self, tcp_address: String) -> Result<(), Error>;
 }

extensions/positron-r/amalthea/crates/amalthea/src/language/lsp_handler.rs

@@ -21,9 +21,8 @@ use crate::wire::is_complete_request::IsCompleteRequest;
 use crate::wire::kernel_info_reply::KernelInfoReply;
 use crate::wire::kernel_info_request::KernelInfoRequest;
 
-use std::sync::mpsc::SyncSender;
-
 use async_trait::async_trait;
+use crossbeam::channel::Sender;
 
 #[async_trait]
 pub trait ShellHandler: Send {
@@ -89,5 +88,5 @@ pub trait ShellHandler: Send {
     /// input and deliver it via the `handle_input_reply` method.
     ///
     /// https://jupyter-client.readthedocs.io/en/stable/messaging.html#messages-on-the-stdin-router-dealer-channel
-    fn establish_input_handler(&mut self, handler: SyncSender<ShellInputRequest>);
+    fn establish_input_handler(&mut self, handler: Sender<ShellInputRequest>);
 }

extensions/positron-r/amalthea/crates/amalthea/src/language/shell_handler.rs

@@ -18,8 +18,8 @@ use crate::wire::jupyter_message::ProtocolMessage;
 use crate::wire::status::ExecutionState;
 use crate::wire::status::KernelStatus;
 use crate::wire::stream::StreamOutput;
+use crossbeam::channel::Receiver;
 use log::{trace, warn};
-use std::sync::mpsc::Receiver;
 
 pub struct IOPub {
     /// The underlying IOPub socket

extensions/positron-r/amalthea/crates/amalthea/src/socket/iopub.rs

@@ -31,12 +31,13 @@ use crate::wire::kernel_info_reply::KernelInfoReply;
 use crate::wire::kernel_info_request::KernelInfoRequest;
 use crate::wire::status::ExecutionState;
 use crate::wire::status::KernelStatus;
+use crossbeam::channel::Sender;
 use futures::executor::block_on;
 use log::{debug, trace, warn};
 use std::collections::HashMap;
-use std::sync::mpsc::SyncSender;
-use std::sync::{Arc, Mutex};
 use std::str::FromStr;
+use std::sync::Arc;
+use std::sync::Mutex;
 
 /// Wrapper for the Shell socket; receives requests for execution, etc. from the
 /// front end and handles them or dispatches them to the execution thread.
@@ -45,7 +46,7 @@ pub struct Shell {
     socket: Socket,
 
     /// Sends messages to the IOPub socket (owned by another thread)
-    iopub_sender: SyncSender<IOPubMessage>,
+    iopub_sender: Sender<IOPubMessage>,
 
     /// Language-provided shell handler object
     handler: Arc<Mutex<dyn ShellHandler>>,
@@ -62,7 +63,7 @@ impl Shell {
     /// * `handler` - The language's shell channel handler
     pub fn new(
         socket: Socket,
-        iopub_sender: SyncSender<IOPubMessage>,
+        iopub_sender: Sender<IOPubMessage>,
         handler: Arc<Mutex<dyn ShellHandler>>,
     ) -> Self {
         Self {

extensions/positron-r/amalthea/crates/amalthea/src/socket/shell.rs

@@ -5,15 +5,17 @@
  *
  */
 
+use std::sync::Arc;
+use std::sync::Mutex;
+
 use crate::language::shell_handler::ShellHandler;
 use crate::socket::socket::Socket;
 use crate::wire::input_request::ShellInputRequest;
 use crate::wire::jupyter_message::JupyterMessage;
 use crate::wire::jupyter_message::Message;
+use crossbeam::channel::bounded;
 use futures::executor::block_on;
 use log::{trace, warn};
-use std::sync::mpsc::sync_channel;
-use std::sync::{Arc, Mutex};
 
 pub struct Stdin {
     /// The ZeroMQ stdin socket
@@ -40,7 +42,7 @@ impl Stdin {
     /// 1. Wait for
     pub fn listen(&self) {
         // Create the communication channel for the shell handler and inject it
-        let (sender, receiver) = sync_channel::<ShellInputRequest>(1);
+        let (sender, receiver) = bounded::<ShellInputRequest>(1);
         {
             let mut shell_handler = self.handler.lock().unwrap();
             shell_handler.establish_input_handler(sender);