✅ Get not initialized test case passing

Author: cowboyd

json-rpc-connection.ts

@@ -1,5 +1,5 @@
 import { Readable, Writable } from "node:stream";
-import { type Operation, resource } from "effection";
+import { type Operation, resource, useAbortSignal } from "effection";
 import * as rpc from "vscode-jsonrpc/node.js";
 
 export type { MessageConnection } from "vscode-jsonrpc";
@@ -13,15 +13,24 @@ export function useConnection(
   options: JSONRPCConnectionOptions,
 ): Operation<rpc.MessageConnection> {
   return resource(function* (provide) {
-    let connection = rpc.createMessageConnection(
+    let signal = yield* useAbortSignal();
+
+    let readable = new rpc.StreamMessageReader(
       //@ts-expect-error 🤷
-      new rpc.StreamMessageReader(Readable.fromWeb(options.read)),
-      new rpc.StreamMessageWriter(Writable.fromWeb(options.write)),
+      Readable.fromWeb(options.read, { signal }),
+    );
+    let writable = new rpc.StreamMessageWriter(
+      Writable.fromWeb(options.write, { signal }),
     );
+
+    let connection = rpc.createMessageConnection(readable, writable);
+
     connection.listen();
     try {
       yield* provide(connection);
     } finally {
+      readable.dispose();
+      writable.dispose();
       connection.dispose();
     }
   });

middleware.ts

@@ -0,0 +1,17 @@
+import type { Operation } from "effection";
+
+export interface Middleware<In, Out> {
+  (request: In, next: (input: In) => Operation<Out>): Operation<Out>;
+}
+
+export function concat<A, B>(
+  ...middlewares: Middleware<A, B>[]
+): Middleware<A, B> {
+  if (middlewares.length === 0) {
+    return (request, next) => next(request);
+  } else {
+    return middlewares.reduceRight((rest, middleware) => {
+      return (request, next) => middleware(request, (req) => rest(req, next));
+    });
+  }
+}

server.ts

@@ -1,9 +1,14 @@
 import {
   call,
+  createContext,
   createSignal,
+  ensure,
   type Operation,
+  race,
   resource,
+  suspend,
   useScope,
+  withResolvers,
 } from "effection";
 import type {
   Disposable,
@@ -12,14 +17,15 @@ import type {
   RequestParams,
 } from "./types.ts";
 
-import { useCommand, useDaemon } from "./use-command.ts";
+import { useDaemon } from "./use-command.ts";
 import { useConnection } from "./json-rpc-connection.ts";
 import {
   ErrorCodes,
   type MessageConnection,
   ResponseError,
   type StarRequestHandler,
 } from "vscode-jsonrpc";
+import { concat, type Middleware } from "./middleware.ts";
 
 export interface LSPXOptions {
   interactive?: boolean;
@@ -34,69 +40,134 @@ export function start(opts: LSPXOptions): Operation<LSPXServer> {
     let connections: MessageConnection[] = [];
     let disposables: Disposable[] = [];
 
-    try {
-      for (let command of opts.commands) {
-        let [exe, ...args] = command.split(/\s/g);
-        let process = yield* useDaemon(exe, {
-          args,
-          stdin: "piped",
-          stdout: "piped",
-          stderr: "piped",
-        });
-        let connection = yield* useConnection({
-          read: process.stdout,
-          write: process.stdin,
-        });
-        connections.push(connection);
-        disposables.push(connection);
-        disposables.push(
-          connection.onNotification((method, params) =>
-            notifications.send({ method, params })
-          ),
-        );
-      }
-
-      let client = yield* useConnection({
-        read: opts.input ?? ReadableStream.from([]),
-        write: opts.output ?? new WritableStream(),
+    for (let command of opts.commands) {
+      let [exe, ...args] = command.split(/\s/g);
+      let process = yield* useDaemon(exe, {
+        args,
+        stdin: "piped",
+        stdout: "piped",
+        stderr: "piped",
+      });
+      let connection = yield* useConnection({
+        read: process.stdout,
+        write: process.stdin,
       });
+      connections.push(connection);
+      disposables.push(connection);
+      disposables.push(
+        connection.onNotification((method, params) =>
+          notifications.send({ method, params })
+        ),
+      );
+    }
 
-      let scope = yield* useScope();
-      let dispatch = createDispatch(connections);
+    let client = yield* useConnection({
+      read: opts.input ?? ReadableStream.from([]),
+      write: opts.output ?? new WritableStream(),
+    });
 
-      let handler: StarRequestHandler = (...params) =>
-        scope.run(() => dispatch(...params));
+    let scope = yield* useScope();
+    let dispatch = createDispatch(connections);
 
-      disposables.push(client.onRequest(handler));
+    let handler: StarRequestHandler = (...params) =>
+      scope.run(() => dispatch(...params));
 
-      yield* provide({
-        notifications,
-        *request<T>(...params: RequestParams): Operation<T> {
-          const result = yield* dispatch<T, unknown>(...params);
-          if (result instanceof ResponseError) {
-            throw result;
-          }
-          return result;
-        },
-      });
-    } finally {
+    disposables.push(client.onRequest(handler));
+
+    yield* ensure(() => {
       for (let disposable of disposables) {
         disposable.dispose();
       }
-    }
+    });
+
+    yield* provide({
+      notifications,
+      *request<T>(...params: RequestParams): Operation<T> {
+        const result = yield* dispatch<T, unknown>(...params);
+        if (result instanceof ResponseError) {
+          throw result;
+        }
+        return result;
+      },
+    });
   });
 }
 
+const ResponseErrorContext = createContext<
+  <T>(error: ResponseError<T>) => void
+>("lspx.responseError");
+
 function createDispatch(
   connections: MessageConnection[],
 ): <T, E>(...params: RequestParams) => Operation<T | ResponseError<E>> {
-  return function* dispatch(...params) {
-    for (let connection of connections) {
-      return (yield* call(() => connection.sendRequest(...params)));
+  let middleware: Middleware<RequestParams, unknown> = concat(
+    ensureInitialized(),
+    multiplex(connections),
+  );
+
+  return function* dispatch<T, E>(
+    ...params: RequestParams
+  ): Operation<T | ResponseError<E>> {
+    let { operation: errored, resolve: raise } = withResolvers<
+      ResponseError<unknown>
+    >();
+    yield* ResponseErrorContext.set(raise);
+
+    return yield* race([
+      errored as Operation<ResponseError<E>>,
+      middleware(
+        params,
+        () => responseError(ErrorCodes.InternalError, "unhandled request"),
+      ) as Operation<T>,
+    ]);
+  };
+}
+
+function* responseError<T = void>(
+  ...args: ConstructorParameters<typeof ResponseError<T>>
+  // deno-lint-ignore no-explicit-any
+): Operation<any> {
+  let raise = yield* ResponseErrorContext.expect();
+
+  raise<T>(new ResponseError(...args));
+
+  yield* suspend();
+}
+
+function ensureInitialized(): Middleware<RequestParams, unknown> {
+  let initialized = false;
+  return function* (request, next) {
+    let [method] = request;
+    if (initialized && method === "initialize") {
+      return yield* responseError(
+        ErrorCodes.InvalidRequest,
+        "initialize invoked twice",
+      );
+    } else if (!initialized && method !== "initialize") {
+      return yield* responseError(
+        ErrorCodes.ServerNotInitialized,
+        "server not initialized",
+      );
+    } else {
+      initialized = true;
+      return yield* next(request);
+    }
+  };
+}
+
+function multiplex(
+  connections: MessageConnection[],
+): Middleware<RequestParams, unknown> {
+  let [connection] = connections;
+
+  return function* (params) {
+    if (!connection) {
+      return yield* responseError(
+        ErrorCodes.InternalError,
+        "lspx is not connected to any language servers",
+      );
+    } else {
+      return yield* call(() => connection.sendRequest(...params));
     }
-    return new ResponseError(
-      ErrorCodes.ServerNotInitialized,
-      `no active server connections`,
-    );
   };
 }

test/lspx.test.ts

@@ -24,11 +24,11 @@ describe("lspx", function () {
       });
     });
 
-    it.only("returns an error code: -32002 if a request is made before initialize", function* () {
+    it("returns an error code: -32002 if a request is made before initialize", function* () {
       let response = yield* request("textDocument/didOpen", {
         textDocument: {},
       });
-      expect(response).toBeErr("not initialized");
+      expect(response).toBeErr("server not initialized");
     });
 
     it("does not allow initialize to be sent more than once");

test/sim/initialize.ts

@@ -13,6 +13,7 @@ await main(function* () {
     write: Deno.stdout.writable,
   });
 
+  // deno-lint-ignore no-explicit-any
   const routes: Record<string, (...params: any[]) => Operation<any>> = {
     *initialize(_params: InitializeParams): Operation<InitializeResult> {
       return {

use-command.ts

@@ -17,6 +17,9 @@ export function useCommand(
     } finally {
       controller.abort();
       yield* call(() => process.status);
+      if (options?.stderr === "piped") {
+        process.stderr.cancel();
+      }
     }
   });
 }