Support stdin via service worker (#212)

* Support stdin via service worker

* Linting

* Move service worker implementation to comlink worker

* Make XeusRemoteKernel.mount abstract

Author: ianthomas23

packages/xeus/src/coincident.worker.ts

@@ -65,6 +65,10 @@ export class XeusCoincidentKernel extends XeusRemoteKernel {
     FS.mount(drive, {}, mountpoint);
     FS.chdir(mountpoint);
   }
+
+  protected _initializeStdin(baseUrl: string, browsingContextId: string): void {
+    // TODO: SharedArrayBuffer implementation
+  }
 }
 
 const worker = new XeusCoincidentKernel();

packages/xeus/src/comlink.worker.ts

@@ -7,6 +7,8 @@
 
 import { expose } from 'comlink';
 
+import { URLExt } from '@jupyterlab/coreutils';
+
 import { DriveFS } from '@jupyterlite/contents';
 
 import { XeusRemoteKernel } from './worker';
@@ -41,6 +43,34 @@ export class XeusComlinkKernel extends XeusRemoteKernel {
     FS.mount(drive, {}, mountpoint);
     FS.chdir(mountpoint);
   }
+
+  protected _initializeStdin(baseUrl: string, browsingContextId: string): void {
+    globalThis.get_stdin = (inputRequest: any): any => {
+      // Send a input request to the front-end via the service worker and block until
+      // the reply is received.
+      try {
+        const xhr = new XMLHttpRequest();
+        const url = URLExt.join(baseUrl, '/stdin/kernel');
+        xhr.open('POST', url, false); // Synchronous XMLHttpRequest
+        const msg = JSON.stringify({
+          browsingContextId,
+          data: inputRequest
+        });
+        // Send input request, this blocks until the input reply is received.
+        xhr.send(msg);
+        const inputReply = JSON.parse(xhr.response as string);
+
+        if ('error' in inputReply) {
+          // Service worker may return an error instead of an input reply message.
+          throw new Error(inputReply['error']);
+        }
+
+        return inputReply;
+      } catch (err) {
+        return { error: `Failed to request stdin via service worker: ${err}` };
+      }
+    };
+  }
 }
 
 const worker = new XeusComlinkKernel();

packages/xeus/src/interfaces.ts

@@ -88,5 +88,6 @@ export namespace IXeusWorkerKernel {
     kernelSpec: any;
     mountDrive: boolean;
     empackEnvMetaLink?: string;
+    browsingContextId: string;
   }
 }

packages/xeus/src/web_worker_kernel.ts

@@ -113,7 +113,8 @@ export class WebWorkerKernel implements IKernel {
         kernelSpec: this._kernelSpec,
         baseUrl: PageConfig.getBaseUrl(),
         mountDrive: options.mountDrive,
-        empackEnvMetaLink: this._empackEnvMetaLink
+        empackEnvMetaLink: this._empackEnvMetaLink,
+        browsingContextId: options.browsingContextId
       })
       .then(this._ready.resolve.bind(this._ready));
 

packages/xeus/src/worker.ts

@@ -29,18 +29,10 @@ globalThis.toplevel_promise_py_proxy = null;
 
 declare function createXeusModule(options: any): any;
 
-let resolveInputReply: any;
 let kernelReady: (value: unknown) => void;
 let rawXKernel: any;
 let rawXServer: any;
 
-async function get_stdin() {
-  const replyPromise = new Promise(resolve => {
-    resolveInputReply = resolve;
-  });
-  return replyPromise;
-}
-
 async function fetchJson(url: string): Promise<any> {
   const response = await fetch(url);
   if (!response.ok) {
@@ -50,8 +42,6 @@ async function fetchJson(url: string): Promise<any> {
   return json;
 }
 
-(self as any).get_stdin = get_stdin;
-
 globalThis.ready = new Promise(resolve => {
   kernelReady = resolve;
 });
@@ -76,7 +66,7 @@ export class XeusWorkerLogger implements ILogger {
   private _channel: BroadcastChannel;
 }
 
-export class XeusRemoteKernel {
+export abstract class XeusRemoteKernel {
   constructor(options: XeusRemoteKernel.IOptions = {}) {}
 
   async ready(): Promise<void> {
@@ -116,14 +106,20 @@ export class XeusRemoteKernel {
     }
 
     if (msg_type === 'input_reply') {
-      resolveInputReply(event.msg);
+      // Should never be called as input_reply messages are returned via service worker
     } else {
       rawXServer.notify_listener(event.msg);
     }
   }
 
   async initialize(options: IXeusWorkerKernel.IOptions): Promise<void> {
-    const { baseUrl, kernelSpec, empackEnvMetaLink, kernelId } = options;
+    const {
+      baseUrl,
+      browsingContextId,
+      kernelSpec,
+      empackEnvMetaLink,
+      kernelId
+    } = options;
 
     this._logger = new XeusWorkerLogger(kernelId);
 
@@ -210,6 +206,8 @@ export class XeusRemoteKernel {
         });
       }
 
+      this._initializeStdin(baseUrl, browsingContextId);
+
       rawXKernel = new globalThis.Module.xkernel();
       rawXServer = rawXKernel.get_server();
       if (!rawXServer) {
@@ -232,6 +230,27 @@ export class XeusRemoteKernel {
     kernelReady(1);
   }
 
+  /**
+   * Setup custom Emscripten FileSystem
+   */
+  abstract mount(
+    driveName: string,
+    mountpoint: string,
+    baseUrl: string,
+    browsingContextId: string
+  ): Promise<void>;
+
+  /**
+   * Add get_stdin function to globalThis that takes an input_request message, blocks
+   * until the corresponding input_reply is received and returns the input_reply message.
+   * If an error occurs return an object of the form { error: "Error explanation" }
+   * This function is called by xeus-lite's get_stdin.
+   */
+  protected abstract _initializeStdin(
+    baseUrl: string,
+    browsingContextId: string
+  ): void;
+
   private _logger: XeusWorkerLogger;
 }