Fix: channel.get() does not yield errors to the callback (#832)

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Change log for amqplib
 
+## Unreleased
+- Fix channel.get() not invoking callback with error on channel close; previously only an error event was emitted (fixes #832). **Note:** if you use the callback API, ensure your channel.get() callbacks handle errors — they will now be invoked in error cases where previously they were not. If you use the promise API, the returned promise now rejects with a proper Error object (with .code, .classId and .methodId properties) rather than a raw close frame.
+
 ## v1.0.5
 - Fix ConfirmChannel callbacks silently dropped on channel close when some publishes had no callback (fixes #191)
 

lib/callback_model.js

@@ -1,8 +1,7 @@
 const defs = require('./defs');
 const EventEmitter = require('node:events');
-const BaseChannel = require('./channel').BaseChannel;
-const acceptMessage = require('./channel').acceptMessage;
 const Args = require('./api_args');
+const { BaseChannel, acceptMessage, convertCloseFrameToError } = require('./channel');
 
 class CallbackModel extends EventEmitter {
   constructor(connection) {
@@ -179,17 +178,17 @@ class Channel extends BaseChannel {
     const fields = Args.get(queue, options);
     const cb = callbackWrapper(this, cb0);
     this.sendOrEnqueue(defs.BasicGet, fields, (err, f) => {
-      if (err === null) {
-        if (f.id === defs.BasicGetEmpty) {
-          cb(null, false);
-        } else if (f.id === defs.BasicGetOk) {
-          this.handleMessage = acceptMessage((m) => {
-            m.fields = f.fields;
-            cb(null, m);
-          });
-        } else {
-          cb(new Error(`Unexpected response to BasicGet: ${inspect(f)}`));
-        }
+      if (err instanceof Error) return cb(err);
+      if (err) return cb(convertCloseFrameToError(defs.BasicGet, err));
+      if (f.id === defs.BasicGetEmpty) {
+        cb(null, false);
+      } else if (f.id === defs.BasicGetOk) {
+        this.handleMessage = acceptMessage((m) => {
+          m.fields = f.fields;
+          cb(null, m);
+        });
+      } else {
+        cb(new Error(`Unexpected response to BasicGet: ${inspect(f)}`));
       }
     });
     return this;

lib/channel.js

@@ -116,19 +116,7 @@ class Channel extends EventEmitter {
       else if (err instanceof Error) return cb(err);
       // A close frame will be given if this is the RPC awaiting reply
       // and the channel is closed by the server
-      else {
-        // otherwise, it's a close frame
-        const closeReason = (err.fields.classId << 16) + err.fields.methodId;
-        const e =
-          method === closeReason
-            ? fmt('Operation failed: %s; %s', methodName(method), closeMsg(err))
-            : fmt('Channel closed by server: %s', closeMsg(err));
-        const closeFrameError = new Error(e);
-        closeFrameError.code = err.fields.replyCode;
-        closeFrameError.classId = err.fields.classId;
-        closeFrameError.methodId = err.fields.methodId;
-        return cb(closeFrameError);
-      }
+      else return cb(convertCloseFrameToError(method, err));
     }
 
     this.sendOrEnqueue(method, fields, reply);
@@ -428,6 +416,19 @@ function acceptMessage(continuation) {
   }
 }
 
+function convertCloseFrameToError(method, closeFrame) {
+  const closeReason = (closeFrame.fields.classId << 16) + closeFrame.fields.methodId;
+  const e =
+    method === closeReason
+      ? fmt('Operation failed: %s; %s', methodName(method), closeMsg(closeFrame))
+      : fmt('Channel closed by server: %s', closeMsg(closeFrame));
+  const error = new Error(e);
+  error.code = closeFrame.fields.replyCode;
+  error.classId = closeFrame.fields.classId;
+  error.methodId = closeFrame.fields.methodId;
+  return error;
+}
+
 // This adds just a bit more stuff useful for the APIs, but not
 // low-level machinery.
 class BaseChannel extends Channel {
@@ -471,3 +472,4 @@ class BaseChannel extends Channel {
 module.exports.acceptMessage = acceptMessage;
 module.exports.BaseChannel = BaseChannel;
 module.exports.Channel = Channel;
+module.exports.convertCloseFrameToError = convertCloseFrameToError;

lib/channel_model.js

@@ -1,9 +1,8 @@
 const EventEmitter = require('node:events');
-const promisify = require('node:util').promisify;
+const { promisify } = require('node:util');
 const defs = require('./defs');
-const { BaseChannel } = require('./channel');
-const { acceptMessage } = require('./channel');
 const Args = require('./api_args');
+const { BaseChannel, acceptMessage, convertCloseFrameToError } = require('./channel');
 const { inspect } = require('./format');
 
 class ChannelModel extends EventEmitter {
@@ -159,7 +158,8 @@ class Channel extends BaseChannel {
     const fields = Args.get(queue, options);
     return new Promise((resolve, reject) => {
       this.sendOrEnqueue(defs.BasicGet, fields, (err, f) => {
-        if (err) return reject(err);
+        if (err instanceof Error) return reject(err);
+        if (err) return reject(convertCloseFrameToError(defs.BasicGet, err));
         if (f.id === defs.BasicGetEmpty) {
           return resolve(false);
         } else if (f.id === defs.BasicGetOk) {

test/callback_api.test.js

@@ -401,39 +401,18 @@ describe('Callback API', () => {
       });
     });
 
-    /*
-    When this test was refactored as part of the move to the node test framework
-    it failed because only the domain error handler was ever invoked and not the
-    channel.get error callback.
-
-    The original test passed because twice.first (refactored to use util.latch) short circuits on error rather
-    than waiting for twice.second to be invoked. I have verified that the pre-refactored
-    amqplib does not invoke the channel.get callback when the queue does not exist.
-    See lib/callback_model.js
-    */
     error_test('Get from non-queue invokes error', (ch, done, dom) => {
       const decrementLatch = latch(2, () => done());
       dom.on('error', (err) => {
         assert.match(err.message, /404 \(NOT-FOUND\)/);
-        decrementLatch(err);
+        decrementLatch();
       });
       ch.get('', {}, (err) => {
         assert.match(err.message, /404 \(NOT-FOUND\)/)
-        decrementLatch(err)
+        decrementLatch()
       });
     });
 
-    /*
-    When this test was refactored as part of the move to the node test framework
-    it failed because only the domain error handler was ever invoked and not the
-    channel.consume error callback. Unlike the above channel.get test,
-    channel.consume does invoke the channel.consume callback with an error, but
-    the original test did not supply one, mistakenly providing a message handler
-    function instead.
-
-    The original test passed because twice.first (refactored to use util.latch) short circuits on error
-    rather than waiting for twice.second to be invoked.
-    */
     error_test('Consume from non-queue invokes error', (ch, done, dom) => {
       const decrementLatch = latch(2, done);
       dom.on('error', (err) => {

test/channel_api.test.js

@@ -176,6 +176,48 @@ describe('sendMessage', () => {
   });
 });
 
+describe('get', () => {
+  it('returns false when queue is empty', () => {
+    return withChannel((ch) =>
+      ch.assertQueue('test.get-empty', QUEUE_OPTS)
+        .then(() => ch.purgeQueue('test.get-empty'))
+        .then(() => ch.get('test.get-empty', { noAck: true }))
+        .then((m) => assert.strictEqual(false, m))
+    );
+  });
+
+  it('returns a message when queue has messages', () => {
+    const msg = randomString();
+    return withChannel((ch) =>
+      ch.assertQueue('test.get-message', QUEUE_OPTS)
+        .then(() => ch.purgeQueue('test.get-message'))
+        .then(() => {
+          ch.sendToQueue('test.get-message', Buffer.from(msg));
+          return waitForMessages('test.get-message');
+        })
+        .then(() => ch.get('test.get-message', { noAck: true }))
+        .then((m) => {
+          assert(m);
+          assert.equal(msg, m.content.toString());
+        })
+    );
+  });
+
+  it('rejects with a proper error when queue does not exist', () => {
+    return withChannel((ch) => {
+      ch.once('error', ignore);
+      return assert.rejects(() => ch.get('', {}), (err) => {
+        assert(err instanceof Error);
+        assert.match(err.message, /NOT_FOUND/);
+        assert.strictEqual(404, err.code);
+        assert.strictEqual(60, err.classId);
+        assert.strictEqual(70, err.methodId);
+        return true;
+      });
+    });
+  });
+});
+
 describe('binding, consuming', () => {
   // bind, publish, get
   it('route message', () => {