Removes @MessageRequest decorator in favor of subscribeToResponseOf() method.

Author: mkaufmaner

integration/microservices/src/kafka/kafka.controller.ts

@@ -1,10 +1,8 @@
-import * as util from 'util';
-import { Body, Controller, HttpCode, Post, Query } from '@nestjs/common';
+import { Body, Controller, HttpCode, Post, OnModuleInit } from '@nestjs/common';
 import {
   Client,
-  ClientProxy,
   Transport,
-  MessageRequest,
+  ClientKafka,
 } from '@nestjs/microservices';
 import { Logger } from '@nestjs/common/services/logger.service';
 
@@ -13,7 +11,7 @@ import { UserDto } from './dtos/user.dto';
 import { BusinessDto } from './dtos/business.dto';
 
 @Controller()
-export class KafkaController {
+export class KafkaController implements OnModuleInit {
   protected readonly logger = new Logger(KafkaController.name);
   static IS_NOTIFIED = false;
   static MATH_SUM = 0;
@@ -26,12 +24,28 @@ export class KafkaController {
       },
     },
   })
-  private readonly client: ClientProxy;
+  private readonly client: ClientKafka;
+
+  onModuleInit(){
+    const requestPatterns = [
+      'math.sum.sync.kafka.message',
+      'math.sum.sync.without.key',
+      'math.sum.sync.plain.object',
+      'math.sum.sync.array',
+      'math.sum.sync.string',
+      'math.sum.sync.number',
+      'user.create',
+      'business.create',
+    ];
+
+    requestPatterns.forEach((pattern) => {
+      this.client.subscribeToResponseOf(pattern);
+    });
+  }
 
   // sync send kafka message
   @Post('mathSumSyncKafkaMessage')
   @HttpCode(200)
-  @MessageRequest('math.sum.sync.kafka.message', 'math.sum.sync.kafka.message.reply')
   async mathSumSyncKafkaMessage(
     @Body() data: number[],
   ): Promise<Observable<any>> {
@@ -47,7 +61,6 @@ export class KafkaController {
   // sync send kafka(ish) message without key and only the value
   @Post('mathSumSyncWithoutKey')
   @HttpCode(200)
-  @MessageRequest('math.sum.sync.without.key', 'math.sum.sync.without.key.reply')
   async mathSumSyncWithoutKey(
     @Body() data: number[],
   ): Promise<Observable<any>> {
@@ -62,7 +75,6 @@ export class KafkaController {
   // sync send message without key or value
   @Post('mathSumSyncPlainObject')
   @HttpCode(200)
-  @MessageRequest('math.sum.sync.plain.object', 'math.sum.sync.plain.object.reply')
   async mathSumSyncPlainObject(
     @Body() data: number[],
   ): Promise<Observable<any>> {
@@ -75,7 +87,6 @@ export class KafkaController {
   // sync send message without key or value
   @Post('mathSumSyncArray')
   @HttpCode(200)
-  @MessageRequest('math.sum.sync.array', 'math.sum.sync.array.reply')
   async mathSumSyncArray(
     @Body() data: number[],
   ): Promise<Observable<any>> {
@@ -85,7 +96,6 @@ export class KafkaController {
 
   @Post('mathSumSyncString')
   @HttpCode(200)
-  @MessageRequest('math.sum.sync.string', 'math.sum.sync.string.reply')
   async mathSumSyncString(
     @Body() data: number[],
   ): Promise<Observable<any>> {
@@ -96,7 +106,6 @@ export class KafkaController {
 
   @Post('mathSumSyncNumber')
   @HttpCode(200)
-  @MessageRequest('math.sum.sync.number', 'math.sum.sync.number.reply')
   async mathSumSyncNumber(
     @Body() data: number[],
   ): Promise<Observable<any>> {
@@ -113,7 +122,6 @@ export class KafkaController {
   // Complex data to send.
   @Post('/user')
   @HttpCode(200)
-  @MessageRequest('user.create', 'user.create.reply')
   async createUser(@Body() user: UserDto): Promise<Observable<any>> {
     const result = await this.client.send('user.create', {
       key: '1',
@@ -127,7 +135,6 @@ export class KafkaController {
   // Complex data to send.
   @Post('/business')
   @HttpCode(200)
-  @MessageRequest('business.create', 'business.create.reply')
   async createBusiness(@Body() business: BusinessDto) {
     const result = await this.client.send('business.create', {
       key: '1',

integration/microservices/src/kafka/kafka.messages.controller.ts

@@ -1,30 +1,16 @@
-import * as util from 'util';
 import { Controller } from '@nestjs/common';
-import { Client, ClientProxy, EventPattern, MessagePattern, MessageRequest, Transport } from '@nestjs/microservices';
+import { EventPattern, MessagePattern } from '@nestjs/microservices';
 import { Logger } from '@nestjs/common/services/logger.service';
 import { KafkaController } from './kafka.controller';
 import { BusinessDto } from './dtos/business.dto';
 import { UserEntity } from './entities/user.entity';
 import { BusinessEntity } from './entities/business.entity';
 import { UserDto } from './dtos/user.dto';
-import { KafkaClient } from 'kafka-node';
-import { map } from 'bluebird';
 
 @Controller()
 export class KafkaMessagesController {
   protected readonly logger = new Logger(KafkaMessagesController.name);
   static IS_NOTIFIED = false;
-  static MATH_SUM = 0;
-
-  @Client({
-    transport: Transport.KAFKA,
-    options: {
-      client: {
-        brokers: ['localhost:9092'],
-      },
-    },
-  })
-  private readonly client: ClientProxy;
 
   @MessagePattern('math.sum.sync.kafka.message')
   mathSumSyncKafkaMessage(data: any){

packages/microservices/client/client-kafka.ts

@@ -39,8 +39,7 @@ export class ClientKafka extends ClientProxy {
   private readonly groupId: string;
 
   protected consumerAssignments: {[key: string]: number[]} = {};
-
-  protected static readonly REPLY_PATTERN_AFFIX: string = '.reply';
+  protected responsePatterns: string[] = [];
 
   constructor(protected readonly options: KafkaOptions['options']) {
     super();
@@ -59,6 +58,15 @@ export class ClientKafka extends ClientProxy {
     kafkaPackage = loadPackage('kafkajs', ClientKafka.name, () => require('kafkajs'));
   }
 
+  public subscribeToResponseOf(pattern: any): void {
+    const request = this.normalizePattern(pattern);
+    this.responsePatterns.push(this.getResponsePatternName(request));
+  }
+
+  protected getResponsePatternName(pattern: string): string {
+    return `${pattern}.reply`;
+  }
+
   public close(): void {
     this.producer && this.producer.disconnect();
     this.consumer && this.consumer.disconnect();
@@ -100,15 +108,10 @@ export class ClientKafka extends ClientProxy {
   }
 
   public async bindTopics(): Promise<void> {
-    const requestPatterns = [...this.requestMap.keys()];
-
-    await Promise.all(requestPatterns.map(async requestPattern => {
-      // get the reply pattern
-      const replyPattern = this.getReplyPattern(requestPattern, ClientKafka.REPLY_PATTERN_AFFIX);
-
+    await Promise.all(this.responsePatterns.map(async responsePattern => {
       // subscribe to the pattern of the topic
       await this.consumer.subscribe({
-        topic: replyPattern
+        topic: responsePattern
       });
     }));
 
@@ -198,7 +201,7 @@ export class ClientKafka extends ClientProxy {
     }, this.options.send || {}));
   }
 
-  protected getReplyPartition(topic: string): string {
+  protected getReplyTopicPartition(topic: string): string {
     // get topic assignment
     const topicAssignments = this.consumerAssignments[topic];
 
@@ -226,8 +229,8 @@ export class ClientKafka extends ClientProxy {
 
       // get the response meta
       const pattern = this.normalizePattern(partialPacket.pattern);
-      const replyTopic = this.getReplyPattern(pattern, ClientKafka.REPLY_PATTERN_AFFIX);
-      const replyPartition = this.getReplyPartition(replyTopic);
+      const replyTopic = this.getResponsePatternName(pattern);
+      const replyPartition = this.getReplyTopicPartition(replyTopic);
 
       // set the route
       this.routingMap.set(packet.id, callback);

packages/microservices/client/client-proxy.ts

@@ -25,31 +25,6 @@ export abstract class ClientProxy {
   public abstract close(): any;
 
   protected routingMap = new Map<string, Function>();
-  protected requestMap = new Map<string, string>();
-
-  public addMessageRequest(
-    requestPattern: any,
-    replyPattern?: any
-  ) {
-    const request = this.normalizePattern(requestPattern);
-    let reply = null;
-
-    if (!isNil(replyPattern)) {
-      reply = this.normalizePattern(replyPattern);
-    }
-
-    this.requestMap.set(request, reply);
-  }
-
-  protected getReplyPattern(requestPattern: string, affix: string): string {
-    const replyPattern = this.requestMap.get(requestPattern);
-
-    if (isNil(replyPattern)) {
-      return `${requestPattern}${affix}`;
-    }
-
-    return replyPattern;
-  }
 
   public send<TResult = any, TInput = any>(
     pattern: any,

packages/microservices/decorators/index.ts

@@ -2,4 +2,3 @@ export * from './client.decorator';
 export * from './event-pattern.decorator';
 export * from './grpc-service.decorator';
 export * from './message-pattern.decorator';
-export * from './message-request.decorator';

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

@@ -84,41 +84,4 @@ export class ListenerMetadataExplorer {
       yield { property, metadata };
     }
   }
-
-  public exploreMessageRequests(
-    instance: Controller,
-  ): MessageRequestProperties[] {
-    const instancePrototype = Object.getPrototypeOf(instance);
-    return this.metadataScanner.scanFromPrototype<
-      Controller,
-      MessageRequestProperties
-    >(instance, instancePrototype, method =>
-      this.exploreMessageRequestMethodMetadata(instancePrototype, method),
-    );
-  }
-
-  public exploreMessageRequestMethodMetadata(
-    instancePrototype: any,
-    methodKey: string,
-  ): MessageRequestProperties {
-    const targetCallback = instancePrototype[methodKey];
-
-    const requestPattern = Reflect.getMetadata(
-      REQUEST_PATTERN_METADATA,
-      targetCallback,
-    );
-    const replyPattern = Reflect.getMetadata(
-      REPLY_PATTERN_METADATA,
-      targetCallback,
-    );
-
-    if (isUndefined(requestPattern) || isUndefined(replyPattern)) {
-      return;
-    }
-
-    return {
-      requestPattern,
-      replyPattern,
-    };
-  }
 }

packages/microservices/listener-metadata-explorer.ts

@@ -76,18 +76,6 @@ export class ListenersController {
       metadata,
     } of this.metadataExplorer.scanForClientHooks(instance)) {
       const client = this.clientFactory.create(metadata);
-
-      const messageRequests = this.metadataExplorer.exploreMessageRequests(
-        instance,
-      );
-
-      messageRequests.forEach(messageRequest => {
-        client.addMessageRequest(
-          messageRequest.requestPattern,
-          messageRequest.replyPattern,
-        );
-      });
-
       this.clientsContainer.addClient(client);
 
       this.assignClientToInstance(instance, property, client);