fix(core): add subscription of generic topics

GH-0

Author: akshatdubeysf

README.md

@@ -23,7 +23,6 @@
 </a>
 </p>
 
-
 ## Overview
 
 A Kafka Client for Loopback4 built on top of [KafkaJS](https://kafka.js.org/).
@@ -97,7 +96,13 @@ export class TestStream implements IStreamDefinition {
 
 ### Consumer
 
-A Consumer is a [`loopback extension`](https://loopback.io/doc/en/lb4/Extension-point-and-extensions.html) that is used by the [`KafkaConsumerService`](./src/services/kafka-consumer.service.ts) to initialize consumers. It must implement the `IConsumer` interface and should be using the `@consumer()` decorator. If you want the consumers to start at the start of your application, you should pass the `initObservers` config to the Component configuration.
+A Consumer is a [`loopback extension`](https://loopback.io/doc/en/lb4/Extension-point-and-extensions.html) that is used by the [`KafkaConsumerService`](./src/services/kafka-consumer.service.ts) to initialize consumers. It must implement one of the `IConsumer`, `ISharedConsumer` or `IGenericConsumer` interfaces and should be using the `asConsumer` binding template. If you want the consumers to start at the start of your application, you should pass the `initObservers` config to the Component configuration.
+
+- `IConsumer` - simple consumer for 1 event in a stream
+- `ISharedConsumer` - consumer that consumes data for multiple events in a stream (defined with an array of events)
+- `IGenericConsumer` - consumer that consumes data for all events in a stream/topic (defined without any event). By default it is not triggered for an event if a more specific consumer is bound for that event. This behaviour can be changed using the `alwaysRunGenericConsumer` option in consumer configuration.
+
+You can bind any consumer related configuration using the `KafkaClientBindings.ConsumerConfiguration` key. It accepts all the options of KafkaJS, along with an additional option - `alwaysRunGenericConsumer` - this option runs any generic consumer if available always, even if more specific consumers are bound by the client(only the specific consumer would run if this option is false or not provided).
 
 ##### Example
 
@@ -112,13 +117,14 @@ this.configure(KafkaConnectorComponentBindings.COMPONENT).to({
 
 ```ts
 // start.consumer.ts
+// use @genericConsumer for a generic consumer
 @consumer<TestStream, Events.start>()
 export class StartConsumer implements IConsumer<TestStream, Events.start> {
   constructor(
     @inject('test.handler.start')
     public handler: StreamHandler<TestStream, Events.start>,
   ) {}
-  topic: Topics.First = Topics.First;
+  topic = Topics.First;
   event: Events.start = Events.start;
   // you can write the handler as a method
   handler(payload: StartEvent) {
@@ -149,7 +155,7 @@ export class StartConsumer implements IConsumer<TestStream, Events.start> {
     @eventHandler<TestStream>(Events.Start)
     public handler: StreamHandler<TestStream, Events.start>,
   ) {}
-  topic: Topics.First = Topics.First;
+  topic = Topics.First;
   event: Events.start = Events.start;
 }
 ```

package-lock.json

@@ -95,16 +95,17 @@ describe('end-to-end', () => {
 
     it('should consume from a generic consumer without events for a single topic', async () => {
       const producerInstance = producerApp.getSync<Producer<TestStream>>(
-        producerKey(Topics.First),
+        producerKey(Topics.Generic),
       );
       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)),
-      );
+      expect(genericHandler.getCalls()[0].args[0]).to.be.deepEqual({
+        data: JSON.parse(JSON.stringify(close)),
+        event: Events.close,
+      });
     });
 
     it('should not handle an unspecified events', async () => {
@@ -118,7 +119,7 @@ describe('end-to-end', () => {
       };
       await producerInstance.connect();
       producerInstance.send({
-        topic: Topics.First,
+        topic: Topics.Generic,
         messages: [payload],
       });
 

src/__tests__/acceptance/application.test.ts

@@ -1,15 +1,14 @@
 import {inject, injectable} from '@loopback/core';
 import {asConsumer} from '../../../../keys';
-import {TestStream} from '../stream';
-import {IGenericConsumer, StreamHandler} from '../../../../types';
+import {GenericStream} from '../stream';
+import {GenericStreamHandler, IGenericConsumer} from '../../../../types';
 import {Topics} from '../topics.enum';
-import {Events} from '../events.enum';
 
 @injectable(asConsumer)
-export class GenericConsumer implements IGenericConsumer<TestStream> {
+export class GenericConsumer implements IGenericConsumer<GenericStream> {
   constructor(
     @inject('eventHandler.generic')
-    public handler: StreamHandler<TestStream, Events>,
+    public handler: GenericStreamHandler<GenericStream>,
   ) {}
-  topic: Topics.First = Topics.First;
+  topic: Topics.Generic = Topics.Generic;
 }

src/__tests__/acceptance/fixtures/consumer/generic-consumer.extension.ts

@@ -13,3 +13,10 @@ export interface TestStream extends IStreamDefinition {
     [Events.close]: {};
   };
 }
+
+export interface GenericStream extends IStreamDefinition {
+  topic: Topics.Generic;
+  messages: {
+    [Events.close]: {};
+  };
+}

src/__tests__/acceptance/fixtures/stream.ts

@@ -1,4 +1,5 @@
 export enum Topics {
   First = 'first',
   Second = 'second',
+  Generic = 'generic',
 }

src/__tests__/acceptance/fixtures/topics.enum.ts

@@ -0,0 +1,9 @@
+import {Constructor, injectable} from '@loopback/core';
+import {asConsumer} from '../keys';
+import {IGenericConsumer, IStreamDefinition} from '../types';
+
+export function genericConsumer<T extends IStreamDefinition>() {
+  return injectable(asConsumer) as (
+    target: Constructor<IGenericConsumer<T>>,
+  ) => void;
+}

src/decorators/generic-consumer.decorator.ts

@@ -1,3 +1,4 @@
 export * from './producer.decorator';
 export * from './handler.decorator';
 export * from './consumer.decorator';
+export * from './generic-consumer.decorator';

src/decorators/index.ts

@@ -41,7 +41,7 @@ export class KafkaConsumerService<T extends IStreamDefinition> {
 
     const {consumerMap, genericConsumerMap} = await this.buildConsumerMaps();
     const topics: string[] = Array.from(consumerMap.keys());
-
+    topics.push(...Array.from(genericConsumerMap.keys()));
     await kafkaConsumerClient.subscribe({
       topics,
     });
@@ -70,7 +70,7 @@ export class KafkaConsumerService<T extends IStreamDefinition> {
             (!consumer || this.configuration.alwaysRunGenericConsumer) &&
             genericConsumer
           ) {
-            await genericConsumer.handler(message.data);
+            await genericConsumer.handler(message);
           }
         } else {
           this.logger.warn(

src/services/kafka-consumer.service.ts

@@ -46,7 +46,7 @@ export interface ISharedConsumer<Stream extends IStreamDefinition> {
 
 export interface IGenericConsumer<Stream extends IStreamDefinition> {
   topic: TopicForStream<Stream>;
-  handler: StreamHandler<Stream, EventsInStream<Stream>>;
+  handler: GenericStreamHandler<Stream>;
 }
 
 export function isGenericConsumer<
@@ -89,3 +89,7 @@ export type StreamHandler<
   Stream extends IStreamDefinition,
   K extends EventsInStream<Stream>,
 > = (payload: Stream['messages'][K]) => Promise<void>;
+
+export type GenericStreamHandler<Stream extends IStreamDefinition> = (
+  payload: Stream['messages'],
+) => Promise<void>;