fix: log issue with non-json payload, #6

README.md

@@ -81,8 +81,8 @@ app.connectMicroservice<MicroserviceOptions>({
 import { Injectable } from '@nestjs/common';
 import { EventPattern, Ctx, Payload } from '@nestjs/microservices';
 
-@Injectable()
-export class SomeService {
+@Controller()
+export class AppController {
 
   @EventPattern({
     topic: 'topic1',

src/responder/inbound-message-deserializer.ts

@@ -5,21 +5,21 @@ export class InboundMessageDeserializer implements ConsumerDeserializer {
   private readonly logger = new Logger('InboundMessageIdentityDeserializer');
 
   deserialize(value: any, options?: Record<string, any>): IncomingRequest {
-    let msg;
+    let json;
+
     try {
-      msg = value.json();
-    } catch (err) {
-      msg = value;
-    }
+      json = value.json();
+    } catch (err) {}
 
     this.logger.verbose(
       `<<-- deserializing inbound message:\n
-${JSON.stringify(msg)}
+${(json && JSON.stringify(json)) || value.body}
 \n\twith options: ${JSON.stringify(options)}`,
     );
+
     return {
-      id: msg.id,
-      data: msg,
+      id: value.id,
+      data: json || value.body,
       pattern: options,
     };
   }

src/responder/server-nsq.ts

@@ -136,6 +136,7 @@ export class ServerNsq extends Server implements CustomTransportStrategy {
               `consumer reader failed to process message with error: ${err.message}, topic: ${topic}, channel: ${channel}`,
             );
             msg.requeue(consumerOptions.requeueDelay || DEFAULT_REQUEUE_DELAY);
+            return;
           }
 
           const wrappedErrObservable = source.pipe(

test/app.e2e-spec.ts

@@ -72,6 +72,7 @@ describe('AppController (e2e)', () => {
       discardCalled = mockDiscardHdlr.mock.calls.length;
     }
     expect(discardCalled).toBe(1);
+    mockDiscardHdlr.mockClear();
   }, 10000);
 
   it('should be able to dispatch event', async () => {