Working on Kafka Events.

Author: mkaufmaner

integration/docker-compose.yml

@@ -43,3 +43,11 @@ services:
       - "15672:15672"
       - "5672:5672"
     tty: true
+  kafka:
+    container_name: test-kafka
+    hostname: kafka
+    image: wurstmeister/kafka
+    ports:
+      - "9092:9092"
+
+

package.json

@@ -83,7 +83,6 @@
     "@types/cors": "2.8.5",
     "@types/express": "4.17.0",
     "@types/fastify-cors": "2.1.0",
-    "@types/kafka-node": "2.0.8",
     "@types/mocha": "5.2.7",
     "@types/node": "10.14.8",
     "@types/redis": "2.8.13",
@@ -113,6 +112,7 @@
     "husky": "1.3.1",
     "imports-loader": "0.8.0",
     "json-loader": "0.5.7",
+    "kafkajs": "^1.8.0",
     "lerna": "3.14.1",
     "lint-staged": "8.2.0",
     "memory-usage": "1.2.1",

packages/microservices/constants.ts

@@ -36,5 +36,5 @@ export const DISCONNECTED_RMQ_MESSAGE = `Disconnected from RMQ. Trying to reconn
 export const GRPC_DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH = 4 * 1024 * 1024;
 export const GRPC_DEFAULT_MAX_SEND_MESSAGE_LENGTH = 4 * 1024 * 1024;
 
-export const KAFKA_DEFAULT_CLIENT = 'nestjs';
-export const KAFKA_DEFAULT_GROUP = 'nestjs';
+export const KAFKA_DEFAULT_CLIENT = 'nestjs-consumer';
+export const KAFKA_DEFAULT_GROUP = 'nestjs-group';

packages/microservices/decorators/message-pattern.decorator.ts

@@ -1,6 +1,7 @@
 import { PATTERN_HANDLER_METADATA, PATTERN_METADATA } from '../constants';
 import { PatternHandler } from '../enums/pattern-handler.enum';
 import { PatternMetadata } from '../interfaces/pattern-metadata.interface';
+import { ConsumerConfig } from '../external/kafka.interface';
 
 export enum GrpcMethodStreamingType {
   NO_STREAMING = 'no_stream',
@@ -40,7 +41,7 @@ export function GrpcMethod(service: string, method?: string): MethodDecorator {
     key: string | symbol,
     descriptor: PropertyDescriptor,
   ) => {
-    const metadata = createMethodMetadata(target, key, service, method);
+    const metadata = createGrpcMethodMetadata(target, key, service, method);
     return MessagePattern(metadata)(target, key, descriptor);
   };
 }
@@ -62,7 +63,7 @@ export function GrpcStreamMethod(service: string, method?: string) {
     key: string | symbol,
     descriptor: PropertyDescriptor,
   ) => {
-    const metadata = createMethodMetadata(
+    const metadata = createGrpcMethodMetadata(
       target,
       key,
       service,
@@ -90,7 +91,7 @@ export function GrpcStreamCall(service: string, method?: string) {
     key: string | symbol,
     descriptor: PropertyDescriptor,
   ) => {
-    const metadata = createMethodMetadata(
+    const metadata = createGrpcMethodMetadata(
       target,
       key,
       service,
@@ -101,7 +102,7 @@ export function GrpcStreamCall(service: string, method?: string) {
   };
 }
 
-export function createMethodMetadata(
+export function createGrpcMethodMetadata(
   target: any,
   key: string | symbol,
   service: string | undefined,
@@ -124,3 +125,73 @@ export function createMethodMetadata(
   }
   return { service, rpc: method, streaming };
 }
+
+export enum KafkaConsumerHandlerType {
+  EACH_MESSAGE = 'eachMessage',
+  EACH_BATCH = 'eachBatch',
+}
+
+export interface KafkaConsumerSubscriptionOptions {
+  topic: string | RegExp;
+  fromBeginning?: boolean;
+}
+
+export interface KafkaConsumerRunOptions {
+  autoCommit?: boolean;
+  autoCommitInterval?: number | null;
+  autoCommitThreshold?: number | null;
+  eachBatchAutoResolve?: boolean;
+  partitionsConsumedConcurrently?: number;
+}
+
+/**
+ * Registers a consumer handler for processing each message of a topic individually.
+ *
+ * @param subscription The options for the subscription to the topic.
+ * @param run The options for the consumption of messages from the topic.
+ */
+export function KafkaConsumer(subscription: KafkaConsumerSubscriptionOptions): MethodDecorator;
+export function KafkaConsumer(subscription: KafkaConsumerSubscriptionOptions, run?: KafkaConsumerRunOptions): MethodDecorator;
+export function KafkaConsumer(subscription: KafkaConsumerSubscriptionOptions, run?: KafkaConsumerRunOptions): MethodDecorator {
+  return (
+    target: any,
+    key: string | symbol,
+    descriptor: PropertyDescriptor,
+  ) => {
+    const metadata = createKafkaMethodMetadata(target, key, subscription, run);
+    return MessagePattern(metadata)(target, key, descriptor);
+  };
+}
+
+/**
+ * Registers a consumer handler for processing each message of a topic in bitch.
+ *
+ * @param subscription The options for the subscription to the topic.
+ * @param run The options for the consumption of messages from the topic.
+ */
+export function KafkaBatchConsumer(subscription: KafkaConsumerSubscriptionOptions): MethodDecorator;
+export function KafkaBatchConsumer(subscription: KafkaConsumerSubscriptionOptions, run?: KafkaConsumerRunOptions): MethodDecorator;
+export function KafkaBatchConsumer(subscription: KafkaConsumerSubscriptionOptions, run?: KafkaConsumerRunOptions): MethodDecorator {
+  return (
+    target: any,
+    key: string | symbol,
+    descriptor: PropertyDescriptor,
+  ) => {
+    const metadata = createKafkaMethodMetadata(target, key, subscription, run, KafkaConsumerHandlerType.EACH_BATCH);
+    return MessagePattern(metadata)(target, key, descriptor);
+  };
+}
+
+export function createKafkaMethodMetadata(
+  target: any,
+  key: string | symbol,
+  subscription: KafkaConsumerSubscriptionOptions,
+  run: KafkaConsumerRunOptions | undefined,
+  handler = KafkaConsumerHandlerType.EACH_MESSAGE
+) {
+  return {
+    subscription,
+    run,
+    handler
+  };
+}