retry support

jest.config.js

@@ -21,10 +21,10 @@ module.exports = {
   // An object that configures minimum threshold enforcement for coverage results
   coverageThreshold: {
     global: {
-      branches: 69.23,
-      functions: 88.88,
-      lines: 93.75,
-      statements: 93.75,
+      branches: 78.94,
+      functions: 100,
+      lines: 96.27,
+      statements: 96.27,
     },
   },
 

package.json

@@ -46,6 +46,7 @@
     "eslint-plugin-jsdoc": "^36.1.0",
     "eslint-plugin-node": "^11.1.0",
     "eslint-plugin-prettier": "^3.3.1",
+    "extension-port-stream": "^2.0.1",
     "jest": "^27.5.1",
     "jest-it-up": "^2.0.2",
     "json-rpc-engine": "^6.1.0",
@@ -54,7 +55,8 @@
     "rimraf": "^3.0.2",
     "ts-jest": "^27.1.4",
     "ts-node": "^10.7.0",
-    "typescript": "^4.2.4"
+    "typescript": "^4.2.4",
+    "webextension-polyfill-ts": "^0.26.0"
   },
   "engines": {
     "node": ">=14.0.0"

src/createStreamMiddleware.ts

@@ -20,16 +20,21 @@ interface IdMap {
   [requestId: string]: IdMapValue;
 }
 
+interface Options {
+  retryOnMessage?: string;
+}
+
 /**
  * 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.
  *
+ * @param options - Configuration options for middleware.
  * @returns The event emitter, middleware, and stream.
  */
-export default function createStreamMiddleware() {
-  const idMap: IdMap = {};
+export default function createStreamMiddleware(options: Options = {}) {
+  const idMap: IdMap = {}; // TODO: replace with actual Map
   const stream = new Duplex({
     objectMode: true,
     read: () => undefined,
@@ -45,13 +50,23 @@ export default function createStreamMiddleware() {
     end,
   ) => {
     // write req to stream
-    stream.push(req);
+    sendToStream(req);
     // register request on id map
     idMap[req.id as unknown as string] = { req, res, next, end };
   };
 
   return { events, middleware, stream };
 
+  /**
+   * Forwards JSON-RPC request to the stream.
+   *
+   * @param req - The JSON-RPC request object.
+   */
+  function sendToStream(req: JsonRpcRequest<unknown>) {
+    // TODO: limiting retries could be implemented here
+    stream.push(req);
+  }
+
   /**
    * Writes a JSON-RPC object to the stream.
    *
@@ -104,6 +119,19 @@ export default function createStreamMiddleware() {
    * @param notif - The notification to process.
    */
   function processNotification(notif: JsonRpcNotification<unknown>) {
+    if (options?.retryOnMessage && notif.method === options.retryOnMessage) {
+      retryStuckRequests();
+    }
     events.emit('notification', notif);
   }
+
+  /**
+   * Retry pending requests.
+   */
+  function retryStuckRequests() {
+    Object.values(idMap).forEach(({ req }) => {
+      // TODO: limiting retries could be implemented here
+      sendToStream(req);
+    });
+  }
 }

src/index.test.ts

@@ -1,6 +1,14 @@
+import { Duplex } from 'stream';
 import { JsonRpcEngine } from 'json-rpc-engine';
+import PortStream from 'extension-port-stream';
+import { Runtime } from 'webextension-polyfill-ts';
 import { createStreamMiddleware, createEngineStream } from '.';
 
+const artificialDelay = (t = 0) =>
+  new Promise((resolve) => setTimeout(resolve, t));
+// eslint-disable-next-line @typescript-eslint/no-empty-function
+const noop = function (_a: any) {};
+
 const jsonrpc = '2.0' as const;
 
 describe('createStreamMiddleware', () => {
@@ -98,3 +106,77 @@ describe('middleware and engine to stream', () => {
     expect(response).toStrictEqual(res);
   });
 });
+
+const RECONNECTED = 'CONNECTED';
+describe('retry logic in middleware connected to a port', () => {
+  it('retries requests on reconnect message', async () => {
+    // create guest
+    const engineA = new JsonRpcEngine();
+    const jsonRpcConnection = createStreamMiddleware({
+      retryOnMessage: RECONNECTED,
+    });
+    engineA.push(jsonRpcConnection.middleware);
+
+    // create port
+    let messageConsumer = noop;
+    const messages: any[] = [];
+    const extensionPort = {
+      onMessage: {
+        addListener: (cb: any) => {
+          messageConsumer = cb;
+        },
+      },
+      onDisconnect: {
+        addListener: noop,
+      },
+      postMessage(m: any) {
+        messages.push(m);
+      },
+    };
+
+    const connectionStream = new PortStream(
+      extensionPort as unknown as Runtime.Port,
+    );
+
+    // connect both
+    const clientSideStream = jsonRpcConnection.stream;
+    clientSideStream
+      .pipe(connectionStream as unknown as Duplex)
+      .pipe(clientSideStream);
+
+    // request and expected result
+    const req1 = { id: 1, jsonrpc, method: 'test' };
+    const req2 = { id: 2, jsonrpc, method: 'test' };
+    const res = { id: 1, jsonrpc, result: 'test' };
+
+    // Initially sent once
+    const responsePromise1 = engineA.handle(req1);
+    engineA.handle(req2);
+    await artificialDelay();
+
+    expect(messages).toHaveLength(2);
+
+    // Reconnected, gets sent again
+    messageConsumer({
+      method: RECONNECTED,
+    });
+    await artificialDelay();
+
+    expect(messages).toHaveLength(4);
+    expect(messages[0]).toBe(messages[2]);
+    expect(messages[1]).toBe(messages[3]);
+
+    messageConsumer(res);
+
+    expect(await responsePromise1).toStrictEqual(res);
+
+    // Handled messages don't get retried but unhandled still do
+
+    messageConsumer({
+      method: RECONNECTED,
+    });
+    await artificialDelay();
+
+    expect(messages).toHaveLength(5);
+  });
+});

yarn.lock

@@ -2113,6 +2113,13 @@ extend@~3.0.2:
   resolved "https://registry.yarnpkg.com/extend/-/extend-3.0.2.tgz#f8b1136b4071fbd8eb140aff858b1019ec2915fa"
   integrity sha512-fjquC59cD7CyW6urNXK0FBufkZcoiGG80wTuPujX590cB5Ttln20E2UB4S/WARVqhXffZl2LNgS+gQdPIIim/g==
 
+extension-port-stream@^2.0.1:
+  version "2.0.1"
+  resolved "https://registry.yarnpkg.com/extension-port-stream/-/extension-port-stream-2.0.1.tgz#d374820c581418c2275d3c4439ade0b82c4cfac6"
+  integrity sha512-ltrv4Dh/979I04+D4Te6TFygfRSOc5EBzzlHRldWMS8v73V80qWluxH88hqF0qyUsBXTb8NmzlmSipcre6a+rg==
+  dependencies:
+    webextension-polyfill-ts "^0.22.0"
+
 extsprintf@1.3.0:
   version "1.3.0"
   resolved "https://registry.yarnpkg.com/extsprintf/-/extsprintf-1.3.0.tgz#96918440e3041a7a414f8c52e3c574eb3c3e1e05"
@@ -4692,6 +4699,30 @@ walker@^1.0.7:
   dependencies:
     makeerror "1.0.12"
 
+webextension-polyfill-ts@^0.22.0:
+  version "0.22.0"
+  resolved "https://registry.yarnpkg.com/webextension-polyfill-ts/-/webextension-polyfill-ts-0.22.0.tgz#86cfd7bab4d9d779d98c8340983f4b691b2343f3"
+  integrity sha512-3P33ClMwZ/qiAT7UH1ROrkRC1KM78umlnPpRhdC/292UyoTTW9NcjJEqDsv83HbibcTB6qCtpVeuB2q2/oniHQ==
+  dependencies:
+    webextension-polyfill "^0.7.0"
+
+webextension-polyfill-ts@^0.26.0:
+  version "0.26.0"
+  resolved "https://registry.yarnpkg.com/webextension-polyfill-ts/-/webextension-polyfill-ts-0.26.0.tgz#80b7063ddaf99abaa1ca73aad0cec09f306612d3"
+  integrity sha512-XEFL+aYVEsm/d4RajVwP75g56c/w2aSHnPwgtUv8/nCzbLNSzRQIix6aj1xqFkA5yr7OIDkk3OD/QTnPp8ThYA==
+  dependencies:
+    webextension-polyfill "^0.8.0"
+
+webextension-polyfill@^0.7.0:
+  version "0.7.0"
+  resolved "https://registry.yarnpkg.com/webextension-polyfill/-/webextension-polyfill-0.7.0.tgz#0df1120ff0266056319ce1a622b09ad8d4a56505"
+  integrity sha512-su48BkMLxqzTTvPSE1eWxKToPS2Tv5DLGxKexLEVpwFd6Po6N8hhSLIvG6acPAg7qERoEaDL+Y5HQJeJeml5Aw==
+
+webextension-polyfill@^0.8.0:
+  version "0.8.0"
+  resolved "https://registry.yarnpkg.com/webextension-polyfill/-/webextension-polyfill-0.8.0.tgz#f80e9f4b7f81820c420abd6ffbebfa838c60e041"
+  integrity sha512-a19+DzlT6Kp9/UI+mF9XQopeZ+n2ussjhxHJ4/pmIGge9ijCDz7Gn93mNnjpZAk95T4Tae8iHZ6sSf869txqiQ==
+
 webidl-conversions@^5.0.0:
   version "5.0.0"
   resolved "https://registry.yarnpkg.com/webidl-conversions/-/webidl-conversions-5.0.0.tgz#ae59c8a00b121543a2acc65c0434f57b0fc11aff"