@nats-io/nats-core - npm Package Compare versions

Comparing version 3.0.0-36 to 3.0.0-37

lib/core.d.ts

@@ -14,3 +14,5 @@ /**
     /** Client received an async error from the server */
-    Error = "error"
+    Error = "error",
+    /** Slow Consumer - a buffered subscription (iterator) that is accumulating messages beyond a specify threshold */
+    SlowConsumer = "slow_consumer"
 }
@@ -57,2 +59,10 @@ /**
     callback?: MsgCallback<T>;
+    /**
+     * Number of pending messages in a subscription to exceed prior to considering
+     * a subscription a Slow Consumer. By default, slow consumer is on a subscription
+     * is not accounted for.
+     *
+     * This is an experimental option.
+     */
+    slow?: number;
 }
@@ -59,0 +69,0 @@ export interface DnsResolveFn {

lib/core.js

@@ -37,2 +37,4 @@ "use strict";
     Events["Error"] = "error";
+    /** Slow Consumer - a buffered subscription (iterator) that is accumulating messages beyond a specify threshold */
+    Events["SlowConsumer"] = "slow_consumer";
 })(Events || (exports.Events = Events = {}));
@@ -39,0 +41,0 @@ /**

lib/nats.js

@@ -104,2 +104,11 @@ "use strict";
         const sub = new protocol_1.SubscriptionImpl(this.protocol, subject, opts);
+        if (typeof opts.callback !== "function" && typeof opts.slow === "number") {
+            const subj = sub.getSubject();
+            sub.setSlowNotificationFn(opts.slow, (pending) => {
+                this.protocol.dispatchStatus({
+                    type: core_1.Events.SlowConsumer,
+                    data: `subscription (${sub.sid}) ${subj} is slow: msgs ${pending}`,
+                });
+            });
+        }
         this.protocol.subscribe(sub);
@@ -106,0 +115,0 @@ return sub;

lib/protocol.d.ts

@@ -37,2 +37,9 @@ import type { Transport } from "./transport";
 }
+declare class SlowNotifier {
+    slow: number;
+    cb: (pending: number) => void;
+    notified: boolean;
+    constructor(slow: number, cb: (pending: number) => void);
+    maybeNotify(pending: number): void;
+}
 export declare class SubscriptionImpl extends QueuedIteratorImpl<Msg> implements Subscription {
@@ -51,3 +58,5 @@ sid: number;
     requestSubject?: string;
+    slow?: SlowNotifier;
     constructor(protocol: ProtocolHandler, subject: string, opts?: SubscriptionOptions);
+    setSlowNotificationFn(slow: number, fn?: (pending: number) => void): void;
     callback(err: Error | null, msg: Msg): void;
@@ -154,1 +163,2 @@ close(): void;
 }
+export {};

lib/protocol.js

@@ -69,2 +69,25 @@ "use strict";
 exports.Connect = Connect;
+class SlowNotifier {
+    slow;
+    cb;
+    notified;
+    constructor(slow, cb) {
+        this.slow = slow;
+        this.cb = cb;
+        this.notified = false;
+    }
+    maybeNotify(pending) {
+        // if we are below the threshold reset the ability to notify
+        if (pending <= this.slow) {
+            this.notified = false;
+        }
+        else {
+            if (!this.notified) {
+                // crossed the threshold, notify and silence.
+                this.cb(pending);
+                this.notified = true;
+            }
+        }
+    }
+}
 class SubscriptionImpl extends queued_iterator_1.QueuedIteratorImpl {
@@ -83,2 +106,3 @@ sid;
     requestSubject;
+    slow;
     constructor(protocol, subject, opts = {}) {
@@ -117,5 +141,17 @@ super();
     }
+    setSlowNotificationFn(slow, fn) {
+        this.slow = undefined;
+        if (fn) {
+            if (this.noIterator) {
+                throw new Error("callbacks don't support slow notifications");
+            }
+            this.slow = new SlowNotifier(slow, fn);
+        }
+    }
     callback(err, msg) {
         this.cancelTimeout();
         err ? this.stop(err) : this.push(msg);
+        if (!err && this.slow) {
+            this.slow.maybeNotify(this.getPending());
+        }
     }
@@ -122,0 +158,0 @@ close() {

lib/version.d.ts

@@ -1,1 +0,1 @@
-export declare const version = "3.0.0-36";
+export declare const version = "3.0.0-37";

lib/version.js

@@ -5,3 +5,3 @@ "use strict";
 // This file is generated - do not edit
-exports.version = "3.0.0-36";
+exports.version = "3.0.0-37";
 //# sourceMappingURL=version.js.map

package.json

 {
   "name": "@nats-io/nats-core",
-  "version": "3.0.0-36",
+  "version": "3.0.0-37",
   "files": [
@@ -5,0 +5,0 @@ "lib/",

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

Improved metrics

Total package byte prevSize

543876

increased by0.67%

Lines of code

8542

increased by0.79%

No dependency changes