Fix hook cancellation stuff.

src/modules/consumer.js

@@ -206,7 +206,7 @@ class Consumer {
           content: body,
         });
         if (!shouldContinue) {
-          await this._rejectMessageAfterProcess(channel, queue, msg, body);
+          await this._rejectMessageAfterProcess(channel, queue, msg, body, false);
           return;
         }
         const res = await callback(body, msg.properties);
@@ -219,7 +219,7 @@ class Consumer {
           params: { queue, message: messageString },
         });
 
-        await this._rejectMessageAfterProcess(channel, queue, msg, body, error);
+        await this._rejectMessageAfterProcess(channel, queue, msg, body, this._connection.config.requeue, error);
         return;
       }
 
@@ -254,9 +254,14 @@ class Consumer {
   }
 
   /** @private */
-  async _rejectMessageAfterProcess(channel, queue, msg, parsedBody, error) {
+  async _rejectMessageAfterProcess(channel, queue, msg, parsedBody, requeue, error) {
     try {
-      channel.reject(msg, this._connection.config.requeue);
+      channel.reject(msg, requeue);
+
+      if (!requeue) {
+        // If not requeued and message will be removed from the queue, return rpc error response if needed.
+        await this.checkRpc(msg.properties, queue, error instanceof Error ? { error } : undefined);
+      }
     } catch (rejectError) {
       await this.hooks.trigger(this, ConsumerHooks.afterProcessMessageEvent, {
         queue,

src/modules/hooks/base_hooks.js

@@ -55,7 +55,7 @@ module.exports = class BaseHooks {
       return;
     }
 
-    const callbacks = this._getCallbacks();
+    const callbacks = this._getCallbacks(event);
     if (callbacks) {
       return;
     }
@@ -69,7 +69,7 @@ module.exports = class BaseHooks {
    * @private
    */
   _manyOff(event, callbacks) {
-    const registered = this._getCallbacks();
+    const registered = this._getCallbacks(event);
     if (registered) {
       return;
     }

src/modules/hooks/producer_hooks.js

@@ -46,7 +46,7 @@ class ProducerHooks extends BaseHooks {
    * - result - The value return from publication. If rpc, will be the deserialized object.
    * - error - The error object in case the publication failed, or received an erroneous RPC response.
    * - shouldRetry - If received an error, 'true' if the publication will be retried (if retry configured).
-   * In case the hook callback was called with an error, it can return false in order to abort any further publish retries (if retry is configured).
+   * In case the hook callback was called with an error, it can return `false` in order to abort any further publish retries (if retry is configured).
    * @param {Function | Function[]} callback A callback or callbacks array to register.
    */
   afterPublish(callback) {

src/modules/producer.js

@@ -256,11 +256,11 @@ class Producer {
         properties: settings,
         currentRetry: currentRetryNumber,
       });
-      if (!shouldContinue) {
-        return null;
-      }
 
-      const result = await this.checkRpc(queue, parsedMessage, settings);
+      let result;
+      if (shouldContinue) {
+        result = await this.checkRpc(queue, parsedMessage, settings);
+      }
 
       await this.hooks.trigger(this, ProducerHooks.afterPublish, {
         queue,

test/disconnect-spec.js

@@ -80,8 +80,9 @@ describe('disconnections', function () {
       });
 
       it('calls event hooks on connecting', async () => {
+        const hookTestsQueue = 'disco:test:hooks';
         const consumedAllPromise = pDefer();
-        await arnavmq.consumer.consume(queue, () => {
+        await arnavmq.consumer.consume(hookTestsQueue, () => {
           consumedAllPromise.resolve();
         });
 
@@ -91,8 +92,7 @@ describe('disconnections', function () {
         arnavmq.hooks.connection.afterConnect(afterConnectHook);
 
         await docker.connectNetwork();
-        await arnavmq.producer.produce(queue);
-
+        await arnavmq.producer.produce(hookTestsQueue);
         await consumedAllPromise.promise;
 
         sinon.assert.called(beforeConnectHook);