Moves kafka logger into its own helper.

Author: mkaufmaner

packages/microservices/client/client-kafka.ts

@@ -10,7 +10,7 @@ import {
 import { ReadPacket, KafkaOptions, WritePacket, PacketId } from '../interfaces';
 import { ClientProxy } from './client-proxy';
 
-import { KafkaSerializer, KafkaRoundRobinByTimePartitionAssigner } from '../helpers';
+import { KafkaSerializer, KafkaRoundRobinByTimePartitionAssigner, KafkaLogger } from '../helpers';
 
 import {
   KafkaConfig,
@@ -21,7 +21,6 @@ import {
   EachMessagePayload,
   Message,
   KafkaMessage,
-  logLevel,
   ConsumerGroupJoinEvent
 } from '../external/kafka.interface';
 import { KafkaHeaders } from '../enums';
@@ -118,34 +117,10 @@ export class ClientKafka extends ClientProxy {
   }
 
   public createClient<T = any>(): T {
-    const kafkaLogger = kafkaLogLevel => ({namespace, level, label, log}) => {
-      let loggerMethod: string;
-
-      switch (level) {
-        case logLevel.ERROR:
-        case logLevel.NOTHING:
-          loggerMethod = 'error';
-          break;
-        case logLevel.WARN:
-          loggerMethod = 'warn';
-          break;
-        case logLevel.INFO:
-          loggerMethod = 'log';
-          break;
-        case logLevel.DEBUG:
-        default:
-          loggerMethod = 'debug';
-          break;
-      }
-
-      const { message, ...others } = log;
-      this.logger[loggerMethod](`${label} [${namespace}] ${message} ${JSON.stringify(others)}`);
-    };
-
     return new kafkaPackage.Kafka(Object.assign(this.options.client || {}, {
       clientId: this.clientId,
       brokers: this.brokers,
-      logCreator: kafkaLogger,
+      logCreator: KafkaLogger.bind(null, this.logger),
     }) as KafkaConfig);
   }
 

packages/microservices/helpers/index.ts

@@ -1,3 +1,4 @@
 export * from './json-socket';
 export * from './kafka-serializer';
+export * from './kafka-logger';
 export * from './kafka-round-robin-by-time-partition-assigner';

packages/microservices/helpers/kafka-logger.ts

@@ -0,0 +1,27 @@
+import {
+  logLevel
+} from '../external/kafka.interface';
+
+export const KafkaLogger = (logger: any, kafkaLogLevel) => ({namespace, level, label, log}) => {
+  let loggerMethod: string;
+
+  switch (level) {
+    case logLevel.ERROR:
+    case logLevel.NOTHING:
+      loggerMethod = 'error';
+      break;
+    case logLevel.WARN:
+      loggerMethod = 'warn';
+      break;
+    case logLevel.INFO:
+      loggerMethod = 'log';
+      break;
+    case logLevel.DEBUG:
+    default:
+      loggerMethod = 'debug';
+      break;
+  }
+
+  const { message, ...others } = log;
+  logger[loggerMethod](`${label} [${namespace}] ${message} ${JSON.stringify(others)}`);
+};

packages/microservices/server/server-kafka.ts

@@ -21,7 +21,7 @@ import { CustomTransportStrategy, KafkaOptions, ReadPacket, PacketId, WritePacke
 import { KafkaHeaders } from '../enums';
 import { Server } from './server';
 
-import { KafkaSerializer } from '../helpers/kafka-serializer';
+import { KafkaSerializer, KafkaLogger } from '../helpers';
 
 let kafkaPackage: any = {};
 
@@ -79,34 +79,10 @@ export class ServerKafka extends Server implements CustomTransportStrategy {
   }
 
   public createClient<T = any>(): T {
-    const kafkaLogger = kafkaLogLevel => ({namespace, level, label, log}) => {
-      let loggerMethod: string;
-
-      switch (level) {
-        case logLevel.ERROR:
-        case logLevel.NOTHING:
-          loggerMethod = 'error';
-          break;
-        case logLevel.WARN:
-          loggerMethod = 'warn';
-          break;
-        case logLevel.INFO:
-          loggerMethod = 'log';
-          break;
-        case logLevel.DEBUG:
-        default:
-          loggerMethod = 'debug';
-          break;
-      }
-
-      const { message, ...others } = log;
-      this.logger[loggerMethod](`${label} [${namespace}] ${message} ${JSON.stringify(others)}`);
-    };
-
     return new kafkaPackage.Kafka(Object.assign(this.options.client || {}, {
       clientId: this.clientId,
       brokers: this.brokers,
-      logCreator: kafkaLogger,
+      logCreator: KafkaLogger.bind(null, this.logger),
     }) as KafkaConfig);
   }
 

packages/microservices/test/helpers/kafka-logger.spec.ts

@@ -0,0 +1,93 @@
+import { expect } from 'chai';
+import * as sinon from 'sinon';
+import { KafkaLogger } from '../../helpers/kafka-logger';
+import { logLevel } from '../../external/kafka.interface';
+
+const namespace = 'namespace';
+const label = 'label';
+const entry = {
+  message: 'message',
+  other: {
+    stuff: 'here'
+  }
+};
+
+describe('kafka logger', () => {
+  let kafkaLogger: any;
+
+  let error: sinon.SinonSpy;
+  let warn: sinon.SinonSpy;
+  let log: sinon.SinonSpy;
+  let debug: sinon.SinonSpy;
+
+  beforeEach(() => {
+    // set
+    error = sinon.spy();
+    warn = sinon.spy();
+    log = sinon.spy();
+    debug = sinon.spy();
+
+    kafkaLogger = KafkaLogger({
+      error,
+      warn,
+      log,
+      debug
+    }, logLevel.DEBUG);
+  });
+
+  it('error', () => {
+    kafkaLogger({
+      namespace,
+      level: logLevel.ERROR,
+      label,
+      log: entry
+    });
+
+    expect(error.calledOnce).to.be.true;
+    expect(error.args[0][0]).to.eq('label [namespace] message {"other":{"stuff":"here"}}');
+  });
+
+  it('nothing', () => {
+    kafkaLogger({
+      namespace,
+      level: logLevel.NOTHING,
+      label,
+      log: entry
+    });
+
+    expect(error.calledOnce).to.be.true;
+  });
+
+  it('warn', () => {
+    kafkaLogger({
+      namespace,
+      level: logLevel.WARN,
+      label,
+      log: entry
+    });
+
+    expect(warn.calledOnce).to.be.true;
+  });
+
+  it('info', () => {
+    kafkaLogger({
+      namespace,
+      level: logLevel.INFO,
+      label,
+      log: entry
+    });
+
+    expect(log.calledOnce).to.be.true;
+  });
+
+  it('debug', () => {
+    kafkaLogger({
+      namespace,
+      level: logLevel.DEBUG,
+      label,
+      log: entry
+    });
+
+    expect(debug.calledOnce).to.be.true;
+  });
+});

packages/microservices/test/server/server-kafka.spec.ts

@@ -16,7 +16,6 @@ class NoopLogger extends Logger {
 
 describe('ServerKafka', () => {
   let server: ServerKafka;
-
   let callback: sinon.SinonSpy;
   let bindEventsStub: sinon.SinonStub;
   let connect: sinon.SinonSpy;
@@ -25,6 +24,7 @@ describe('ServerKafka', () => {
   let consumerStub: sinon.SinonStub;
   let producerStub: sinon.SinonStub;
   let client;
+
   beforeEach(() => {
     server = new ServerKafka({});
     callback = sinon.spy();
@@ -33,17 +33,17 @@ describe('ServerKafka', () => {
     run = sinon.spy();
 
     consumerStub = sinon.stub(server, 'consumer')
-      .callsFake( () => {
+      .callsFake(() => {
         return {
           connect,
           subscribe,
           run,
         };
       });
     producerStub = sinon.stub(server, 'producer')
-      .callsFake( () => {
+      .callsFake(() => {
         return {
-          connect,
+          connect
         };
       });
     client = {
@@ -78,16 +78,24 @@ describe('ServerKafka', () => {
   };
 
   describe('close', () => {
+    const consumer = {disconnect: sinon.spy()};
+    const producer = {disconnect: sinon.spy()};
+    beforeEach(() => {
+      (server as any).consumer = consumer;
+      (server as any).producer = producer;
+    });
     it('should close server', () => {
       server.close();
+
+      expect(consumer.disconnect.calledOnce).to.be.true;
+      expect(producer.disconnect.calledOnce).to.be.true;
       expect(server.consumer).to.be.null;
       expect(server.producer).to.be.null;
       expect(server.client).to.be.null;
     });
   });
 
   describe('listen', () => {
-
     it('should call "bindEvents"', async () => {
       bindEventsStub = sinon
         .stub(server, 'bindEvents')