fix: clean up engine listener

Author: iantanwx

src/createEngineStream.ts

@@ -1,5 +1,5 @@
-import { Duplex } from 'readable-stream';
-import { JsonRpcEngine, JsonRpcRequest } from 'json-rpc-engine';
+import { Duplex } from "readable-stream";
+import { JsonRpcEngine, JsonRpcRequest } from "json-rpc-engine";
 
 interface EngineStreamOptions {
   engine: JsonRpcEngine;
@@ -14,16 +14,16 @@ interface EngineStreamOptions {
  */
 export default function createEngineStream(opts: EngineStreamOptions): Duplex {
   if (!opts || !opts.engine) {
-    throw new Error('Missing engine parameter!');
+    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);
-    });
+    const onNotification = (message: unknown) => stream.push(message);
+    engine.on("notification", onNotification);
+    stream.on("end", () => engine.off("notification", onNotification));
   }
   return stream;
 
@@ -34,7 +34,7 @@ export default function createEngineStream(opts: EngineStreamOptions): Duplex {
   function write(
     req: JsonRpcRequest<unknown>,
     _encoding: unknown,
-    cb: (error?: Error | null) => void,
+    cb: (error?: Error | null) => void
   ) {
     engine.handle(req, (_err, res) => {
       stream.push(res);

test/index.js

@@ -1,57 +1,59 @@
-const test = require('tape');
-const { JsonRpcEngine } = require('json-rpc-engine');
-const { createStreamMiddleware, createEngineStream } = require('../dist');
-
-test('middleware - raw test', (t) => {
+const test = require("tape");
+const { JsonRpcEngine } = require("json-rpc-engine");
+const { createStreamMiddleware, createEngineStream } = require("../dist");
 
+test("middleware - raw test", (t) => {
   const jsonRpcConnection = createStreamMiddleware();
-  const req = { id: 1, jsonrpc: '2.0', method: 'test' };
-  const initRes = { id: 1, jsonrpc: '2.0' };
-  const res = { id: 1, jsonrpc: '2.0', result: 'test' };
+  const req = { id: 1, jsonrpc: "2.0", method: "test" };
+  const initRes = { id: 1, jsonrpc: "2.0" };
+  const res = { id: 1, jsonrpc: "2.0", result: "test" };
 
   // listen for incomming requests
-  jsonRpcConnection.stream.on('data', (_req) => {
-    t.equal(req, _req, 'got the expected request');
+  jsonRpcConnection.stream.on("data", (_req) => {
+    t.equal(req, _req, "got the expected request");
     jsonRpcConnection.stream.write(res);
   });
 
   // run middleware, expect end fn to be called
-  jsonRpcConnection.middleware(req, initRes, () => {
-    t.fail('should not call next');
-  }, (err) => {
-    t.notOk(err, 'should not error');
-    t.deepEqual(initRes, res, 'got the expected response');
-    t.end();
-  });
+  jsonRpcConnection.middleware(
+    req,
+    initRes,
+    () => {
+      t.fail("should not call next");
+    },
+    (err) => {
+      t.notOk(err, "should not error");
+      t.deepEqual(initRes, res, "got the expected response");
+      t.end();
+    }
+  );
 });
 
-test('engine to stream - raw test', (t) => {
-
+test("engine to stream - raw test", (t) => {
   const engine = new JsonRpcEngine();
   engine.push((_req, res, _next, end) => {
-    res.result = 'test';
+    res.result = "test";
     end();
   });
 
   const stream = createEngineStream({ engine });
-  const req = { id: 1, jsonrpc: '2.0', method: 'test' };
-  const res = { id: 1, jsonrpc: '2.0', result: 'test' };
+  const req = { id: 1, jsonrpc: "2.0", method: "test" };
+  const res = { id: 1, jsonrpc: "2.0", result: "test" };
 
   // listen for incomming requests
-  stream.on('data', (_res) => {
-    t.deepEqual(res, _res, 'got the expected response');
+  stream.on("data", (_res) => {
+    t.deepEqual(res, _res, "got the expected response");
     t.end();
   });
 
-  stream.on('error', (err) => {
+  stream.on("error", (err) => {
     t.fail(err.message);
   });
 
   stream.write(req);
 });
 
-test('middleware and engine to stream', (t) => {
-
+test("middleware and engine to stream", (t) => {
   // create guest
   const engineA = new JsonRpcEngine();
   const jsonRpcConnection = createStreamMiddleware();
@@ -60,35 +62,33 @@ test('middleware and engine to stream', (t) => {
   // create host
   const engineB = new JsonRpcEngine();
   engineB.push((_req, res, _next, end) => {
-    res.result = 'test';
+    res.result = "test";
     end();
   });
 
   // connect both
   const clientSideStream = jsonRpcConnection.stream;
   const hostSideStream = createEngineStream({ engine: engineB });
-  clientSideStream
-    .pipe(hostSideStream)
-    .pipe(clientSideStream);
+  clientSideStream.pipe(hostSideStream).pipe(clientSideStream);
 
   // request and expected result
-  const req = { id: 1, jsonrpc: '2.0', method: 'test' };
-  const res = { id: 1, jsonrpc: '2.0', result: 'test' };
+  const req = { id: 1, jsonrpc: "2.0", method: "test" };
+  const res = { id: 1, jsonrpc: "2.0", result: "test" };
 
   engineA.handle(req, (err, _res) => {
-    t.notOk(err, 'does not error');
-    t.deepEqual(res, _res, 'got the expected response');
+    t.notOk(err, "does not error");
+    t.deepEqual(res, _res, "got the expected response");
     t.end();
   });
 });
 
-test('server notification', (t) => {
+test("server notification", (t) => {
   t.plan(1);
 
   const jsonRpcConnection = createStreamMiddleware();
-  const notif = { jsonrpc: '2.0', method: 'test_notif' };
+  const notif = { jsonrpc: "2.0", method: "test_notif" };
 
-  jsonRpcConnection.events.once('notification', (message) => {
+  jsonRpcConnection.events.once("notification", (message) => {
     t.equals(message.method, notif.method);
     t.end();
   });
@@ -97,21 +97,32 @@ test('server notification', (t) => {
   jsonRpcConnection.stream.write(notif);
 });
 
-test('server notification in stream', (t) => {
+test("server notification in stream", (t) => {
   const engine = new JsonRpcEngine();
 
   const stream = createEngineStream({ engine });
-  const notif = { jsonrpc: '2.0', method: 'test_notif' };
+  const notif = { jsonrpc: "2.0", method: "test_notif" };
 
   // listen for incomming requests
-  stream.once('data', (_notif) => {
-    t.deepEqual(notif, _notif, 'got the expected notification');
+  stream.once("data", (_notif) => {
+    t.deepEqual(notif, _notif, "got the expected notification");
     t.end();
   });
 
-  stream.on('error', (err) => {
+  stream.on("error", (err) => {
     t.fail(err.message);
   });
 
-  engine.emit('notification', notif);
+  engine.emit("notification", notif);
+});
+
+test("clean up listener", (t) => {
+  const n = 10;
+  const engine = new JsonRpcEngine();
+  for (let i = 0; i < n; i++) {
+    const stream = createEngineStream({ engine });
+    stream.end();
+  }
+  t.equal(engine.listenerCount(), 0);
+  t.end();
 });