feat!: removing handler_processing debugger in favour of status (#470)

Author: nicholasgriffintn

src/consumer.ts

@@ -164,12 +164,12 @@ export class Consumer extends TypedEventEmitter {
       Date.now() - this.stopRequestedAtTimestamp >
       this.pollingCompleteWaitTimeMs;
     if (exceededTimeout) {
-      logger.debug('waiting_for_polling_to_complete_timeout_exceeded');
+      this.emit('waiting_for_polling_to_complete_timeout_exceeded');
       this.emit('stopped');
       return;
     }
 
-    logger.debug('waiting_for_polling_to_complete');
+    this.emit('waiting_for_polling_to_complete');
     setTimeout(this.waitForPollingToComplete, 1000);
   }
 
@@ -304,20 +304,12 @@ export class Consumer extends TypedEventEmitter {
     response: ReceiveMessageCommandOutput
   ): Promise<void> {
     if (hasMessages(response)) {
-      const handlerProcessingDebugger = setInterval(() => {
-        logger.debug('handler_processing', {
-          detail: 'The handler is still processing the message(s)...'
-        });
-      }, 1000);
-
       if (this.handleMessageBatch) {
         await this.processMessageBatch(response.Messages);
       } else {
         await Promise.all(response.Messages.map(this.processMessage));
       }
 
-      clearInterval(handlerProcessingDebugger);
-
       this.emit('response_processed');
     } else if (response) {
       this.emit('empty');

src/types.ts

@@ -197,6 +197,14 @@ export interface Events {
    * Fired when an option is updated
    */
   option_updated: [UpdatableOptions, ConsumerOptions[UpdatableOptions]];
+  /**
+   * Fired when the Consumer is waiting for polling to complete before stopping.
+   */
+  waiting_for_polling_to_complete: [];
+  /**
+   * Fired when the Consumer has waited for polling to complete and is stopping due to a timeout.
+   */
+  waiting_for_polling_to_complete_timeout_exceeded: [];
 }
 
 export type AWSError = {

test/tests/consumer.test.ts

@@ -1221,7 +1221,7 @@ describe('Consumer', () => {
           VisibilityTimeout: 40
         })
       );
-      sandbox.assert.calledTwice(clearIntervalSpy);
+      sandbox.assert.calledOnce(clearIntervalSpy);
     });
 
     it('passes in the correct visibility timeout for long running batch handler functions', async () => {
@@ -1305,7 +1305,7 @@ describe('Consumer', () => {
           ]
         })
       );
-      sandbox.assert.calledTwice(clearIntervalSpy);
+      sandbox.assert.calledOnce(clearIntervalSpy);
     });
 
     it('emit error when changing visibility timeout fails', async () => {
@@ -1483,13 +1483,22 @@ describe('Consumer', () => {
     });
 
     it('waits for in-flight messages before emitting stopped (within timeout)', async () => {
+      sqs.send.withArgs(mockReceiveMessage).resolves({
+        Messages: [
+          { MessageId: '1', ReceiptHandle: 'receipt-handle-1', Body: 'body-1' }
+        ]
+      });
       const handleStop = sandbox.stub().returns(null);
       const handleResponseProcessed = sandbox.stub().returns(null);
+      const waitingForPollingComplete = sandbox.stub().returns(null);
+      const waitingForPollingCompleteTimeoutExceeded = sandbox
+        .stub()
+        .returns(null);
 
       // A slow message handler
-      handleMessage = sandbox.stub().callsFake(async () => {
-        await new Promise((resolve) => setTimeout(resolve, 5000));
-      });
+      handleMessage = sandbox
+        .stub()
+        .resolves(new Promise((resolve) => setTimeout(resolve, 5000)));
 
       consumer = new Consumer({
         queueUrl: QUEUE_URL,
@@ -1502,16 +1511,24 @@ describe('Consumer', () => {
 
       consumer.on('stopped', handleStop);
       consumer.on('response_processed', handleResponseProcessed);
+      consumer.on('waiting_for_polling_to_complete', waitingForPollingComplete);
+      consumer.on(
+        'waiting_for_polling_to_complete_timeout_exceeded',
+        waitingForPollingCompleteTimeoutExceeded
+      );
 
       consumer.start();
-      await clock.nextAsync();
+      await Promise.all([clock.tickAsync(1)]);
       consumer.stop();
 
       await clock.runAllAsync();
 
       sandbox.assert.calledOnce(handleStop);
       sandbox.assert.calledOnce(handleResponseProcessed);
       sandbox.assert.calledOnce(handleMessage);
+      assert(waitingForPollingComplete.callCount === 5);
+      assert(waitingForPollingCompleteTimeoutExceeded.callCount === 0);
+
       assert.ok(handleMessage.calledBefore(handleStop));
 
       // handleResponseProcessed is called after handleMessage, indicating
@@ -1520,13 +1537,22 @@ describe('Consumer', () => {
     });
 
     it('waits for in-flight messages before emitting stopped (timeout reached)', async () => {
+      sqs.send.withArgs(mockReceiveMessage).resolves({
+        Messages: [
+          { MessageId: '1', ReceiptHandle: 'receipt-handle-1', Body: 'body-1' }
+        ]
+      });
       const handleStop = sandbox.stub().returns(null);
       const handleResponseProcessed = sandbox.stub().returns(null);
+      const waitingForPollingComplete = sandbox.stub().returns(null);
+      const waitingForPollingCompleteTimeoutExceeded = sandbox
+        .stub()
+        .returns(null);
 
       // A slow message handler
-      handleMessage = sandbox.stub().callsFake(async () => {
-        await new Promise((resolve) => setTimeout(resolve, 5000));
-      });
+      handleMessage = sandbox
+        .stub()
+        .resolves(new Promise((resolve) => setTimeout(resolve, 5000)));
 
       consumer = new Consumer({
         queueUrl: QUEUE_URL,
@@ -1539,16 +1565,23 @@ describe('Consumer', () => {
 
       consumer.on('stopped', handleStop);
       consumer.on('response_processed', handleResponseProcessed);
+      consumer.on('waiting_for_polling_to_complete', waitingForPollingComplete);
+      consumer.on(
+        'waiting_for_polling_to_complete_timeout_exceeded',
+        waitingForPollingCompleteTimeoutExceeded
+      );
 
       consumer.start();
-      await clock.nextAsync();
+      await Promise.all([clock.tickAsync(1)]);
       consumer.stop();
 
       await clock.runAllAsync();
 
       sandbox.assert.calledOnce(handleStop);
       sandbox.assert.calledOnce(handleResponseProcessed);
       sandbox.assert.calledOnce(handleMessage);
+      sandbox.assert.calledOnce(waitingForPollingComplete);
+      sandbox.assert.calledOnce(waitingForPollingCompleteTimeoutExceeded);
       assert(handleMessage.calledBefore(handleStop));
 
       // Stop was called before the message could be processed, because we reached timeout.
@@ -1768,33 +1801,5 @@ describe('Consumer', () => {
       sandbox.assert.calledWithMatch(loggerDebug, 'stopping');
       sandbox.assert.calledWithMatch(loggerDebug, 'stopped');
     });
-
-    it('logs a debug event while the handler is processing, for every second', async () => {
-      const loggerDebug = sandbox.stub(logger, 'debug');
-      const clearIntervalSpy = sinon.spy(global, 'clearInterval');
-
-      sqs.send.withArgs(mockReceiveMessage).resolves({
-        Messages: [
-          { MessageId: '1', ReceiptHandle: 'receipt-handle-1', Body: 'body-1' }
-        ]
-      });
-      consumer = new Consumer({
-        queueUrl: QUEUE_URL,
-        region: REGION,
-        handleMessage: () =>
-          new Promise((resolve) => setTimeout(resolve, 4000)),
-        sqs
-      });
-
-      consumer.start();
-      await Promise.all([clock.tickAsync(5000)]);
-      sandbox.assert.calledOnce(clearIntervalSpy);
-      consumer.stop();
-
-      sandbox.assert.callCount(loggerDebug, 15);
-      sandbox.assert.calledWith(loggerDebug, 'handler_processing', {
-        detail: 'The handler is still processing the message(s)...'
-      });
-    });
   });
 });