Migrate to TypeScript

.circleci/config.yml

@@ -7,13 +7,13 @@ workflows:
       - test-lint:
           requires:
             - prep-deps
-      - test-unit:
+      - test-build-unit:
           requires:
             - prep-deps
       - all-tests-pass:
           requires:
             - test-lint
-            - test-unit
+            - test-build-unit
 
 jobs:
   prep-deps:
@@ -46,7 +46,7 @@ jobs:
           name: Lint
           command: yarn lint
 
-  test-unit:
+  test-build-unit:
     docker:
       - image: circleci/node:10
     steps:
@@ -55,7 +55,7 @@ jobs:
           at: .
       - run:
           name: Unit tests
-          command: yarn test
+          command: yarn build && yarn test
 
   all-tests-pass:
     docker:

.eslintrc.js

@@ -8,9 +8,24 @@ module.exports = {
   extends: [
     '@metamask/eslint-config',
     '@metamask/eslint-config/config/nodejs',
+    '@metamask/eslint-config/config/typescript',
   ],
+  overrides: [{
+    files: [
+      '*.js',
+      '*.json',
+    ],
+    parserOptions: {
+      sourceType: 'script',
+    },
+    rules: {
+      '@typescript-eslint/no-require-imports': 'off',
+      '@typescript-eslint/no-var-requires': 'off',
+    },
+  }],
   ignorePatterns: [
-    '!.eslintrc.js',
+    '!eslintrc.js',
+    'dist/',
     'node_modules/',
   ],
-}
+};

.gitignore

@@ -1,4 +1,4 @@
-
+dist
 package-lock.json
 
 # Created by https://www.gitignore.io/api/osx,node

package.json

@@ -1,29 +1,35 @@
 {
   "name": "json-rpc-middleware-stream",
   "version": "2.1.1",
-  "description": "A small toolset for streaming JSON RPC data and matching requests and responses.",
+  "description": "A small toolset for streaming JSON-RPC data and matching requests and responses.",
   "license": "ISC",
   "main": "index.js",
   "files": [
     "index.js",
     "engineStream.js"
   ],
   "scripts": {
+    "build": "tsc --project .",
     "test": "node test/index.js",
     "lint": "eslint . --ext js,json",
     "lint:fix": "eslint . --ext js,json --fix"
   },
   "dependencies": {
-    "readable-stream": "^2.3.3",
-    "safe-event-emitter": "^1.0.1"
+    "@metamask/safe-event-emitter": "^2.0.0",
+    "readable-stream": "^2.3.3"
   },
   "devDependencies": {
-    "@metamask/eslint-config": "^2.1.1",
-    "eslint": "^7.0.0",
+    "@metamask/eslint-config": "^4.1.0",
+    "@types/readable-stream": "^2.3.9",
+    "@typescript-eslint/eslint-plugin": "^4.9.0",
+    "@typescript-eslint/parser": "^4.9.0",
+    "eslint": "^7.14.0",
     "eslint-plugin-import": "^2.20.2",
     "eslint-plugin-json": "^2.1.1",
-    "json-rpc-engine": "^5.1.8",
-    "tape": "^5.0.0"
+    "eslint-plugin-node": "^11.1.0",
+    "json-rpc-engine": "^6.1.0",
+    "tape": "^5.0.0",
+    "typescript": "^4.1.2"
   },
   "repository": {
     "type": "git",

src/createEngineStream.ts

@@ -0,0 +1,44 @@
+import { Duplex } from 'readable-stream';
+import { JsonRpcEngine, JsonRpcRequest } from 'json-rpc-engine';
+
+interface EngineStreamOptions {
+  engine: JsonRpcEngine;
+}
+
+/**
+ * Takes a JsonRpcEngine and returns a Duplex stream wrapping it.
+ * 
+ * @param opts - Options bag.
+ * @param opts.engine - The JsonRpcEngine to wrap in a stream.
+ * @returns The stream wrapping the engine.
+ */
+export default function createEngineStream(opts: EngineStreamOptions): Duplex {
+  if (!opts || !opts.engine) {
+    throw new Error('Missing engine parameter!');
+  }
+
+  const { engine } = opts;
+  const stream = new Duplex({ objectMode: true, read, write });
+  // forward notifications
+  if (engine.on) {
+    engine.on('notification', (message) => {
+      stream.push(message);
+    });
+  }
+  return stream;
+
+  function read() {
+    return undefined;
+  }
+
+  function write(
+    req: JsonRpcRequest<unknown>,
+    _encoding: unknown,
+    cb: (error?: Error | null) => void
+  ) {
+    engine.handle(req, (_err, res) => {
+      stream.push(res);
+    });
+    cb();
+  }
+}

src/createStreamMiddleware.ts

@@ -0,0 +1,96 @@
+import SafeEventEmitter from '@metamask/safe-event-emitter';
+import { Duplex } from 'readable-stream';
+import {
+  JsonRpcEngineNextCallback,
+  JsonRpcEngineEndCallback,
+  JsonRpcNotification,
+  JsonRpcMiddleware,
+  JsonRpcRequest,
+  PendingJsonRpcResponse,
+} from 'json-rpc-engine';
+
+interface IdMapValue {
+  req: JsonRpcRequest<unknown>;
+  res: PendingJsonRpcResponse<unknown>;
+  next: JsonRpcEngineNextCallback;
+  end: JsonRpcEngineEndCallback;
+}
+
+interface IdMap {
+  [requestId: string]: IdMapValue;
+}
+
+/**
+ * Creates a JsonRpcEngine middleware with an associated Duplex stream and
+ * EventEmitter. The middleware, and by extension stream, assume that middleware
+ * parameters are properly formatted. No runtime type checking or validation is
+ * performed.
+ * 
+ * @returns The event emitter, middleware, and stream.
+ */
+export default function createStreamMiddleware() {
+  const idMap: IdMap = {};
+  const stream = new Duplex({
+    objectMode: true,
+    read: readNoop,
+    write: processMessage,
+  });
+
+  const events = new SafeEventEmitter();
+
+  const middleware: JsonRpcMiddleware<unknown, unknown> = (
+    req,
+    res,
+    next,
+    end
+  ) => {
+    // write req to stream
+    stream.push(req);
+    // register request on id map
+    idMap[(req.id as unknown) as string] = { req, res, next, end };
+  };
+
+  return { events, middleware, stream };
+
+  function readNoop() {
+    return false;
+  }
+
+  function processMessage(
+    res: PendingJsonRpcResponse<unknown>,
+    _encoding: unknown,
+    cb: (error?: Error | null) => void
+  ) {
+    let err;
+    try {
+      const isNotification = !res.id;
+      if (isNotification) {
+        processNotification((res as unknown) as JsonRpcNotification<unknown>);
+      } else {
+        processResponse(res);
+      }
+    } catch (_err) {
+      err = _err;
+    }
+    // continue processing stream
+    cb(err);
+  }
+
+  function processResponse(res: PendingJsonRpcResponse<unknown>) {
+    const context = idMap[(res.id as unknown) as string];
+    if (!context) {
+      throw new Error(`StreamMiddleware - Unknown response id "${res.id}"`);
+    }
+
+    delete idMap[(res.id as unknown) as string];
+    // copy whole res onto original res
+    Object.assign(context.res, res);
+    // run callback on empty stack,
+    // prevent internal stream-handler from catching errors
+    setTimeout(context.end);
+  }
+
+  function processNotification(res: JsonRpcNotification<unknown>) {
+    events.emit('notification', res);
+  }
+}

src/index.ts

@@ -0,0 +1,4 @@
+import createEngineStream from './createEngineStream';
+import createStreamMiddleware from './createStreamMiddleware';
+
+export { createEngineStream, createStreamMiddleware };