Merge pull request #33 from sourcefuse/GH-30

yeshamavani · web-flow · commit e6e8621b479d · 2023-10-19T13:15:13.000+05:30
fix(core): add support for generic and common consumer
diff --git a/package-lock.json b/package-lock.json
diff --git a/src/__tests__/acceptance/application.test.ts b/src/__tests__/acceptance/application.test.ts
@@ -19,6 +19,8 @@ describe('end-to-end', () => {
   let producerApp: Application;
   let startHandler: sinon.SinonStub;
   let stopHandler: sinon.SinonStub;
+  let commonHandler: sinon.SinonStub;
+  let genericHandler: sinon.SinonStub;
   let logger: ILogger;
   const kafkaClient = new KafkaClientStub();
   before(setupApplications);
@@ -29,6 +31,8 @@ describe('end-to-end', () => {
   afterEach(() => {
     startHandler.reset();
     stopHandler.reset();
+    commonHandler.reset();
+    genericHandler.reset();
   });
 
   describe('Acceptance: event stream with single topic', () => {
@@ -68,7 +72,42 @@ describe('end-to-end', () => {
       );
     });
 
-    it('should not handle a unspecified events', async () => {
+    it('should consume from a common consumer with multiple events for a single topic', async () => {
+      const producerInstance = producerApp.getSync<Producer<TestStream>>(
+        producerKey(Topics.First),
+      );
+      const reset = {
+        resetTime: new Date(),
+      };
+      const pause = {
+        pauseTime: new Date(),
+      };
+      await producerInstance.send(Events.reset, [reset]);
+      await producerInstance.send(Events.pause, [pause]);
+      sinon.assert.called(commonHandler);
+      expect(commonHandler.getCalls()[0].args[0]).to.be.deepEqual(
+        JSON.parse(JSON.stringify(reset)),
+      );
+      expect(commonHandler.getCalls()[1].args[0]).to.be.deepEqual(
+        JSON.parse(JSON.stringify(pause)),
+      );
+    });
+
+    it('should consume from a generic consumer without events for a single topic', async () => {
+      const producerInstance = producerApp.getSync<Producer<TestStream>>(
+        producerKey(Topics.First),
+      );
+      const close = {
+        closeTime: new Date(),
+      };
+      await producerInstance.send(Events.close, [close]);
+      sinon.assert.called(genericHandler);
+      expect(genericHandler.getCalls()[0].args[0]).to.be.deepEqual(
+        JSON.parse(JSON.stringify(close)),
+      );
+    });
+
+    it('should not handle an unspecified events', async () => {
       const warnStub = sinon.stub(logger, 'warn');
       const producerInstance = kafkaClient.producer();
       const payload = {
@@ -82,21 +121,21 @@ describe('end-to-end', () => {
         topic: Topics.First,
         messages: [payload],
       });
+
       sinon.assert.calledOnce(warnStub);
       expect(warnStub.lastCall.args[0]).to.equal(
-        `${KafkaErrorKeys.UnhandledEvent}: ${JSON.stringify({
-          topic: Topics.First,
-          message: payload,
-        })}`,
+        `${KafkaErrorKeys.HandleByGenericConsumer}:${undefined}`,
       );
     });
   });
 
   async function setupApplications() {
     startHandler = sinon.stub().callsFake(() => {});
     stopHandler = sinon.stub().callsFake(() => {});
+    commonHandler = sinon.stub().callsFake(() => {});
+    genericHandler = sinon.stub().callsFake(() => {});
     logger = {
-      warn: e => console.warn(e),
+      warn: e => console.log(e),
       log: e => console.log(e),
       error: e => console.error(e),
       info: e => console.info(e),
@@ -106,6 +145,8 @@ describe('end-to-end', () => {
       kafkaClient,
       startHandler,
       stopHandler,
+      commonHandler,
+      genericHandler,
       logger,
     );
     producerApp = await setupProducerApplication(kafkaClient);
diff --git a/src/__tests__/acceptance/fixtures/consumer/consumer-app.ts b/src/__tests__/acceptance/fixtures/consumer/consumer-app.ts
@@ -8,6 +8,8 @@ import {KafkaClientBindings} from '../../../../keys';
 import {KafkaClientStub} from '../../../stubs';
 import {StartConsumer} from './start-consumer.extension';
 import {StopConsumer} from './stop-consumer.extension';
+import {CommonConsumer} from './shared-consumer.extension';
+import {GenericConsumer} from './generic-consumer.extension';
 
 export class ConsumerApp extends BootMixin(
   ServiceMixin(RepositoryMixin(RestApplication)),
@@ -26,6 +28,8 @@ export class ConsumerApp extends BootMixin(
     this.component(KafkaClientComponent);
     this.service(StartConsumer);
     this.service(StopConsumer);
+    this.service(CommonConsumer);
+    this.service(GenericConsumer);
 
     this.projectRoot = __dirname;
     // Customize @loopback/boot Booter Conventions here
diff --git a/src/__tests__/acceptance/fixtures/consumer/generic-consumer.extension.ts b/src/__tests__/acceptance/fixtures/consumer/generic-consumer.extension.ts
@@ -0,0 +1,15 @@
+import {inject, injectable} from '@loopback/core';
+import {asConsumer} from '../../../../keys';
+import {TestStream} from '../stream';
+import {IGenericConsumer, StreamHandler} from '../../../../types';
+import {Topics} from '../topics.enum';
+import {Events} from '../events.enum';
+
+@injectable(asConsumer)
+export class GenericConsumer implements IGenericConsumer<TestStream> {
+  constructor(
+    @inject('eventHandler.generic')
+    public handler: StreamHandler<TestStream, Events>,
+  ) {}
+  topic: Topics.First = Topics.First;
+}
diff --git a/src/__tests__/acceptance/fixtures/consumer/shared-consumer.extension.ts b/src/__tests__/acceptance/fixtures/consumer/shared-consumer.extension.ts
@@ -0,0 +1,16 @@
+import {inject, injectable} from '@loopback/core';
+import {asConsumer} from '../../../../keys';
+import {TestStream} from '../stream';
+import {ISharedConsumer, StreamHandler} from '../../../../types';
+import {Topics} from '../topics.enum';
+import {Events} from '../events.enum';
+
+@injectable(asConsumer)
+export class CommonConsumer implements ISharedConsumer<TestStream> {
+  constructor(
+    @inject('eventHandler.common')
+    public handler: StreamHandler<TestStream, Events>,
+  ) {}
+  topic: Topics.First = Topics.First;
+  events = [Events.pause, Events.reset];
+}
diff --git a/src/__tests__/acceptance/fixtures/consumer/start-consumer.extension.ts b/src/__tests__/acceptance/fixtures/consumer/start-consumer.extension.ts
@@ -1,12 +1,13 @@
-import {consumer} from '../../../../decorators';
+import {injectable} from '@loopback/core';
 import {eventHandler} from '../../../../decorators/handler.decorator';
-import {StreamHandler} from '../../../../types';
+import {asConsumer} from '../../../../keys';
+import {IConsumer, StreamHandler} from '../../../../types';
 import {Events} from '../events.enum';
 import {TestStream} from '../stream';
 import {Topics} from '../topics.enum';
 
-@consumer<TestStream, Events.start>()
-export class StartConsumer {
+@injectable(asConsumer)
+export class StartConsumer implements IConsumer<TestStream, Events.start> {
   constructor(
     @eventHandler<TestStream>(Events.start)
     public handler: StreamHandler<TestStream, Events.start>,
diff --git a/src/__tests__/acceptance/fixtures/consumer/stop-consumer.extension.ts b/src/__tests__/acceptance/fixtures/consumer/stop-consumer.extension.ts
@@ -1,11 +1,12 @@
-import {consumer} from '../../../../decorators';
+import {injectable} from '@loopback/core';
 import {eventHandler} from '../../../../decorators/handler.decorator';
+import {asConsumer} from '../../../../keys';
 import {IConsumer, StreamHandler} from '../../../../types';
 import {Events} from '../events.enum';
 import {TestStream} from '../stream';
 import {Topics} from '../topics.enum';
 
-@consumer<TestStream, Events.stop>()
+@injectable(asConsumer)
 export class StopConsumer implements IConsumer<TestStream, Events.stop> {
   constructor(
     @eventHandler<TestStream>(Events.stop)
diff --git a/src/__tests__/acceptance/fixtures/events.enum.ts b/src/__tests__/acceptance/fixtures/events.enum.ts
@@ -1,4 +1,7 @@
 export enum Events {
   start = 'start',
   stop = 'stop',
+  reset = 'reset',
+  pause = 'pause',
+  close = 'close',
 }
diff --git a/src/__tests__/acceptance/fixtures/stream.ts b/src/__tests__/acceptance/fixtures/stream.ts
@@ -8,5 +8,8 @@ export interface TestStream extends IStreamDefinition {
   messages: {
     [Events.start]: StartEvent;
     [Events.stop]: StopEvent;
+    [Events.reset]: {};
+    [Events.pause]: {};
+    [Events.close]: {};
   };
 }
diff --git a/src/__tests__/acceptance/test-helper.ts b/src/__tests__/acceptance/test-helper.ts
@@ -13,6 +13,8 @@ export async function setupConsumerApplication(
   kafkaStub: KafkaClientStub,
   startHandler: StreamHandler<TestStream, Events.start>,
   stopHandler: StreamHandler<TestStream, Events.stop>,
+  commonHandler: StreamHandler<TestStream, Events>,
+  genericHandler: StreamHandler<TestStream, Events>,
   logger: ILogger,
 ): Promise<Application> {
   const restConfig = givenHttpServerConfig({});
@@ -29,6 +31,8 @@ export async function setupConsumerApplication(
   app
     .bind(eventHandlerKey<TestStream, Events.stop>(Events.stop))
     .to(stopHandler);
+  app.bind('eventHandler.common').to(commonHandler);
+  app.bind('eventHandler.generic').to(genericHandler);
   app.bind(LOGGER.LOGGER_INJECT).to(logger);
   await app.boot();
   await app.start();
diff --git a/src/__tests__/stubs/kafka.stub.ts b/src/__tests__/stubs/kafka.stub.ts
@@ -47,8 +47,14 @@ export class KafkaClientStub {
       disconnect: async () => {
         delete this.subscribers[id];
       },
-      subscribe: async (params: {topic: string}) => {
-        this.subscribers[id].topics.push(params.topic);
+      subscribe: async (params: {topic?: string; topics?: string[]}) => {
+        if (params.topic) {
+          this.subscribers[id].topics.push(params.topic);
+        } else if (params.topics) {
+          this.subscribers[id].topics.push(...params.topics);
+        } else {
+          // do nothing
+        }
       },
       run: async (params: {eachMessage: (message: AnyObject) => void}) => {
         this.subscribers[id].handler = params.eachMessage;
diff --git a/src/component.ts b/src/component.ts
@@ -36,7 +36,8 @@ export class KafkaClientComponent implements Component {
     }
     app
       .bind(KafkaClientBindings.ProducerFactory)
-      .toProvider(KafkaProducerFactoryProvider);
+      .toProvider(KafkaProducerFactoryProvider)
+      .inScope(BindingScope.SINGLETON);
 
     app.service(KafkaConsumerService);
 
diff --git a/src/error-keys.ts b/src/error-keys.ts
@@ -5,4 +5,6 @@ export enum KafkaErrorKeys {
   EventWithoutValue = 'EventWithoutValue',
   UnhandledEvent = 'Unhandled Event',
   PublishFailed = 'Publish Failed',
+  MultipleGenericConsumers = 'Multiple Generic Consumers',
+  HandleByGenericConsumer = 'Handled By Generic Consumer',
 }
diff --git a/src/keys.ts b/src/keys.ts
@@ -1,8 +1,9 @@
 import {BindingKey, BindingTemplate, extensionFor} from '@loopback/core';
-import {ConsumerConfig, Kafka, ProducerConfig} from 'kafkajs';
+import {Kafka, ProducerConfig} from 'kafkajs';
 import {KafkaClientComponent} from './component';
 import {KafkaConsumerService} from './services/kafka-consumer.service';
 import {
+  ConsumerConfig,
   IStreamDefinition,
   Producer,
   ProducerFactoryType,
diff --git a/src/providers/kafka-producer-factory.provider.ts b/src/providers/kafka-producer-factory.provider.ts
@@ -1,6 +1,6 @@
 import {inject, Provider} from '@loopback/core';
 import {ILogger, LOGGER} from '@sourceloop/core';
-import {Kafka, ProducerConfig} from 'kafkajs';
+import {CompressionTypes, Kafka, ProducerConfig} from 'kafkajs';
 import {KafkaErrorKeys} from '../error-keys';
 import {KafkaClientBindings} from '../keys';
 import {IStreamDefinition, ProducerFactoryType} from '../types';
@@ -33,6 +33,7 @@ export class KafkaProducerFactoryProvider<T extends IStreamDefinition>
             await producer.connect();
             await producer.send({
               topic: topic,
+              compression: CompressionTypes.GZIP,
               messages: payload.map(message => ({
                 key,
                 value: JSON.stringify({
diff --git a/src/services/kafka-consumer.service.ts b/src/services/kafka-consumer.service.ts
diff --git a/src/types.ts b/src/types.ts

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,7 @@`
`1`	`1`	`export enum Events {`
`2`	`2`	`start = 'start',`
`3`	`3`	`stop = 'stop',`
	`4`	`+ reset = 'reset',`
	`5`	`+ pause = 'pause',`
	`6`	`+ close = 'close',`
`4`	`7`	`}`
Original file line number	Diff line number	Diff line change
`@@ -8,5 +8,8 @@ export interface TestStream extends IStreamDefinition {`
`8`	`8`	`messages: {`
`9`	`9`	`[Events.start]: StartEvent;`
`10`	`10`	`[Events.stop]: StopEvent;`
	`11`	`+ [Events.reset]: {};`
	`12`	`+ [Events.pause]: {};`
	`13`	`+ [Events.close]: {};`
`11`	`14`	`};`
`12`	`15`	`}`