Merge branch 'master' into Upperfoot-feature/socket-options

Author: kamilmysliwiec

packages/microservices/client/index.ts

@@ -4,4 +4,5 @@ export * from './client-nats';
 export * from './client-proxy';
 export { ClientProxyFactory } from './client-proxy-factory';
 export * from './client-redis';
+export * from './client-rmq';
 export * from './client-tcp';

packages/microservices/server/index.ts

@@ -3,4 +3,5 @@ export * from './server-grpc';
 export * from './server-mqtt';
 export * from './server-nats';
 export * from './server-redis';
+export * from './server-rmq';
 export * from './server-tcp';

packages/microservices/server/server-grpc.ts

@@ -30,14 +30,12 @@ export class ServerGrpc extends Server implements CustomTransportStrategy {
   private readonly url: string;
   private grpcClient: any;
 
-  constructor(private readonly options: MicroserviceOptions['options']) {
+  constructor(private readonly options: GrpcOptions['options']) {
     super();
-    this.url =
-      this.getOptionsProp<GrpcOptions>(options, 'url') || GRPC_DEFAULT_URL;
+    this.url = this.getOptionsProp(options, 'url') || GRPC_DEFAULT_URL;
 
     const protoLoader =
-      this.getOptionsProp<GrpcOptions>(options, 'protoLoader') ||
-      GRPC_DEFAULT_PROTO_LOADER;
+      this.getOptionsProp(options, 'protoLoader') || GRPC_DEFAULT_PROTO_LOADER;
 
     grpcPackage = this.loadPackage('grpc', ServerGrpc.name, () =>
       require('grpc'),
@@ -58,10 +56,7 @@ export class ServerGrpc extends Server implements CustomTransportStrategy {
 
   public async bindEvents() {
     const grpcContext = this.loadProto();
-    const packageName = this.getOptionsProp<GrpcOptions>(
-      this.options,
-      'package',
-    );
+    const packageName = this.getOptionsProp(this.options, 'package');
     const grpcPkg = this.lookupPackage(grpcContext, packageName);
     if (!grpcPkg) {
       const invalidPackageError = new InvalidGrpcPackageException();
@@ -165,21 +160,18 @@ export class ServerGrpc extends Server implements CustomTransportStrategy {
 
   public createClient(): any {
     const server = new grpcPackage.Server({
-      'grpc.max_send_message_length': this.getOptionsProp<GrpcOptions>(
+      'grpc.max_send_message_length': this.getOptionsProp(
         this.options,
         'maxSendMessageLength',
         GRPC_DEFAULT_MAX_SEND_MESSAGE_LENGTH,
       ),
-      'grpc.max_receive_message_length': this.getOptionsProp<GrpcOptions>(
+      'grpc.max_receive_message_length': this.getOptionsProp(
         this.options,
         'maxReceiveMessageLength',
         GRPC_DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH,
       ),
     });
-    const credentials = this.getOptionsProp<GrpcOptions>(
-      this.options,
-      'credentials',
-    );
+    const credentials = this.getOptionsProp(this.options, 'credentials');
     server.bind(
       this.url,
       credentials || grpcPackage.ServerCredentials.createInsecure(),
@@ -198,8 +190,8 @@ export class ServerGrpc extends Server implements CustomTransportStrategy {
 
   public loadProto(): any {
     try {
-      const file = this.getOptionsProp<GrpcOptions>(this.options, 'protoPath');
-      const loader = this.getOptionsProp<GrpcOptions>(this.options, 'loader');
+      const file = this.getOptionsProp(this.options, 'protoPath');
+      const loader = this.getOptionsProp(this.options, 'loader');
 
       const packageDefinition = grpcProtoLoaderPackage.loadSync(file, loader);
       const packageObject = grpcPackage.loadPackageDefinition(

packages/microservices/server/server-mqtt.ts

@@ -21,10 +21,9 @@ export class ServerMqtt extends Server implements CustomTransportStrategy {
   private readonly url: string;
   private mqttClient: MqttClient;
 
-  constructor(private readonly options: MicroserviceOptions['options']) {
+  constructor(private readonly options: MqttOptions['options']) {
     super();
-    this.url =
-      this.getOptionsProp<MqttOptions>(options, 'url') || MQTT_DEFAULT_URL;
+    this.url = this.getOptionsProp(options, 'url') || MQTT_DEFAULT_URL;
 
     mqttPackage = this.loadPackage('mqtt', ServerMqtt.name, () =>
       require('mqtt'),
@@ -59,10 +58,7 @@ export class ServerMqtt extends Server implements CustomTransportStrategy {
   }
 
   public createMqttClient(): MqttClient {
-    return mqttPackage.connect(
-      this.url,
-      this.options as MqttOptions,
-    );
+    return mqttPackage.connect(this.url, this.options as MqttOptions);
   }
 
   public getMessageHandler(pub: MqttClient): Function {

packages/microservices/server/server-nats.ts

@@ -21,10 +21,9 @@ export class ServerNats extends Server implements CustomTransportStrategy {
   private readonly url: string;
   private natsClient: Client;
 
-  constructor(private readonly options: MicroserviceOptions['options']) {
+  constructor(private readonly options: NatsOptions['options']) {
     super();
-    this.url =
-      this.getOptionsProp<NatsOptions>(this.options, 'url') || NATS_DEFAULT_URL;
+    this.url = this.getOptionsProp(this.options, 'url') || NATS_DEFAULT_URL;
 
     natsPackage = this.loadPackage('nats', ServerNats.name, () =>
       require('nats'),
@@ -43,7 +42,7 @@ export class ServerNats extends Server implements CustomTransportStrategy {
   }
 
   public bindEvents(client: Client) {
-    const queue = this.getOptionsProp<NatsOptions>(this.options, 'queue');
+    const queue = this.getOptionsProp(this.options, 'queue');
     const subscribe = queue
       ? (channel: string) =>
           client.subscribe(

packages/microservices/server/server-redis.ts

@@ -27,11 +27,9 @@ export class ServerRedis extends Server implements CustomTransportStrategy {
   private pubClient: RedisClient;
   private isExplicitlyTerminated = false;
 
-  constructor(private readonly options: MicroserviceOptions['options']) {
+  constructor(private readonly options: RedisOptions['options']) {
     super();
-    this.url =
-      this.getOptionsProp<RedisOptions>(this.options, 'url') ||
-      REDIS_DEFAULT_URL;
+    this.url = this.getOptionsProp(this.options, 'url') || REDIS_DEFAULT_URL;
 
     redisPackage = this.loadPackage('redis', ServerRedis.name, () =>
       require('redis'),
@@ -148,12 +146,11 @@ export class ServerRedis extends Server implements CustomTransportStrategy {
     }
     if (
       this.isExplicitlyTerminated ||
-      !this.getOptionsProp<RedisOptions>(this.options, 'retryAttempts') ||
-      options.attempt >
-        this.getOptionsProp<RedisOptions>(this.options, 'retryAttempts')
+      !this.getOptionsProp(this.options, 'retryAttempts') ||
+      options.attempt > this.getOptionsProp(this.options, 'retryAttempts')
     ) {
       return undefined;
     }
-    return this.getOptionsProp<RedisOptions>(this.options, 'retryDelay') || 0;
+    return this.getOptionsProp(this.options, 'retryDelay') || 0;
   }
 }

packages/microservices/server/server-rmq.ts

@@ -26,22 +26,19 @@ export class ServerRMQ extends Server implements CustomTransportStrategy {
   private readonly queueOptions: any;
   private readonly isGlobalPrefetchCount: boolean;
 
-  constructor(private readonly options: MicroserviceOptions) {
+  constructor(private readonly options: RmqOptions['options']) {
     super();
-    this.urls = this.getOptionsProp<RmqOptions>(this.options, 'urls') || [
-      RQM_DEFAULT_URL,
-    ];
+    this.urls = this.getOptionsProp(this.options, 'urls') || [RQM_DEFAULT_URL];
     this.queue =
-      this.getOptionsProp<RmqOptions>(this.options, 'queue') ||
-      RQM_DEFAULT_QUEUE;
+      this.getOptionsProp(this.options, 'queue') || RQM_DEFAULT_QUEUE;
     this.prefetchCount =
-      this.getOptionsProp<RmqOptions>(this.options, 'prefetchCount') ||
+      this.getOptionsProp(this.options, 'prefetchCount') ||
       RQM_DEFAULT_PREFETCH_COUNT;
     this.isGlobalPrefetchCount =
-      this.getOptionsProp<RmqOptions>(this.options, 'isGlobalPrefetchCount') ||
+      this.getOptionsProp(this.options, 'isGlobalPrefetchCount') ||
       RQM_DEFAULT_IS_GLOBAL_PREFETCH_COUNT;
     this.queueOptions =
-      this.getOptionsProp<RmqOptions>(this.options, 'queueOptions') ||
+      this.getOptionsProp(this.options, 'queueOptions') ||
       RQM_DEFAULT_QUEUE_OPTIONS;
 
     this.loadPackage('amqplib', ServerRMQ.name, () => require('amqplib'));

packages/microservices/server/server-tcp.ts

@@ -26,12 +26,11 @@ export class ServerTCP extends Server implements CustomTransportStrategy {
   private isExplicitlyTerminated = false;
   private retryAttemptsCount = 0;
 
-  constructor(private readonly options: MicroserviceOptions['options']) {
+  constructor(private readonly options: TcpOptions['options']) {
     super();
-    this.port =
-      this.getOptionsProp<TcpOptions>(options, 'port') || TCP_DEFAULT_PORT;
+    this.port = this.getOptionsProp(options, 'port') || TCP_DEFAULT_PORT;
     this.host =
-      this.getOptionsProp<TcpOptions>(options, 'host') || TCP_DEFAULT_HOST;
+      this.getOptionsProp(options, 'host') || TCP_DEFAULT_HOST;
 
     this.init();
   }
@@ -86,16 +85,16 @@ export class ServerTCP extends Server implements CustomTransportStrategy {
   public handleClose(): undefined | number | NodeJS.Timer {
     if (
       this.isExplicitlyTerminated ||
-      !this.getOptionsProp<TcpOptions>(this.options, 'retryAttempts') ||
+      !this.getOptionsProp(this.options, 'retryAttempts') ||
       this.retryAttemptsCount >=
-        this.getOptionsProp<TcpOptions>(this.options, 'retryAttempts')
+        this.getOptionsProp(this.options, 'retryAttempts')
     ) {
       return undefined;
     }
     ++this.retryAttemptsCount;
     return setTimeout(
       () => this.server.listen(this.port),
-      this.getOptionsProp<TcpOptions>(this.options, 'retryDelay') || 0,
+      this.getOptionsProp(this.options, 'retryDelay') || 0,
     );
   }
 

packages/microservices/server/server.ts

@@ -73,12 +73,11 @@ export abstract class Server {
     return resultOrDeffered;
   }
 
-  public getOptionsProp<T extends { options?: any }>(
-    obj: MicroserviceOptions['options'],
-    prop: keyof T['options'],
-    defaultValue: any = undefined,
-  ) {
-    return (obj && obj[prop as string]) || defaultValue;
+  public getOptionsProp<
+    T extends MicroserviceOptions['options'],
+    K extends keyof T
+  >(obj: T, prop: K, defaultValue: T[K] = undefined) {
+    return (obj && obj[prop]) || defaultValue;
   }
 
   protected handleError(error: string) {