@trivago/samsa - npm Package Compare versions

Comparing version 0.3.0-beta.1 to 0.3.0-beta.2

lib/combinators/join.d.ts

@@ -15,2 +15,3 @@ /// <reference types="node" />
     constructor(primaryKTable: KTable, foreignKTable: KTable, projection: JoinProjection<any, any, any>);
+    private startProcessInterval;
     finishUp(): Promise<unknown>;
@@ -17,0 +18,0 @@ finalizeBuffer(): EventEmitter;

lib/combinators/join.js

@@ -91,7 +91,12 @@ "use strict";
         _this.subProcesses = new Set();
-        _this.keyBufferProcessInterval = setInterval(function () {
+        _this.keyBufferProcessInterval = _this.startProcessInterval();
+        return _this;
+    }
+    Joiner.prototype.startProcessInterval = function () {
+        var _this = this;
+        this.keyBufferProcessInterval = setInterval(function () {
             _this.finalizeBuffer();
         }, 1000);
-        return _this;
-    }
+        return this.keyBufferProcessInterval;
+    };
     Joiner.prototype.finishUp = function () {
@@ -146,2 +151,7 @@ var _this = this;
                 this.keyBuffer.add(key);
+                if (this.keyBuffer.size >= 1000) {
+                    clearInterval(this.keyBufferProcessInterval);
+                    this.finalizeBuffer();
+                    this.startProcessInterval();
+                }
                 next();
@@ -148,0 +158,0 @@ return [2 /*return*/];

package.json

 {
   "name": "@trivago/samsa",
-  "version": "0.3.0-beta.1",
+  "version": "0.3.0-beta.2",
   "types": "lib/index.d.ts",
@@ -5,0 +5,0 @@ "main": "lib/index.js",

src/combinators/join.ts

@@ -31,5 +31,10 @@ import { Readable, TransformCallback } from "stream";
 
+        this.keyBufferProcessInterval = this.startProcessInterval();
+    }
+
+    private startProcessInterval() {
         this.keyBufferProcessInterval = setInterval(() => {
             this.finalizeBuffer();
         }, 1000);
+        return this.keyBufferProcessInterval;
     }
@@ -97,2 +102,9 @@
         this.keyBuffer.add(key);
+
+        if (this.keyBuffer.size >= 1000) {
+            clearInterval(this.keyBufferProcessInterval);
+            this.finalizeBuffer();
+            this.startProcessInterval();
+        }
+
         next();
@@ -99,0 +111,0 @@ }

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Worsened metrics

Total package byte prevSize

200743

decreased by-14.05%

Number of package files

155

decreased by-0.64%

Lines of code

4821

decreased by-13.51%

No dependency changes