Adds MessageRequest decorator to define topic for client consumer subscriptions.

Author: mkaufmaner

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

@@ -5,12 +5,14 @@ import {
   EventPattern,
   MessagePattern,
   Transport,
+  MessageRequest,
 } from '@nestjs/microservices';
 import { Logger } from '@nestjs/common/services/logger.service';
 import * as util from 'util';
 
 import { Observable } from 'rxjs';
 import * as uuid from 'uuid';
+import * as Bluebird from 'bluebird';
 
 @Controller()
 export class KafkaController implements OnModuleInit {
@@ -69,6 +71,7 @@ export class KafkaController implements OnModuleInit {
 
   @Post()
   @HttpCode(200)
+  @MessageRequest('math.sum', 'math.sum.reply')
   async call(
     @Query('command') cmd,
     @Body() data: number[],
@@ -82,7 +85,9 @@ export class KafkaController implements OnModuleInit {
       }),
     }).toPromise();
 
-    this.logger.error(util.format('@Query math.sum result %o', result));
+    await Bluebird.delay(30000);
+
+    // this.logger.error(util.format('@Query math.sum result %o', result));
 
     const sum = JSON.parse(result.value.toString());
 
@@ -91,7 +96,7 @@ export class KafkaController implements OnModuleInit {
 
   @MessagePattern('math.sum')
   mathSum(data: any){
-    this.logger.error(util.format('@MessagePattern math.sum data %o', data));
+    // this.logger.error(util.format('@MessagePattern math.sum data %o', data));
 
     const value = JSON.parse(data.value);
 
@@ -103,6 +108,7 @@ export class KafkaController implements OnModuleInit {
   }
 
   @Post('notify')
+  @MessageRequest('test.one', 'test.one.reply')
   async sendNotification(): Promise<any> {
     return this.client.emit('notification', [{
       key: 'notify',

packages/microservices/client/client-kafka.ts

@@ -1,5 +1,5 @@
 import * as util from 'util';
-import { isUndefined } from '@nestjs/common/utils/shared.utils';
+import { isUndefined, isNil } from '@nestjs/common/utils/shared.utils';
 import { Logger } from '@nestjs/common/services/logger.service';
 import { loadPackage } from '@nestjs/common/utils/load-package.util';
 import { Observable } from 'rxjs';
@@ -20,7 +20,8 @@ import {
   EachMessagePayload,
   Message,
   KafkaMessage,
-  logLevel
+  logLevel,
+  ConsumerGroupJoinEvent
 } from '../external/kafka.interface';
 import { KafkaHeaders } from '../enums';
 
@@ -40,6 +41,10 @@ export class ClientKafka extends ClientProxy {
   private readonly clientId: string;
   private readonly groupId: string;
 
+  private consumerAssignments: {[key: string]: number[]} = {};
+
+  private static REPLY_PATTERN_AFFIX: string = '.reply';
+
   constructor(protected readonly options: KafkaOptions['options']) {
     super();
     this.brokers = this.getOptionsProp(this.options.client, 'brokers') || [KAFKA_DEFAULT_BROKER];
@@ -66,22 +71,44 @@ export class ClientKafka extends ClientProxy {
       return this.producer;
     }
     this.client = this.createClient();
+
     this.producer = this.client.producer(this.options.producer || {});
     this.consumer = this.client.consumer(Object.assign(this.options.consumer || {}, {
       groupId: this.groupId
     }));
 
+    // set member assignments on join and rebalance
+    this.consumer.on(this.consumer.events.GROUP_JOIN, (data: ConsumerGroupJoinEvent) => {
+      this.consumerAssignments = data.payload.memberAssignment;
+    });
+
+    // connect the producer and consumer
     await this.producer.connect();
     await this.consumer.connect();
 
-    // @TODO: Use descriptors to define the reply topics
-    // Run the consumer
-    await this.consumer.subscribe({topic: 'math.sum.reply'});
+    // bind the topics
+    await this.bindTopics();
+
+    return this.producer;
+  }
+
+  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);
+
+      // subscribe to the pattern of the topic
+      await this.consumer.subscribe({
+        topic: replyPattern
+      });
+    }));
+
+    // run the consumer to start listening on the reply topics
     await this.consumer.run(Object.assign(this.options.run || {}, {
       eachMessage: this.createResponseCallback()
     }));
-
-    return this.producer;
   }
 
   public createClient<T = any>(): T {
@@ -118,19 +145,10 @@ export class ClientKafka extends ClientProxy {
 
   public createResponseCallback(): (payload: EachMessagePayload) => any {
     return (payload: EachMessagePayload) => {
-      // const { err, response, isDisposed, id } = JSON.parse(
-      //   buffer.toString(),
-      // ) as WritePacket & PacketId;
-
       const packet = this.deserialize(payload);
 
-      this.logger.error(util.format('createResponseCallback() fn() packet: %o', packet));
-
       const callback = this.routingMap.get(packet.id);
 
-      this.logger.error(util.format('createResponseCallback() fn() this.routingMap: %o', this.routingMap));
-      this.logger.error(util.format('createResponseCallback() fn() callback: %o', callback));
-
       if (!callback) {
         return undefined;
       }
@@ -188,19 +206,37 @@ export class ClientKafka extends ClientProxy {
     }, this.options.send || {}));
   }
 
-  private getReplyTopic(pattern: string): string {
-    return `${pattern}.reply`;
-  }
-
-  private getReplyPartition(topic: string): string {
+  private getReplyPartition(topic: string): number {
     // this.consumer.describeGroup().then((description) => {
-    //   this.logger.error(util.format('getReplyTopicPartition(): topic: %s groupDescription: %o', topic, description));
+    //   // this.logger.error(util.format('getReplyTopicPartition(): groupDescription: %o', description));
+
+    //   description.members.forEach((member) => {
+    //     // const memberMetadata =  kafkaPackage.AssignerProtocol.MemberMetadata.decode(member.memberMetadata);
+    //     const memberAssignment = kafkaPackage.AssignerProtocol.MemberAssignment.decode(member.memberAssignment);
+
+    //     this.logger.error(util.format('getReplyTopicPartition(): groupDescription.member[i]: %o', member));
+    //     // this.logger.error(util.format('getReplyTopicPartition(): groupDescription.member[i] metadata: %o', memberMetadata));
+    //     this.logger.error(util.format('getReplyTopicPartition(): groupDescription.member[i] assignment: %o', memberAssignment));
+    //   });
     // });
 
-    return '0';
-  }
+    // return 0;
 
-  // private getReplyTopic
+    // get topic assignment
+    const topicAssignments = this.consumerAssignments[topic];
+
+    // throw error
+    if (isUndefined(topicAssignments)) {
+      throw new Error(`Unable to send the message request because the client consumer is not subscribed to the topic (${topic}).`);
+    }
+
+    // if the current member isn't listening to any partitions on the topic then throw an error.
+    if (isUndefined(topicAssignments[0])) {
+      throw new Error(`Unable to send the message request because the client consumer subscribed to the topic (${topic}) is not assigned any partitions.`);
+    }
+
+    return topicAssignments[0];
+  }
 
   protected publish(
     partialPacket: ReadPacket,
@@ -209,18 +245,9 @@ export class ClientKafka extends ClientProxy {
     try {
       const packet = this.assignPacketId(partialPacket);
       const pattern = this.normalizePattern(partialPacket.pattern);
-      const replyTopic = this.getReplyTopic(pattern);
+      const replyTopic = this.getReplyPattern(pattern, ClientKafka.REPLY_PATTERN_AFFIX);
       const replyPartition = this.getReplyPartition(replyTopic);
 
-      // subscribe
-      // this.consumer.subscribe({
-      //   topic: replyTopic
-      // }).then(() => {
-      //   return this.consumer.run(Object.assign(this.options.run || {}, {
-      //     eachMessage: this.createResponseCallback()
-      //   }));
-      // });
-
       // set the route
       this.routingMap.set(packet.id, callback);
 
@@ -234,8 +261,6 @@ export class ClientKafka extends ClientProxy {
       packet.data.headers[KafkaHeaders.REPLY_TOPIC] = replyTopic;
       packet.data.headers[KafkaHeaders.REPLY_PARTITION] = replyPartition;
 
-      this.logger.error(util.format('publish() packet %o', packet));
-
       // send through unhandled promise
       this.producer.send(Object.assign({
         topic: pattern,

packages/microservices/client/client-proxy.ts

@@ -25,6 +25,27 @@ 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);
+    const 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/constants.ts

@@ -21,6 +21,9 @@ export const CLIENT_CONFIGURATION_METADATA = 'microservices:client';
 export const PATTERN_HANDLER_METADATA = 'microservices:handler_type';
 export const CLIENT_METADATA = 'microservices:is_client_instance';
 
+export const REQUEST_PATTERN_METADATA = 'microservices:request_pattern';
+export const REPLY_PATTERN_METADATA = 'microservices:reply_pattern';
+
 export const RQM_DEFAULT_QUEUE = 'default';
 export const RQM_DEFAULT_PREFETCH_COUNT = 0;
 export const RQM_DEFAULT_IS_GLOBAL_PREFETCH_COUNT = false;

packages/microservices/decorators/index.ts

@@ -2,3 +2,4 @@ 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

@@ -0,0 +1,20 @@
+import { REQUEST_PATTERN_METADATA, REPLY_PATTERN_METADATA } from '../constants';
+import { PatternMetadata } from '../interfaces/pattern-metadata.interface';
+
+/**
+ * Makes outgoing to incoming messages that fulfils the chosen patterns.
+ */
+export const MessageRequest = <T = PatternMetadata | string>(
+  requestMetadata?: T,
+  replyMetadata?: T
+): MethodDecorator => {
+  return (
+    target: any,
+    key: string | symbol,
+    descriptor: PropertyDescriptor,
+  ) => {
+    Reflect.defineMetadata(REQUEST_PATTERN_METADATA, requestMetadata, descriptor.value);
+    Reflect.defineMetadata(REPLY_PATTERN_METADATA, replyMetadata, descriptor.value);
+    return descriptor;
+  };
+};

packages/microservices/enums/pattern-handler.enum.ts

@@ -1,4 +1,4 @@
 export enum PatternHandler {
   MESSAGE = 1,
-  EVENT = 2,
+  EVENT = 2
 }

packages/microservices/listener-metadata-explorer.ts

@@ -6,6 +6,8 @@ import {
   CLIENT_METADATA,
   PATTERN_HANDLER_METADATA,
   PATTERN_METADATA,
+  REQUEST_PATTERN_METADATA,
+  REPLY_PATTERN_METADATA,
 } from './constants';
 import { PatternHandler } from './enums/pattern-handler.enum';
 import { ClientOptions } from './interfaces/client-metadata.interface';
@@ -23,6 +25,11 @@ export interface PatternProperties {
   targetCallback: (...args: any[]) => any;
 }
 
+export interface MessageRequestProperties {
+  requestPattern: PatternMetadata;
+  replyPattern: PatternMetadata;
+}
+
 export class ListenerMetadataExplorer {
   constructor(private readonly metadataScanner: MetadataScanner) {}
 
@@ -32,12 +39,11 @@ export class ListenerMetadataExplorer {
       Controller,
       PatternProperties
     >(instance, instancePrototype, method =>
-      this.exploreMethodMetadata(instance, instancePrototype, method),
+      this.exploreMethodMetadata(instancePrototype, method),
     );
   }
 
   public exploreMethodMetadata(
-    instance: object,
     instancePrototype: any,
     methodKey: string,
   ): PatternProperties {
@@ -78,4 +84,41 @@ 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/listeners-controller.ts

@@ -77,7 +77,19 @@ export class ListenersController {
     } 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);
     }
   }