Implement back end -> front end comm messages

extensions/jupyter-adapter/src/JupyterCommClose.ts

@@ -10,4 +10,7 @@
 export interface JupyterCommClose {
 	/** The ID of the comm to tear down (as a GUID) */
 	comm_id: string;  // eslint-disable-line
+
+	/** The message payload */
+	data: object;
 }

extensions/jupyter-adapter/src/JupyterSocket.ts

@@ -19,7 +19,7 @@ export class JupyterSocket implements vscode.Disposable {
 	 * @param _channel The output channel to use for debugging
 	 */
 	constructor(title: string, socket: zmq.Socket,
-		private readonly _channel: vscode.OutputChannel) {
+		private readonly _logger: (msg: string) => void) {
 		this._socket = socket;
 		this._title = title;
 
@@ -48,7 +48,7 @@ export class JupyterSocket implements vscode.Disposable {
 		const maxTries = 10;
 		this._port = port;
 		this._addr = 'tcp://127.0.0.1:' + port.toString();
-		this._channel.appendLine(`${this._title} socket connecting to ${this._addr}...`);
+		this._logger(`${this._title} socket connecting to ${this._addr}...`);
 
 		// Monitor the socket for events; this is necessary to get events like
 		// `connect` to fire (otherwise we just get `message` events from the
@@ -68,14 +68,14 @@ export class JupyterSocket implements vscode.Disposable {
 		// Resolve the promise when the socket connects
 		return new Promise<void>((resolve, reject) => {
 			this._socket.on('connect', (_evt, addr) => {
-				this._channel.appendLine(`${this._title} socket connected to ${addr}`);
+				this._logger(`${this._title} socket connected to ${addr}`);
 				resolve();
 			});
 
 			// If the socket fails to connect, reject the promise
 			this._socket.on('connect_delay', (_evt, addr) => {
 				if (triesLeft-- === 0) {
-					this._channel.appendLine(`${this._title} socket failed to connect to ${addr} after ${maxTries} attempts`);
+					this._logger(`${this._title} socket failed to connect to ${addr} after ${maxTries} attempts`);
 					reject();
 				}
 			});

extensions/jupyter-adapter/src/LanguageRuntimeAdapter.ts

@@ -24,6 +24,9 @@ import { JupyterIsCompleteRequest } from './JupyterIsCompleteRequest';
 import { JupyterKernelInfoRequest } from './JupyterKernelInfoRequest';
 import { JupyterHistoryReply } from './JupyterHistoryReply';
 import { JupyterHistoryRequest } from './JupyterHistoryRequest';
+import { findAvailablePort } from './PortFinder';
+import { JupyterCommMsg } from './JupyterCommMsg';
+import { JupyterCommClose } from './JupyterCommClose';
 
 /**
  * LangaugeRuntimeAdapter wraps a JupyterKernel in a LanguageRuntime compatible interface.
@@ -75,7 +78,7 @@ export class LanguageRuntimeAdapter
 		// Extract the first 32 characters of the hash as the runtime ID
 		const runtimeId = digest.digest('hex').substring(0, 32);
 
-		this._kernel = new JupyterKernel(this._context, this._spec, runtimeId, this._channel, this._lsp);
+		this._kernel = new JupyterKernel(this._context, this._spec, runtimeId, this._channel);
 
 		// Generate kernel metadata and ID
 		this.metadata = {
@@ -121,7 +124,7 @@ export class LanguageRuntimeAdapter
 		mode: positron.RuntimeCodeExecutionMode,
 		errorBehavior: positron.RuntimeErrorBehavior): void {
 
-		this._channel.appendLine(`Sending code to ${this.metadata.languageName}: ${code}`);
+		this._kernel.log(`Sending code to ${this.metadata.languageName}: ${code}`);
 
 		// Forward execution request to the kernel
 		this._kernel.execute(code, id, mode, errorBehavior);
@@ -155,7 +158,7 @@ export class LanguageRuntimeAdapter
 	 * @param reply The user's response
 	 */
 	public replyToPrompt(id: string, reply: string): void {
-		this._channel.appendLine(`Sending reply to prompt ${id}: ${reply}`);
+		this._kernel.log(`Sending reply to prompt ${id}: ${reply}`);
 		this._kernel.replyToPrompt(id, reply);
 	}
 
@@ -174,7 +177,7 @@ export class LanguageRuntimeAdapter
 			throw new Error('Cannot interrupt kernel; it has already exited.');
 		}
 
-		this._channel.appendLine(`Interrupting ${this.metadata.languageName}`);
+		this._kernel.log(`Interrupting ${this.metadata.languageName}`);
 
 		// We are interrupting the kernel, so it's possible that message IDs
 		// that are currently being processed will never be completed. Clear the
@@ -248,32 +251,38 @@ export class LanguageRuntimeAdapter
 	 * Creates a new client instance.
 	 *
 	 * @param type The type of client to create
+	 * @param params The parameters for the client; the format of this object is
+	 *   specific to the client type
 	 * @returns A new client instance, or empty string if the type is not supported
 	 */
-	public createClient(type: positron.RuntimeClientType): string {
-		if (type === positron.RuntimeClientType.Environment) {
+	public async createClient(type: positron.RuntimeClientType, params: object): Promise<string> {
+		if (type === positron.RuntimeClientType.Environment ||
+			type === positron.RuntimeClientType.Lsp) {
 			// Currently the only supported client type
-			this._channel.appendLine(`Creating ${type} client for ${this.metadata.languageName}`);
+			this._kernel.log(`Creating '${type}' client for ${this.metadata.languageName}`);
 
 			// Create a new client adapter to wrap the comm channel
-			const adapter = new RuntimeClientAdapter(type, this._kernel);
+			const adapter = new RuntimeClientAdapter(type, params, this._kernel);
 
 			// Ensure we clean up the client from our internal state when it disconnects
 			adapter.onDidChangeClientState((e) => {
 				if (e === positron.RuntimeClientState.Closed) {
 					if (!this._comms.delete(adapter.getId())) {
-						this._channel.appendLine(`Warn: Runtime client adapater ${adapter.getId()} (${adapter.getClientType()}) not found`);
+						this._kernel.log(`Warn: Runtime client adapater ${adapter.getId()} (${adapter.getClientType()}) not found`);
 					}
 				}
 			});
 
+			// Open the client (this will send the comm_open message; wait for it to complete)
+			await adapter.open();
+
 			// Add the client to the map
 			this._comms.set(adapter.getId(), adapter);
 
 			// Return the ID of the new client
 			return adapter.getId();
 		} else {
-			this._channel.appendLine(`Info: can't create ${type} client for ${this.metadata.languageName} (not supported)`);
+			this._kernel.log(`Info: can't create ${type} client for ${this.metadata.languageName} (not supported)`);
 		}
 		return '';
 	}
@@ -286,10 +295,10 @@ export class LanguageRuntimeAdapter
 	removeClient(id: string): void {
 		const comm = this._comms.get(id);
 		if (comm) {
-			this._channel.appendLine(`Removing "${comm.getClientType()}" client ${comm.getClientId()} for ${this.metadata.languageName}`);
+			this._kernel.log(`Removing "${comm.getClientType()}" client ${comm.getClientId()} for ${this.metadata.languageName}`);
 			comm.dispose();
 		} else {
-			this._channel.appendLine(`Error: can't remove client ${id} (not found)`);
+			this._kernel.log(`Error: can't remove client ${id} (not found)`);
 		}
 	}
 
@@ -386,6 +395,10 @@ export class LanguageRuntimeAdapter
 			case 'input_request':
 				this.onInputRequest(msg, message as JupyterInputRequest);
 				break;
+			case 'comm_msg':
+				this.onCommMessage(msg, message as JupyterCommMsg);
+			case 'comm_close':
+				this.onCommClose(msg, message as JupyterCommClose);
 		}
 	}
 
@@ -407,6 +420,37 @@ export class LanguageRuntimeAdapter
 		} as positron.LanguageRuntimePrompt);
 	}
 
+	/**
+	 * Delivers a comm_msg message from the kernel to the appropriate client instance.
+	 *
+	 * @param message The outer message packet
+	 * @param msg The inner comm_msg message
+	 */
+	private onCommMessage(message: JupyterMessagePacket, msg: JupyterCommMsg): void {
+		this._messages.fire({
+			id: message.msgId,
+			parent_id: message.originId,
+			when: message.when,
+			type: positron.LanguageRuntimeMessageType.CommData,
+			data: msg.data,
+		} as positron.LanguageRuntimeCommMessage);
+	}
+
+	/**
+	 * Notifies the client that a comm has been closed from the kernel side.
+	 *
+	 * @param message The outer message packet
+	 * @param close The inner comm_msg message
+	 */
+	private onCommClose(message: JupyterMessagePacket, msg: JupyterCommMsg): void {
+		this._messages.fire({
+			id: message.msgId,
+			parent_id: message.originId,
+			when: message.when,
+			type: positron.LanguageRuntimeMessageType.CommClosed,
+			data: msg.data,
+		} as positron.LanguageRuntimeCommClosed);
+	}
 	/**
 	 * Converts a Positron event into a language runtime event and emits it.
 	 *
@@ -580,9 +624,40 @@ export class LanguageRuntimeAdapter
 	 * @param status The new status of the kernel
 	 */
 	onStatus(status: positron.RuntimeState) {
-		this._channel.appendLine(`Kernel status changed to ${status}`);
+		this._kernel.log(`${this._spec.language} kernel status changed to ${status}`);
 		this._kernelState = status;
 		this._state.fire(status);
+
+		// When the kernel becomes ready, start the LSP server if it's configured
+		if (status === positron.RuntimeState.Ready && this._lsp) {
+			findAvailablePort([], 25).then(port => {
+				this._kernel.log(`Kernel ready, connecting to ${this._spec.display_name} LSP server on port ${port}...`);
+				this.startLsp(`127.0.0.1:${port}`);
+				this._lsp!(port).then(() => {
+					this._kernel.log(`${this._spec.display_name} LSP server connected`);
+				});
+			});
+		}
+	}
+
+	/**
+	 * Requests that the kernel start a Language Server Protocol server, and
+	 * connect it to the client with the given TCP address.
+	 *
+	 * Note: This is only useful if the kernel hasn't already started an LSP
+	 * server.
+	 *
+	 * @param clientAddress The client's TCP address, e.g. '127.0.0.1:1234'
+	 */
+	public startLsp(clientAddress: string) {
+		// TODO: Should we query the kernel to see if it can create an LSP
+		// (QueryInterface style) instead of just demanding it?
+		//
+		// The Jupyter kernel spec does not provide a way to query for
+		// supported comms; the only way to know is to try to create one.
+
+		this._kernel.log(`Starting LSP server for ${clientAddress}`);
+		this.createClient(positron.RuntimeClientType.Lsp, { client_address: clientAddress });
 	}
 
 	/**

extensions/jupyter-adapter/src/RuntimeClientAdapter.ts

@@ -25,6 +25,7 @@ export class RuntimeClientAdapter {
 
 	constructor(
 		private readonly _type: positron.RuntimeClientType,
+		private readonly _params: object,
 		private readonly _kernel: JupyterKernel) {
 
 		this.id = uuidv4();
@@ -45,12 +46,15 @@ export class RuntimeClientAdapter {
 		// Bind to status stream from kernel
 		this.onStatus = this.onStatus.bind(this);
 		this._kernel.addListener('status', this.onStatus);
+	}
 
+	/**
+	 * Opens the communications channel between the client and the runtime.
+	 */
+	public async open() {
 		// Ask the kernel to open a comm channel for us
 		this._state.fire(positron.RuntimeClientState.Opening);
-		this._kernel.openComm(this._type, this.id, null);
-
-		// Consider the client connected once we've opened the comm
+		await this._kernel.openComm(this._type, this.id, this._params);
 		this._state.fire(positron.RuntimeClientState.Connected);
 	}
 

extensions/positron-r/amalthea/crates/amalthea/src/comm/comm_channel.rs

@@ -8,16 +8,14 @@
 use serde_json::Value;
 use strum_macros::EnumString;
 
-/// Rust trait that defines a custom Jupyter communication channel
-pub trait CommChannel: Send {
-    fn send_request(&self, data: &Value);
-    fn target_name(&self) -> String;
-    fn close(&self);
-}
-
 #[derive(EnumString, PartialEq)]
 #[strum(serialize_all = "snake_case")]
 pub enum Comm {
     Environment,
-    Lsp
+    Lsp,
+}
+
+pub enum CommChannelMsg {
+    Data(Value),
+    Close,
 }

extensions/positron-r/amalthea/crates/amalthea/src/comm/lsp_comm.rs

@@ -8,9 +8,11 @@
 use std::sync::Arc;
 use std::sync::Mutex;
 
+use crossbeam::channel::Sender;
+use serde_json::json;
 use serde_json::Value;
 
-use crate::comm::comm_channel::CommChannel;
+use crate::comm::comm_channel::CommChannelMsg;
 use crate::error::Error;
 use crate::language::lsp_handler::LspHandler;
 
@@ -23,39 +25,38 @@ pub struct StartLsp {
 }
 
 pub struct LspComm {
-    handler: Arc<Mutex<dyn LspHandler>>
+    handler: Arc<Mutex<dyn LspHandler>>,
+    msg_tx: Sender<Value>,
 }
 
 /**
  * LspComm makes an LSP look like a CommChannel; it's used to start the LSP and
  * track the server thread.
  */
 impl LspComm {
-    pub fn new(handler: Arc<Mutex<dyn LspHandler>>) -> LspComm {
-        LspComm {
-            handler
-        }
+    pub fn new(handler: Arc<Mutex<dyn LspHandler>>, msg_tx: Sender<Value>) -> LspComm {
+        LspComm { handler, msg_tx }
     }
 
-    pub fn start(&self, data: &StartLsp) ->  Result<(), Error> {
+    pub fn start(&self, data: &StartLsp) -> Result<(), Error> {
         let mut handler = self.handler.lock().unwrap();
         handler.start(data.client_address.clone()).unwrap();
+        self.msg_tx
+            .send(json!({
+                "msg_type": "lsp_started",
+                "content": {}
+            }))
+            .unwrap();
         Ok(())
     }
-}
-
-impl CommChannel for LspComm {
-    fn send_request(&self, _data: &Value) {
-        // Not implemented; LSP messages are delivered directly to the LSP
-        // handler via TCP, not proxied here.
-    }
 
-    fn target_name(&self) -> String {
-        "lsp".to_string()
+    /**
+     * Returns a Sender that can accept comm channel messages (required as part of the
+     * `CommChannel` contract). Because the LSP communicates over its own TCP connection, it does
+     * not process messages from the comm, and they are discarded here.
+     */
+    pub fn msg_sender(&self) -> Sender<CommChannelMsg> {
+        let (msg_tx, _msg_rx) = crossbeam::channel::unbounded();
+        msg_tx
     }
-
-    fn close(&self) {
-        // Not implemented; the LSP is closed automatically when the TCP
-        // connection is closed.
-    }
-}
+}

extensions/positron-r/amalthea/crates/amalthea/src/comm/mod.rs

@@ -6,5 +6,4 @@
  */
 
 pub mod comm_channel;
-pub mod environment_comm;
 pub mod lsp_comm;