fix streamSource event emitter memory leak

maartyman · Nov 25, 2024 · 3508209 · 3508209
1 parent 66b2c91
commit 3508209
Showing 1 changed file with 8 additions and 2 deletions.
diff --git a/packages/actor-query-source-identify-stream/lib/StreamQuerySources.ts b/packages/actor-query-source-identify-stream/lib/StreamQuerySources.ts
@@ -49,6 +49,7 @@ export class StreamQuerySources implements IQuerySource {
     context: IActionContext,
   ) {
     this.sourcesEventEmitter = new EventEmitter();
+    this.sourcesEventEmitter.setMaxListeners(Infinity);
     stream.on('data', (item: IStreamQuerySource) => {
       if (item.isAddition) {
         const chunk: ISourceWrapper = {
@@ -144,10 +145,15 @@ export class StreamQuerySources implements IQuerySource {
     };
     iterator.readable = true;
 
-    this.sourcesEventEmitter.on('data', (sourceWrapper: ISourceWrapperSafe) => {
+    const addSourceToBuffer = (sourceWrapper: ISourceWrapperSafe) => {
+      if (iterator.done) {
+        this.sourcesEventEmitter.removeListener('data', addSourceToBuffer);
+        return;
+      }
       buffer.push(sourceWrapper);
       iterator.readable = true;
-    });
+    }
+    this.sourcesEventEmitter.on('data', addSourceToBuffer);
 
     const unionIterator = new UnionIterator<Bindings>(iterator, { autoStart: false });