wip: working on stuff

src/RPCServer.ts

@@ -510,36 +510,18 @@ class RPCServer {
         .pipeTo(passthroughTransform.writable)
         // Ignore any errors here, we only care that it ended
         .catch(() => {});
-      const headerStream = await (async () => {
-        const reader = inputStream.getReader();
-        const header = await reader.read();
-        reader.releaseLock(); // Release the lock to allow other reads
-        // Header must be defined
-        if (header.value == null) {
-          // TEMP: actually define the header error
-          throw new errors.ErrorRPC('header is undefined');
-        }
-        // Return a stream to be piped through the head transform
-        return new ReadableStream<Uint8Array>({
-          start: async (controller) => {
-            controller.enqueue(header.value);
-            controller.close();
-          },
-        });
-      })();
-      // Feed in the single header value into the header stream
-      void headerStream.pipeTo(headTransformStream.writable);
       const cleanUp = async (reason: any) => {
         await inputStream.cancel(reason);
         await rpcStream.writable.abort(reason);
         await inputStreamEndProm;
         timer.cancel(cleanupReason);
         await timer.catch(() => {});
       };
-      const reader = headTransformStream.readable.getReader();
+      const reader = inputStream.getReader();
+      console.log('about to read header message');
       // Allows timing out when waiting for the first message
       let headerMessage:
-        | ReadableStreamDefaultReadResult<JSONRPCRequest>
+        | ReadableStreamDefaultReadResult<Uint8Array>
         | undefined
         | void;
       try {
@@ -570,6 +552,7 @@ class RPCServer {
       }
       // Downgrade back to the raw stream
       reader.releaseLock();
+      console.log('read header message');
       // There are 2 conditions where we just end here
       //  1. The timeout timer resolves before the first message
       //  2. the stream ends before the first message
@@ -601,7 +584,23 @@ class RPCServer {
         );
         return;
       }
-      const method = headerMessage.value.method;
+      console.log('resulting header message', headerMessage);
+      const headerStream = new ReadableStream<Uint8Array>({
+        start: async (controller) => {
+          controller.enqueue(headerMessage!.value);
+          controller.close();
+        },
+      });
+      console.log('piping readable stream to head transform writable');
+      await headerStream.pipeTo(headTransformStream.writable);
+      console.log('piping finished');
+      // Read the transformed header message
+      const transformedReader = headTransformStream.readable.getReader();
+      const transformedHeaderMessage = (await transformedReader.read()).value;
+      if (transformedHeaderMessage == null) utils.never();
+      console.log('got header properly', transformedHeaderMessage);
+      // Use the parsed header message
+      const method = transformedHeaderMessage.method;
       const handler = this.handlerMap.get(method);
       if (handler == null) {
         await cleanUp(new errors.ErrorRPCHandlerFailed('Missing handler'));
@@ -631,7 +630,7 @@ class RPCServer {
       const headerWriter = rpcStream.writable.getWriter();
       try {
         handlerResult = await handler(
-          [headerMessage.value, inputStream],
+          [transformedHeaderMessage, inputStream],
           rpcStream.cancel,
           rpcStream.meta,
           { signal: abortController.signal, timer },

src/utils.ts

@@ -15,7 +15,7 @@ import type {
   PromiseDeconstructed,
   ToError,
 } from './types';
-import { ReadableStream, TransformStream } from 'stream/web';
+import { TransformStream } from 'stream/web';
 import { JSONParser } from '@streamparser/json';
 import { AbstractError } from '@matrixai/errors';
 import * as errors from './errors';