Merge branch 'main' into fix-134

nats-io · Nov 13, 2024 · 46f17ff · 46f17ff
2 parents 18d8b49 + ab296de
commit 46f17ff
Show file tree

Hide file tree

Showing 4 changed files with 17 additions and 11 deletions.
diff --git a/core/src/core.ts b/core/src/core.ts
@@ -59,6 +59,12 @@ export type ForceReconnectStatus = {
   type: "forceReconnect";
 };
 
+export type SlowConsumerStatus = {
+  type: "slowConsumer";
+  sub: Subscription;
+  pending: number;
+};
+
 export type Status =
   | DisconnectStatus
   | ReconnectStatus
@@ -68,6 +74,7 @@ export type Status =
   | ServerErrorStatus
   | ClientPingStatus
   | StaleConnectionStatus
+  | SlowConsumerStatus
   | ForceReconnectStatus;
 
 export type MsgCallback<T> = (err: Error | null, msg: T) => void;

diff --git a/core/src/nats.ts b/core/src/nats.ts
@@ -41,7 +41,7 @@ import type {
   Subscription,
   SubscriptionOptions,
 } from "./core.ts";
-import { createInbox, Events, RequestStrategy } from "./core.ts";
+import { createInbox, RequestStrategy } from "./core.ts";
 import { errors, InvalidArgumentError, TimeoutError } from "./errors.ts";
 
 export class NatsConnectionImpl implements NatsConnection {
@@ -130,11 +130,11 @@ export class NatsConnectionImpl implements NatsConnection {
     const sub = new SubscriptionImpl(this.protocol, subject, opts);
 
     if (typeof opts.callback !== "function" && typeof opts.slow === "number") {
-      const subj = sub.getSubject();
       sub.setSlowNotificationFn(opts.slow, (pending: number) => {
         this.protocol.dispatchStatus({
-          type: Events.SlowConsumer,
-          data: `subscription (${sub.sid}) ${subj} is slow: msgs ${pending}`,
+          type: "slowConsumer",
+          sub,
+          pending,
         });
       });
     }

diff --git a/core/tests/basics_test.ts b/core/tests/basics_test.ts
@@ -28,7 +28,6 @@ import {
   deferred,
   delay,
   Empty,
-  Events,
   Feature,
   headers,
   isIP,
@@ -1547,7 +1546,7 @@ Deno.test("basics - slow", async () => {
   (async () => {
     for await (const m of nc.status()) {
       //@ts-ignore: test
-      if (m.type === Events.SlowConsumer) {
+      if (m.type === "slowConsumer") {
         console.log(m);
         slow++;
       }

diff --git a/core/tests/timeout_test.ts b/core/tests/timeout_test.ts
@@ -17,27 +17,27 @@ import {
   assertRejects,
   assertStringIncludes,
 } from "jsr:@std/assert";
-import { connect } from "./connect.ts";
 import { createInbox, Empty, errors } from "../src/internal_mod.ts";
+import { cleanup, setup } from "../../test_helpers/mod.ts";
 
 Deno.test("timeout - request noMux stack is useful", async () => {
-  const nc = await connect({ servers: "demo.nats.io" });
+  const { ns, nc } = await setup();
   const subj = createInbox();
   const err = await assertRejects(() => {
     return nc.request(subj, Empty, { noMux: true, timeout: 250 });
   }, errors.RequestError);
   assertInstanceOf(err.cause, errors.NoRespondersError);
   assertStringIncludes((err as Error).stack || "", "timeout_test");
-  await nc.close();
+  await cleanup(ns, nc);
 });
 
 Deno.test("timeout - request stack is useful", async () => {
-  const nc = await connect({ servers: "demo.nats.io" });
+  const { ns, nc } = await setup();
   const subj = createInbox();
   const err = await assertRejects(() => {
     return nc.request(subj, Empty, { timeout: 250 });
   }, errors.RequestError);
   assertInstanceOf(err.cause, errors.NoRespondersError);
   assertStringIncludes((err as Error).stack || "", "timeout_test");
-  await nc.close();
+  await cleanup(ns, nc);
 });