FIX 265

nats-base-client/protocol.ts

@@ -415,18 +415,16 @@ export class ProtocolHandler implements Dispatcher<ParserEvent> {
   async processError(m: Uint8Array) {
     const s = decode(m);
     const err = ProtocolHandler.toError(s);
-    const handled = this.subscriptions.handleError(err);
-    if (!handled) {
-      this.dispatchStatus({ type: Events.Error, data: err.code });
-    }
+    this.subscriptions.handleError(err);
+    this.dispatchStatus({ type: Events.Error, data: err.code });
     await this.handleError(err);
   }
 
   async handleError(err: NatsError) {
     if (err.isAuthError()) {
       this.handleAuthError(err);
     }
-    if (err.isPermissionError() || err.isProtocolError()) {
+    if (err.isProtocolError()) {
       await this._close(err);
     }
     this.lastError = err;

nats-base-client/subscription.ts

@@ -70,6 +70,7 @@ export class SubscriptionImpl extends QueuedIteratorImpl<Msg>
       // cleanup - they used break or return from the iterator
       // make sure we clean up, if they didn't call unsub
       this.iterClosed.then(() => {
+        this.closed.resolve();
         this.unsubscribe();
       });
     }

tests/auth_test.ts

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 The NATS Authors
+ * Copyright 2018-2022 The NATS Authors
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
@@ -15,22 +15,23 @@
 
 import {
   assertEquals,
+  assertRejects,
   fail,
 } from "https://deno.land/std@0.125.0/testing/asserts.ts";
 import {
   connect,
   credsAuthenticator,
   ErrorCode,
-  Events,
   jwtAuthenticator,
   nkeyAuthenticator,
   Status,
   StringCodec,
 } from "../src/mod.ts";
-import { assertErrorCode, Lock, NatsServer } from "./helpers/mod.ts";
+import { assertErrorCode, NatsServer } from "./helpers/mod.ts";
 import { deferred, nkeys } from "../nats-base-client/internal_mod.ts";
 import { NKeyAuth } from "../nats-base-client/authenticator.ts";
 import { assert } from "../nats-base-client/denobuffer.ts";
+import { cleanup, setup } from "./jstest_util.ts";
 
 const conf = {
   authorization: {
@@ -100,77 +101,142 @@ Deno.test("auth - un/pw", async () => {
   await ns.stop();
 });
 
-Deno.test("auth - sub permissions", async () => {
-  const ns = await NatsServer.start(conf);
-  const lock = Lock();
-  const nc = await connect(
-    { port: ns.port, user: "derek", pass: "foobar" },
+Deno.test("auth - sub no permissions keeps connection", async () => {
+  const { ns, nc } = await setup({
+    authorization: {
+      users: [{
+        user: "a",
+        password: "a",
+        permissions: { subscribe: "foo" },
+      }],
+    },
+  }, { user: "a", pass: "a", reconnect: false });
+
+  const errStatus = deferred<Status>();
+  const _ = (async () => {
+    for await (const s of nc.status()) {
+      errStatus.resolve(s);
+    }
+  })();
+
+  const cbErr = deferred<Error | null>();
+  const sub = nc.subscribe("bar", {
+    callback: (err, msg) => {
+      cbErr.resolve(err);
+    },
+  });
+
+  const v = await Promise.all([errStatus, cbErr, sub.closed]);
+  assertEquals(v[0].data, ErrorCode.PermissionsViolation);
+  assertEquals(
+    v[1]?.message,
+    "'Permissions Violation for Subscription to \"bar\"'",
   );
+  assertEquals(nc.isClosed(), false);
 
-  const status = nc.status();
+  await cleanup(ns, nc);
+});
 
-  let notifications = 0;
-  (async () => {
-    for await (const _s of status) {
-      notifications++;
-    }
-  })().then();
+Deno.test("auth - sub iterator no permissions keeps connection", async () => {
+  const { ns, nc } = await setup({
+    authorization: {
+      users: [{
+        user: "a",
+        password: "a",
+        permissions: { subscribe: "foo" },
+      }],
+    },
+  }, { user: "a", pass: "a", reconnect: false });
 
-  const done = nc.closed();
+  const errStatus = deferred<Status>();
+  const _ = (async () => {
+    for await (const s of nc.status()) {
+      errStatus.resolve(s);
+    }
+  })();
 
-  const sub = nc.subscribe("foo");
+  const iterErr = deferred<Error | null>();
+  const sub = nc.subscribe("bar");
   (async () => {
-    for await (const _m of sub) {
-      // ignoring
+    for await (const m of sub) {
     }
   })().catch((err) => {
-    lock.unlock();
-    assertErrorCode(err as Error, ErrorCode.PermissionsViolation);
-    notifications++;
+    iterErr.resolve(err);
   });
 
-  nc.publish("foo");
+  await nc.flush();
 
-  await Promise.all([lock, done]);
-  assertEquals(notifications, 1);
-  await ns.stop();
+  const v = await Promise.all([errStatus, iterErr, sub.closed]);
+  assertEquals(v[0].data, ErrorCode.PermissionsViolation);
+  assertEquals(
+    v[1]?.message,
+    "'Permissions Violation for Subscription to \"bar\"'",
+  );
+  assertEquals(sub.isClosed(), true);
+  assertEquals(nc.isClosed(), false);
+
+  await cleanup(ns, nc);
 });
 
-Deno.test("auth - pub perm", async () => {
-  const ns = await NatsServer.start(conf);
-  const lock = Lock();
-  const nc = await connect(
-    { port: ns.port, user: "derek", pass: "foobar" },
-  );
+Deno.test("auth - pub permissions keep connection", async () => {
+  const { ns, nc } = await setup({
+    authorization: {
+      users: [{
+        user: "a",
+        password: "a",
+        permissions: { publish: "foo" },
+      }],
+    },
+  }, { user: "a", pass: "a", reconnect: false });
 
-  const status = nc.status();
   const errStatus = deferred<Status>();
-  (async () => {
-    for await (const s of status) {
+  const _ = (async () => {
+    for await (const s of nc.status()) {
       errStatus.resolve(s);
     }
-  })().then();
+  })();
 
-  nc.closed().then((err) => {
-    assertErrorCode(err as Error, ErrorCode.PermissionsViolation);
-    lock.unlock();
-  });
+  nc.publish("bar");
 
-  const sub = nc.subscribe("bar");
-  const iter = (async () => {
-    for await (const _m of sub) {
-      fail("should not have been called");
+  const v = await errStatus;
+  assertEquals(v.data, ErrorCode.PermissionsViolation);
+  assertEquals(nc.isClosed(), false);
+
+  await cleanup(ns, nc);
+});
+
+Deno.test("auth - req permissions keep connection", async () => {
+  const { ns, nc } = await setup({
+    authorization: {
+      users: [{
+        user: "a",
+        password: "a",
+        permissions: { publish: "foo" },
+      }],
+    },
+  }, { user: "a", pass: "a", reconnect: false });
+
+  const errStatus = deferred<Status>();
+  const _ = (async () => {
+    for await (const s of nc.status()) {
+      console.log(s);
+      errStatus.resolve(s);
     }
   })();
 
-  nc.publish("bar");
+  await assertRejects(
+    async () => {
+      await nc.request("bar");
+    },
+    Error,
+    ErrorCode.Timeout,
+  );
 
-  await lock;
-  await iter;
-  const es = await errStatus;
-  assertEquals(es.type, Events.Error);
-  assertEquals(es.data, ErrorCode.PermissionsViolation);
-  await ns.stop();
+  const v = await errStatus;
+  assertEquals(v.data, ErrorCode.PermissionsViolation);
+  assertEquals(nc.isClosed(), false);
+
+  await cleanup(ns, nc);
 });
 
 Deno.test("auth - user and token is rejected", () => {

tests/heartbeats_test.ts

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2021 The NATS Authors
+ * Copyright 2020-2022 The NATS Authors
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
@@ -103,4 +103,6 @@ Deno.test("heartbeat - recovers from missed", async () => {
   hb.cancel();
   assertEquals(hb.timer, undefined);
   assert(status.length >= 6, `${status.length} >= 6`);
+  // some resources in the test runner are not always cleaned unless we wait a bit
+  await delay(500);
 });

tests/iterators_test.ts

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2021 The NATS Authors
+ * Copyright 2020-2022 The NATS Authors
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
@@ -12,7 +12,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-import { connect, createInbox, ErrorCode, NatsError } from "../src/mod.ts";
+import { connect, createInbox, ErrorCode } from "../src/mod.ts";
 import { assertEquals } from "https://deno.land/std@0.125.0/testing/asserts.ts";
 import { assertErrorCode, Lock, NatsServer } from "./helpers/mod.ts";
 import { assert } from "../nats-base-client/denobuffer.ts";
@@ -85,9 +85,6 @@ Deno.test("iterators - permission error breaks and closes", async () => {
   });
 
   await lock;
-  await nc.closed().then((err) => {
-    assertErrorCode(err as NatsError, ErrorCode.PermissionsViolation);
-  });
   await cleanup(ns, nc);
 });