@nestjs/microservices

@nestjs/microservices

Version 11.1.26
Published 4 days ago
477 kB
2 dependencies
MIT license

Install

npm i @nestjs/microservices

yarn add @nestjs/microservices

pnpm add @nestjs/microservices

Overview

Nest - modern, fast, powerful node.js web framework (@microservices)

Index

Variables

variable CONTEXT

const CONTEXT: string;

variable EventPattern

const EventPattern: {
    <T = string>(metadata?: T): MethodDecorator;
    <T = string>(metadata?: T, transport?: Transport | symbol): MethodDecorator;
    <T = string>(metadata?: T, extras?: Record<string, any>): MethodDecorator;
    <T = string>(
        metadata?: T,
        transport?: Transport | symbol,
        extras?: Record<string, any>
    ): MethodDecorator;
};

Subscribes to incoming events which fulfils chosen pattern.

variable GrpcService

const GrpcService: any;

Defines the GrpcService. The service can inject dependencies through constructor. Those dependencies have to belong to the same module.

variable MessagePattern

const MessagePattern: {
    <T = PatternMetadata>(metadata?: T): MethodDecorator;
    <T = PatternMetadata>(
        metadata?: T,
        transport?: Transport | symbol
    ): MethodDecorator;
    <T = PatternMetadata>(
        metadata?: T,
        extras?: Record<string, any>
    ): MethodDecorator;
    <T = PatternMetadata>(
        metadata?: T,
        transport?: Transport | symbol,
        extras?: Record<string, any>
    ): MethodDecorator;
};

Subscribes to incoming messages which fulfils chosen pattern.

Functions

function Client

Client: (metadata?: ClientOptions) => PropertyDecorator;

Attaches the ClientProxy instance to the given property
Parameter metadata
optional client metadata

function createGrpcMethodMetadata

createGrpcMethodMetadata: (
    target: object,
    key: string | symbol,
    service: string | undefined,
    method: string | undefined,
    streaming?: GrpcMethodStreamingType
) => {
    service: string;
    rpc: string | undefined;
    streaming: GrpcMethodStreamingType;
};

function Ctx

Ctx: () => ParameterDecorator;

function getGrpcPackageDefinition

getGrpcPackageDefinition: (
    options: GrpcOptions['options'],
    grpcProtoLoaderPackage: any
) => any;

function GrpcMethod

GrpcMethod: {
    (service?: string): MethodDecorator;
    (service: string, method?: string): MethodDecorator;
};

Registers gRPC method handler for specified service.

function GrpcStreamCall

GrpcStreamCall: {
    (service?: string): MethodDecorator;
    (service: string, method?: string): MethodDecorator;
};

Registers gRPC call pass through handler for service and method
Parameter service
String parameter reflecting the name of service definition from proto file
Parameter service
String parameter reflecting the name of service definition from proto file
Parameter method
Optional string parameter reflecting the name of method inside of a service definition coming after rpc keyword

function GrpcStreamMethod

GrpcStreamMethod: {
    (service?: string): MethodDecorator;
    (service: string, method?: string): MethodDecorator;
};

Registers gRPC call through RX handler for service and method
Parameter service
String parameter reflecting the name of service definition from proto file
Parameter service
String parameter reflecting the name of service definition from proto file
Parameter method
Optional string parameter reflecting the name of method inside of a service definition coming after rpc keyword

function KafkaLogger

KafkaLogger: (
    logger: any
) => ({
    namespace,
    level,
    label,
    log,
}: {
    namespace: any;
    level: any;
    label: any;
    log: any;
}) => void;

function Payload

Payload: {
    (): ParameterDecorator;
    (...pipes: any[]): ParameterDecorator;
    (propertyKey?: string, ...pipes: any[]): ParameterDecorator;
};

Microservice message pattern payload parameter decorator.
Microservice message pattern payload parameter decorator.
Example:
create(@Payload(new ValidationPipe()) createDto: CreateCatDto)
Parameter pipes
one or more pipes - either instances or classes - to apply to the bound parameter.
Microservice message pattern payload parameter decorator. Extracts a property from the payload object. May also apply pipes to the bound parameter.
For example, extracting all params:
findMany(@Payload() ids: string[])
For example, extracting a single param:
create(@Payload('data') createDto: { data: string })
For example, extracting a single param with pipe:
create(@Payload('data', new ValidationPipe()) createDto: { data: string })
Parameter propertyKey
name of single property to extract from the message payload
Parameter pipes
one or more pipes - either instances or classes - to apply to the bound parameter.

Classes

class BaseRpcContext

class BaseRpcContext<T = unknown[]> {}

constructor

constructor(args: {});

property args

protected readonly args: {};

method getArgByIndex

getArgByIndex: (index: number) => any;

Returns a particular argument by index.
Parameter index
index of argument to retrieve

method getArgs

getArgs: () => T;

Returns the array of arguments being passed to the handler.

class BaseRpcExceptionFilter

class BaseRpcExceptionFilter<T = any, R = any> implements RpcExceptionFilter<T> {}

method catch

catch: (exception: T, host: ArgumentsHost) => Observable<R>;

method handleUnknownError

handleUnknownError: (exception: T, status: string) => Observable<never>;

method isError

isError: (exception: any) => exception is Error;

class ClientGrpcProxy

class ClientGrpcProxy extends ClientProxy<never, never> implements ClientGrpc {}

constructor

constructor(options: {
    url?: string;
    maxSendMessageLength?: number;
    maxReceiveMessageLength?: number;
    maxMetadataSize?: number;
    keepalive?: {
        keepaliveTimeMs?: number;
        keepaliveTimeoutMs?: number;
        keepalivePermitWithoutCalls?: number;
        http2MaxPingsWithoutData?: number;
        http2MinTimeBetweenPingsMs?: number;
        http2MinPingIntervalWithoutDataMs?: number;
        http2MaxPingStrikes?: number;
    };
    channelOptions?: ChannelOptions;
    credentials?: any;
    protoPath?: string | string[];
    package: string | string[];
    protoLoader?: string;
    packageDefinition?: any;
    gracefulShutdown?: boolean;
    onLoadPackageDefinition?: (pkg: any, server: any) => void;
    loader?: {
        keepCase?: boolean;
        alternateCommentMode?: boolean;
        longs?: Function;
        enums?: Function;
        bytes?: Function;
        defaults?: boolean;
        arrays?: boolean;
        objects?: boolean;
        oneofs?: boolean;
        json?: boolean;
        includeDirs?: string[];
    };
});

property clients

protected readonly clients: Map<string, any>;

property grpcClients

protected grpcClients: any[];

property logger

protected readonly logger: Logger;

property options

protected readonly options: {
    url?: string;
    maxSendMessageLength?: number;
    maxReceiveMessageLength?: number;
    maxMetadataSize?: number;
    keepalive?: {
        keepaliveTimeMs?: number;
        keepaliveTimeoutMs?: number;
        keepalivePermitWithoutCalls?: number;
        http2MaxPingsWithoutData?: number;
        http2MinTimeBetweenPingsMs?: number;
        http2MinPingIntervalWithoutDataMs?: number;
        http2MaxPingStrikes?: number;
    };
    channelOptions?: ChannelOptions;
    credentials?: any;
    protoPath?: string | string[];
    package: string | string[];
    protoLoader?: string;
    packageDefinition?: any;
    gracefulShutdown?: boolean;
    onLoadPackageDefinition?: (pkg: any, server: any) => void;
    loader?: {
        keepCase?: boolean;
        alternateCommentMode?: boolean;
        longs?: Function;
        enums?: Function;
        bytes?: Function;
        defaults?: boolean;
        arrays?: boolean;
        objects?: boolean;
        oneofs?: boolean;
        json?: boolean;
        includeDirs?: string[];
    };
};

property status

readonly status: never;

property url

protected readonly url: string;

method close

close: () => void;

method connect

connect: () => Promise<any>;

method createClientByServiceName

createClientByServiceName: (name: string) => any;

method createClients

createClients: () => any[];

method createServiceMethod

createServiceMethod: (
    client: any,
    methodName: string
) => (...args: unknown[]) => Observable<unknown>;

method createStreamServiceMethod

createStreamServiceMethod: (
    client: unknown,
    methodName: string
) => (...args: any[]) => Observable<any>;

method createUnaryServiceMethod

createUnaryServiceMethod: (
    client: any,
    methodName: string
) => (...args: any[]) => Observable<any>;

method dispatchEvent

protected dispatchEvent: (packet: any) => Promise<any>;

method getClient

protected getClient: (name: string) => any;

method getClientByServiceName

getClientByServiceName: <T = unknown>(name: string) => T;

method getKeepaliveOptions

getKeepaliveOptions: () => {};

method getService

getService: <T extends object>(name: string) => T;

method loadProto

loadProto: () => any;

method lookupPackage

lookupPackage: (root: any, packageName: string) => any;

method on

on: <EventKey extends never = never, EventCallback = any>(
    event: EventKey,
    callback: EventCallback
) => void;

method publish

protected publish: (packet: any, callback: (packet: any) => any) => any;

method send

send: <TResult = any, TInput = any>(
    pattern: any,
    data: TInput
) => Observable<TResult>;

method unwrap

unwrap: <T>() => T;

class ClientKafka

class ClientKafka
    extends ClientProxy<never, KafkaStatus>
    implements ClientKafkaProxy {}

constructor

constructor(options: {
    postfixId?: string;
    client?: KafkaConfig;
    consumer?: ConsumerConfig;
    run?: Omit<ConsumerRunConfig, 'eachBatch' | 'eachMessage'>;
    subscribe?: Omit<ConsumerSubscribeTopics, 'topics'>;
    producer?: ProducerConfig;
    send?: Omit<ProducerRecord, 'topic' | 'messages'>;
    serializer?: Serializer<any, any>;
    deserializer?: Deserializer<any, any>;
    parser?: KafkaParserConfig;
    producerOnlyMode?: boolean;
});

property brokers

protected brokers: string[] | BrokersFunction;

property client

protected client: Kafka;

property clientId

protected clientId: string;

property consumer

readonly consumer: Consumer;

property consumerAssignments

protected consumerAssignments: { [key: string]: number };

property groupId

protected groupId: string;

property initialized

protected initialized: Promise<void>;

property logger

protected logger: Logger;

property options

protected readonly options: {
    postfixId?: string;
    client?: KafkaConfig;
    consumer?: ConsumerConfig;
    run?: Omit<ConsumerRunConfig, 'eachBatch' | 'eachMessage'>;
    subscribe?: Omit<ConsumerSubscribeTopics, 'topics'>;
    producer?: ProducerConfig;
    send?: Omit<ProducerRecord, 'topic' | 'messages'>;
    serializer?: Serializer<any, any>;
    deserializer?: Deserializer<any, any>;
    parser?: KafkaParserConfig;
    producerOnlyMode?: boolean;
};

property parser

protected parser: KafkaParser;

property producer

readonly producer: Producer;

property producerOnlyMode

protected producerOnlyMode: boolean;

property responsePatterns

protected responsePatterns: string[];

method bindTopics

bindTopics: () => Promise<void>;

method close

close: () => Promise<void>;

method commitOffsets

commitOffsets: (
    topicPartitions: TopicPartitionOffsetAndMetadata[]
) => Promise<void>;

method connect

connect: () => Promise<Producer>;

method createClient

createClient: <T = any>() => T;

method createResponseCallback

createResponseCallback: () => (payload: EachMessagePayload) => any;

method dispatchBatchEvent

protected dispatchBatchEvent: <TInput = any>(
    packets: ReadPacket<{ messages: TInput[] }>
) => Promise<any>;

method dispatchEvent

protected dispatchEvent: (packet: OutgoingEvent) => Promise<any>;

method emitBatch

emitBatch: <TResult = any, TInput = any>(
    pattern: any,
    data: { messages: TInput[] }
) => Observable<TResult>;

method getConsumerAssignments

getConsumerAssignments: () => { [key: string]: number };

method getReplyTopicPartition

protected getReplyTopicPartition: (topic: string) => string;

method getResponsePatternName

protected getResponsePatternName: (pattern: string) => string;

method initializeDeserializer

protected initializeDeserializer: (options: KafkaOptions['options']) => void;

method initializeSerializer

protected initializeSerializer: (options: KafkaOptions['options']) => void;

method on

on: <
    EventKey extends string | number | symbol = string | number | symbol,
    EventCallback = any
>(
    event: EventKey,
    callback: EventCallback
) => void;

method publish

protected publish: (
    partialPacket: ReadPacket,
    callback: (packet: WritePacket) => any
) => () => void;

method registerConsumerEventListeners

protected registerConsumerEventListeners: () => void;

method registerProducerEventListeners

protected registerProducerEventListeners: () => void;

method setConsumerAssignments

protected setConsumerAssignments: (data: ConsumerGroupJoinEvent) => void;

method subscribeToResponseOf

subscribeToResponseOf: (pattern: unknown) => void;

method unwrap

unwrap: <T>() => T;

class ClientMqtt

class ClientMqtt extends ClientProxy<MqttEvents, MqttStatus> {}

constructor

constructor(
    options: MqttClientOptions & {
        url?: string;
        serializer?: Serializer<any, any>;
        deserializer?: Deserializer<any, any>;
        subscribeOptions?: {
            qos: QoS;
            nl?: boolean;
            rap?: boolean;
            rh?: number;
        };
        userProperties?: Record<string, string | string[]>;
    }
);

property connectionPromise

protected connectionPromise: Promise<any>;

property isInitialConnection

protected isInitialConnection: boolean;

property isReconnecting

protected isReconnecting: boolean;

property logger

protected readonly logger: Logger;

property mqttClient

protected mqttClient: any;

property options

protected readonly options: MqttClientOptions & {
    url?: string;
    serializer?: Serializer<any, any>;
    deserializer?: Deserializer<any, any>;
    subscribeOptions?: { qos: QoS; nl?: boolean; rap?: boolean; rh?: number };
    userProperties?: Record<string, string | string[]>;
};

property pendingEventListeners

protected pendingEventListeners: {
    event: keyof MqttEvents;
    callback: MqttEvents[keyof MqttEvents];
}[];

property subscriptionsCount

protected readonly subscriptionsCount: Map<string, number>;

property url

protected readonly url: string;

method close

close: () => Promise<void>;

method connect

connect: () => Promise<any>;

method createClient

createClient: () => MqttClient;

method createResponseCallback

createResponseCallback: () => (channel: string, buffer: Buffer) => any;

method dispatchEvent

protected dispatchEvent: (packet: ReadPacket) => Promise<any>;

method getRequestPattern

getRequestPattern: (pattern: string) => string;

method getResponsePattern

getResponsePattern: (pattern: string) => string;

method initializeSerializer

protected initializeSerializer: (options: MqttOptions['options']) => void;

method mergeCloseEvent

mergeCloseEvent: <T = any>(
    instance: MqttClient,
    source$: Observable<T>
) => Observable<T>;

method mergePacketOptions

protected mergePacketOptions: (
    requestOptions?: MqttRecordOptions
) => MqttRecordOptions | undefined;

method on

on: <
    EventKey extends keyof MqttEvents = keyof MqttEvents,
    EventCallback extends MqttEvents[EventKey] = MqttEvents[EventKey]
>(
    event: EventKey,
    callback: EventCallback
) => void;

method publish

protected publish: (
    partialPacket: ReadPacket,
    callback: (packet: WritePacket) => any
) => () => void;

method registerCloseListener

registerCloseListener: (client: MqttClient) => void;

method registerConnectListener

registerConnectListener: (client: MqttClient) => void;

method registerDisconnectListener

registerDisconnectListener: (client: MqttClient) => void;

method registerErrorListener

registerErrorListener: (client: MqttClient) => void;

method registerOfflineListener

registerOfflineListener: (client: MqttClient) => void;

method registerReconnectListener

registerReconnectListener: (client: MqttClient) => void;

method unsubscribeFromChannel

protected unsubscribeFromChannel: (channel: string) => void;

method unwrap

unwrap: <T>() => T;

class ClientNats

class ClientNats extends ClientProxy<NatsEvents, NatsStatus> {}

constructor

constructor(options: {
    [key: string]: any;
    headers?: Record<string, string>;
    authenticator?: any;
    debug?: boolean;
    ignoreClusterUpdates?: boolean;
    inboxPrefix?: string;
    encoding?: string;
    name?: string;
    user?: string;
    pass?: string;
    maxPingOut?: number;
    maxReconnectAttempts?: number;
    reconnectTimeWait?: number;
    reconnectJitter?: number;
    reconnectJitterTLS?: number;
    reconnectDelayHandler?: any;
    servers?: string | string[];
    nkey?: any;
    reconnect?: boolean;
    pedantic?: boolean;
    tls?: any;
    queue?: string;
    serializer?: Serializer<any, any>;
    deserializer?: Deserializer<any, any>;
    userJWT?: string;
    nonceSigner?: any;
    userCreds?: any;
    useOldRequestStyle?: boolean;
    pingInterval?: number;
    preserveBuffers?: boolean;
    waitOnFirstConnect?: boolean;
    verbose?: boolean;
    noEcho?: boolean;
    noRandomize?: boolean;
    timeout?: number;
    token?: string;
    yieldTime?: number;
    tokenHandler?: any;
    gracefulShutdown?: boolean;
    gracePeriod?: number;
});

property connectionPromise

protected connectionPromise: Promise<Client>;

property logger

protected readonly logger: Logger;

property natsClient

protected natsClient: Client;

property options

protected readonly options: {
    [key: string]: any;
    headers?: Record<string, string>;
    authenticator?: any;
    debug?: boolean;
    ignoreClusterUpdates?: boolean;
    inboxPrefix?: string;
    encoding?: string;
    name?: string;
    user?: string;
    pass?: string;
    maxPingOut?: number;
    maxReconnectAttempts?: number;
    reconnectTimeWait?: number;
    reconnectJitter?: number;
    reconnectJitterTLS?: number;
    reconnectDelayHandler?: any;
    servers?: string | string[];
    nkey?: any;
    reconnect?: boolean;
    pedantic?: boolean;
    tls?: any;
    queue?: string;
    serializer?: Serializer<any, any>;
    deserializer?: Deserializer<any, any>;
    userJWT?: string;
    nonceSigner?: any;
    userCreds?: any;
    useOldRequestStyle?: boolean;
    pingInterval?: number;
    preserveBuffers?: boolean;
    waitOnFirstConnect?: boolean;
    verbose?: boolean;
    noEcho?: boolean;
    noRandomize?: boolean;
    timeout?: number;
    token?: string;
    yieldTime?: number;
    tokenHandler?: any;
    gracefulShutdown?: boolean;
    gracePeriod?: number;
};

property statusEventEmitter

protected statusEventEmitter: EventEmitter<{
    disconnect: [data?: string | number | undefined];
    reconnect: [data?: string | number | undefined];
    update: [
        data?:
            | string
            | number
            | import('../events/nats.events').ServersChangedEvent
            | undefined
    ];
}>;

method close

close: () => Promise<void>;

method connect

connect: () => Promise<any>;

method createClient

createClient: () => Promise<Client>;

method createSubscriptionHandler

createSubscriptionHandler: (
    packet: ReadPacket & PacketId,
    callback: (packet: WritePacket) => any
) => (error: string | Error | undefined, natsMsg: NatsMsg) => Promise<any>;

method dispatchEvent

protected dispatchEvent: (packet: ReadPacket) => Promise<any>;

method handleStatusUpdates

handleStatusUpdates: (client: Client) => Promise<void>;

method initializeDeserializer

protected initializeDeserializer: (options: NatsOptions['options']) => void;

method initializeSerializer

protected initializeSerializer: (options: NatsOptions['options']) => void;

method mergeHeaders

protected mergeHeaders: <THeaders = any>(requestHeaders?: THeaders) => any;

method on

on: <
    EventKey extends keyof NatsEvents = keyof NatsEvents,
    EventCallback extends NatsEvents[EventKey] = NatsEvents[EventKey]
>(
    event: EventKey,
    callback: EventCallback
) => void;

method publish

protected publish: (
    partialPacket: ReadPacket,
    callback: (packet: WritePacket) => any
) => () => void;

method unwrap

unwrap: <T>() => T;

class ClientProxy

abstract class ClientProxy<
    EventsMap extends Record<never, Function> = Record<never, Function>,
    Status extends string = string
> {}

property deserializer

protected deserializer: ProducerDeserializer;

property routingMap

protected routingMap: Map<string, Function>;

property serializer

protected serializer: ProducerSerializer;

property status

readonly status: Observable<Status>;

Returns an observable that emits status changes.

method assignPacketId

protected assignPacketId: (packet: ReadPacket) => ReadPacket & PacketId;

method close

abstract close: () => any;

Closes the underlying connection to the server/broker.

method connect

abstract connect: () => Promise<any>;

Establishes the connection to the underlying server/broker.

method connect$

protected connect$: (
    instance: any,
    errorEvent?: string,
    connectEvent?: string
) => Observable<any>;

method createObserver

protected createObserver: <T>(
    observer: Observer<T>
) => (packet: WritePacket) => void;

method dispatchEvent

protected abstract dispatchEvent: <T = any>(packet: ReadPacket) => Promise<T>;

method emit

emit: <TResult = any, TInput = any>(
    pattern: any,
    data: TInput
) => Observable<TResult>;

Emits an event to the server/broker. Used for event-driven communication style between microservices.
Parameter pattern
Pattern to identify the event
Parameter data
Data to be sent
Returns
Observable that completes when the event is successfully emitted

method getOptionsProp

protected getOptionsProp: {
    <
        Options extends
            | {
                  url?: string;
                  maxSendMessageLength?: number;
                  maxReceiveMessageLength?: number;
                  maxMetadataSize?: number;
                  keepalive?: {
                      keepaliveTimeMs?: number;
                      keepaliveTimeoutMs?: number;
                      keepalivePermitWithoutCalls?: number;
                      http2MaxPingsWithoutData?: number;
                      http2MinTimeBetweenPingsMs?: number;
                      http2MinPingIntervalWithoutDataMs?: number;
                      http2MaxPingStrikes?: number;
                  };
                  channelOptions?: ChannelOptions;
                  credentials?: any;
                  protoPath?: string | string[];
                  package: string | string[];
                  protoLoader?: string;
                  packageDefinition?: any;
                  gracefulShutdown?: boolean;
                  onLoadPackageDefinition?: (pkg: any, server: any) => void;
                  loader?: {
                      keepCase?: boolean;
                      alternateCommentMode?: boolean;
                      longs?: Function;
                      enums?: Function;
                      bytes?: Function;
                      defaults?: boolean;
                      arrays?: boolean;
                      objects?: boolean;
                      oneofs?: boolean;
                      json?: boolean;
                      includeDirs?: string[];
                  };
              }
            | {
                  postfixId?: string;
                  client?: KafkaConfig;
                  consumer?: ConsumerConfig;
                  run?: Omit<ConsumerRunConfig, 'eachBatch' | 'eachMessage'>;
                  subscribe?: Omit<ConsumerSubscribeTopics, 'topics'>;
                  producer?: ProducerConfig;
                  send?: Omit<ProducerRecord, 'topic' | 'messages'>;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
                  parser?: KafkaParserConfig;
                  producerOnlyMode?: boolean;
              }
            | (MqttClientOptions & {
                  url?: string;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
                  subscribeOptions?: {
                      qos: QoS;
                      nl?: boolean;
                      rap?: boolean;
                      rh?: number;
                  };
                  userProperties?: Record<string, string | string[]>;
              })
            | {
                  [key: string]: any;
                  headers?: Record<string, string>;
                  authenticator?: any;
                  debug?: boolean;
                  ignoreClusterUpdates?: boolean;
                  inboxPrefix?: string;
                  encoding?: string;
                  name?: string;
                  user?: string;
                  pass?: string;
                  maxPingOut?: number;
                  maxReconnectAttempts?: number;
                  reconnectTimeWait?: number;
                  reconnectJitter?: number;
                  reconnectJitterTLS?: number;
                  reconnectDelayHandler?: any;
                  servers?: string | string[];
                  nkey?: any;
                  reconnect?: boolean;
                  pedantic?: boolean;
                  tls?: any;
                  queue?: string;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
                  userJWT?: string;
                  nonceSigner?: any;
                  userCreds?: any;
                  useOldRequestStyle?: boolean;
                  pingInterval?: number;
                  preserveBuffers?: boolean;
                  waitOnFirstConnect?: boolean;
                  verbose?: boolean;
                  noEcho?: boolean;
                  noRandomize?: boolean;
                  timeout?: number;
                  token?: string;
                  yieldTime?: number;
                  tokenHandler?: any;
                  gracefulShutdown?: boolean;
                  gracePeriod?: number;
              }
            | ({
                  host?: string;
                  port?: number;
                  retryAttempts?: number;
                  retryDelay?: number;
                  wildcards?: boolean;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
              } & IORedisOptions)
            | {
                  host?: string;
                  port?: number;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
                  tlsOptions?: ConnectionOptions;
                  socketClass?: Type<TcpSocket>;
                  maxBufferSize?: number;
              }
            | {
                  urls?: string[] | RmqUrl[];
                  queue?: string;
                  prefetchCount?: number;
                  isGlobalPrefetchCount?: boolean;
                  queueOptions?: AmqplibQueueOptions;
                  socketOptions?: AmqpConnectionManagerSocketOptions;
                  noAck?: boolean;
                  consumerTag?: string;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
                  replyQueue?: string;
                  persistent?: boolean;
                  headers?: Record<string, string>;
                  noAssert?: boolean;
                  exchange?: string;
                  exchangeType?:
                      | 'topic'
                      | 'direct'
                      | 'fanout'
                      | 'headers'
                      | (string & {});
                  exchangeArguments?: Record<string, string>;
                  routingKey?: string;
                  wildcards?: boolean;
                  maxConnectionAttempts?: number;
              },
        Attribute extends keyof Options
    >(
        obj: Options,
        prop: Attribute
    ): Options[Attribute];
    <
        Options extends
            | {
                  url?: string;
                  maxSendMessageLength?: number;
                  maxReceiveMessageLength?: number;
                  maxMetadataSize?: number;
                  keepalive?: {
                      keepaliveTimeMs?: number;
                      keepaliveTimeoutMs?: number;
                      keepalivePermitWithoutCalls?: number;
                      http2MaxPingsWithoutData?: number;
                      http2MinTimeBetweenPingsMs?: number;
                      http2MinPingIntervalWithoutDataMs?: number;
                      http2MaxPingStrikes?: number;
                  };
                  channelOptions?: ChannelOptions;
                  credentials?: any;
                  protoPath?: string | string[];
                  package: string | string[];
                  protoLoader?: string;
                  packageDefinition?: any;
                  gracefulShutdown?: boolean;
                  onLoadPackageDefinition?: (pkg: any, server: any) => void;
                  loader?: {
                      keepCase?: boolean;
                      alternateCommentMode?: boolean;
                      longs?: Function;
                      enums?: Function;
                      bytes?: Function;
                      defaults?: boolean;
                      arrays?: boolean;
                      objects?: boolean;
                      oneofs?: boolean;
                      json?: boolean;
                      includeDirs?: string[];
                  };
              }
            | {
                  postfixId?: string;
                  client?: KafkaConfig;
                  consumer?: ConsumerConfig;
                  run?: Omit<ConsumerRunConfig, 'eachBatch' | 'eachMessage'>;
                  subscribe?: Omit<ConsumerSubscribeTopics, 'topics'>;
                  producer?: ProducerConfig;
                  send?: Omit<ProducerRecord, 'topic' | 'messages'>;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
                  parser?: KafkaParserConfig;
                  producerOnlyMode?: boolean;
              }
            | (MqttClientOptions & {
                  url?: string;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
                  subscribeOptions?: {
                      qos: QoS;
                      nl?: boolean;
                      rap?: boolean;
                      rh?: number;
                  };
                  userProperties?: Record<string, string | string[]>;
              })
            | {
                  [key: string]: any;
                  headers?: Record<string, string>;
                  authenticator?: any;
                  debug?: boolean;
                  ignoreClusterUpdates?: boolean;
                  inboxPrefix?: string;
                  encoding?: string;
                  name?: string;
                  user?: string;
                  pass?: string;
                  maxPingOut?: number;
                  maxReconnectAttempts?: number;
                  reconnectTimeWait?: number;
                  reconnectJitter?: number;
                  reconnectJitterTLS?: number;
                  reconnectDelayHandler?: any;
                  servers?: string | string[];
                  nkey?: any;
                  reconnect?: boolean;
                  pedantic?: boolean;
                  tls?: any;
                  queue?: string;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
                  userJWT?: string;
                  nonceSigner?: any;
                  userCreds?: any;
                  useOldRequestStyle?: boolean;
                  pingInterval?: number;
                  preserveBuffers?: boolean;
                  waitOnFirstConnect?: boolean;
                  verbose?: boolean;
                  noEcho?: boolean;
                  noRandomize?: boolean;
                  timeout?: number;
                  token?: string;
                  yieldTime?: number;
                  tokenHandler?: any;
                  gracefulShutdown?: boolean;
                  gracePeriod?: number;
              }
            | ({
                  host?: string;
                  port?: number;
                  retryAttempts?: number;
                  retryDelay?: number;
                  wildcards?: boolean;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
              } & IORedisOptions)
            | {
                  host?: string;
                  port?: number;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
                  tlsOptions?: ConnectionOptions;
                  socketClass?: Type<TcpSocket>;
                  maxBufferSize?: number;
              }
            | {
                  urls?: string[] | RmqUrl[];
                  queue?: string;
                  prefetchCount?: number;
                  isGlobalPrefetchCount?: boolean;
                  queueOptions?: AmqplibQueueOptions;
                  socketOptions?: AmqpConnectionManagerSocketOptions;
                  noAck?: boolean;
                  consumerTag?: string;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
                  replyQueue?: string;
                  persistent?: boolean;
                  headers?: Record<string, string>;
                  noAssert?: boolean;
                  exchange?: string;
                  exchangeType?:
                      | 'topic'
                      | 'direct'
                      | 'fanout'
                      | 'headers'
                      | (string & {});
                  exchangeArguments?: Record<string, string>;
                  routingKey?: string;
                  wildcards?: boolean;
                  maxConnectionAttempts?: number;
              },
        Attribute extends keyof Options,
        DefaultValue extends Options[Attribute] = Options[Attribute]
    >(
        obj: Options,
        prop: Attribute,
        defaultValue: DefaultValue
    ): Required<Options>[Attribute];
};

method initializeDeserializer

protected initializeDeserializer: (options: ClientOptions['options']) => void;

method initializeSerializer

protected initializeSerializer: (options: ClientOptions['options']) => void;

method normalizePattern

protected normalizePattern: (pattern: MsPattern) => string;

method on

on: <
    EventKey extends keyof EventsMap = keyof EventsMap,
    EventCallback extends EventsMap[EventKey] = EventsMap[EventKey]
>(
    event: EventKey,
    callback: EventCallback
) => void;

Registers an event listener for the given event.
Parameter event
Event name
Parameter callback
Callback to be executed when the event is emitted

method publish

protected abstract publish: (
    packet: ReadPacket,
    callback: (packet: WritePacket) => void
) => () => void;

method send

send: <TResult = any, TInput = any>(
    pattern: any,
    data: TInput
) => Observable<TResult>;

Send a message to the server/broker. Used for message-driven communication style between microservices.
Parameter pattern
Pattern to identify the message
Parameter data
Data to be sent
Returns
Observable with the result

method serializeError

protected serializeError: (err: any) => any;

method serializeResponse

protected serializeResponse: (response: any) => any;

method unwrap

abstract unwrap: <T>() => T;

Returns an instance of the underlying server/broker instance, or a group of servers if there are more than one.

class ClientProxyFactory

class ClientProxyFactory {}

method create

static create: {
    (
        clientOptions: { transport: Transport.GRPC } & ClientOptions
    ): ClientGrpcProxy;
    (
        clientOptions: { transport: Transport.KAFKA } & KafkaOptions
    ): ClientKafkaProxy;
    (clientOptions: ClientOptions): ClientProxy<Record<never, Function>, string>;
    (clientOptions: CustomClientOptions): ClientProxy<
        Record<never, Function>,
        string
    >;
};

class ClientRedis

class ClientRedis extends ClientProxy<RedisEvents, RedisStatus> {}

constructor

constructor(
    options: {
        host?: string;
        port?: number;
        retryAttempts?: number;
        retryDelay?: number;
        wildcards?: boolean;
        serializer?: Serializer<any, any>;
        deserializer?: Deserializer<any, any>;
    } & IORedisOptions &
        RedisOutputOptions
);

property connectionPromise

protected connectionPromise: Promise<any>;

property isManuallyClosed

protected isManuallyClosed: boolean;

property logger

protected readonly logger: Logger;

property options

protected readonly options: {
    host?: string;
    port?: number;
    retryAttempts?: number;
    retryDelay?: number;
    wildcards?: boolean;
    serializer?: Serializer<any, any>;
    deserializer?: Deserializer<any, any>;
} & IORedisOptions &
    RedisOutputOptions;

property pendingEventListeners

protected pendingEventListeners: {
    event: keyof RedisEvents;
    callback: RedisEvents[keyof RedisEvents];
}[];

property pubClient

protected pubClient: any;

property subClient

protected subClient: any;

property subscriptionsCount

protected readonly subscriptionsCount: Map<string, number>;

property wasInitialConnectionSuccessful

protected wasInitialConnectionSuccessful: boolean;

method close

close: () => Promise<void>;

method connect

connect: () => Promise<any>;

method createClient

createClient: () => Redis;

method createResponseCallback

createResponseCallback: () => (channel: string, buffer: string) => Promise<void>;

method createRetryStrategy

createRetryStrategy: (times: number) => undefined | number;

method dispatchEvent

protected dispatchEvent: (packet: ReadPacket) => Promise<any>;

method getClientOptions

getClientOptions: () => Partial<RedisOptions['options']>;

method getReplyPattern

getReplyPattern: (pattern: string) => string;

method getRequestPattern

getRequestPattern: (pattern: string) => string;

method handleClose

handleClose: () => void;

method on

on: <
    EventKey extends keyof RedisEvents = keyof RedisEvents,
    EventCallback extends RedisEvents[EventKey] = RedisEvents[EventKey]
>(
    event: EventKey,
    callback: EventCallback
) => void;

method publish

protected publish: (
    partialPacket: ReadPacket,
    callback: (packet: WritePacket) => any
) => () => void;

method registerEndListener

registerEndListener: (client: {
    on: (event: string, fn: () => void) => void;
}) => void;

method registerErrorListener

registerErrorListener: (client: Redis) => void;

method registerReadyListener

registerReadyListener: (client: {
    on: (event: string, fn: () => void) => void;
}) => void;

method registerReconnectListener

registerReconnectListener: (client: {
    on: (event: string, fn: () => void) => void;
}) => void;

method unsubscribeFromChannel

protected unsubscribeFromChannel: (channel: string) => void;

method unwrap

unwrap: <T>() => T;

class ClientRMQ

class ClientRMQ extends ClientProxy<RmqEvents, RmqStatus> {}

constructor

constructor(options: {
    urls?: string[] | RmqUrl[];
    queue?: string;
    prefetchCount?: number;
    isGlobalPrefetchCount?: boolean;
    queueOptions?: AmqplibQueueOptions;
    socketOptions?: AmqpConnectionManagerSocketOptions;
    noAck?: boolean;
    consumerTag?: string;
    serializer?: Serializer<any, any>;
    deserializer?: Deserializer<any, any>;
    replyQueue?: string;
    persistent?: boolean;
    headers?: Record<string, string>;
    noAssert?: boolean;
    exchange?: string;
    exchangeType?: 'topic' | 'direct' | 'fanout' | 'headers' | (string & {});
    exchangeArguments?: Record<string, string>;
    routingKey?: string;
    wildcards?: boolean;
    maxConnectionAttempts?: number;
});

property channel

protected channel: any;

property client

protected client: any;

property connection$

protected connection$: ReplaySubject<any>;

property connectionPromise

protected connectionPromise: Promise<void>;

property isInitialConnect

protected isInitialConnect: boolean;

property logger

protected readonly logger: Logger;

property noAssert

protected noAssert: boolean;

property options

protected readonly options: {
    urls?: string[] | RmqUrl[];
    queue?: string;
    prefetchCount?: number;
    isGlobalPrefetchCount?: boolean;
    queueOptions?: AmqplibQueueOptions;
    socketOptions?: AmqpConnectionManagerSocketOptions;
    noAck?: boolean;
    consumerTag?: string;
    serializer?: Serializer<any, any>;
    deserializer?: Deserializer<any, any>;
    replyQueue?: string;
    persistent?: boolean;
    headers?: Record<string, string>;
    noAssert?: boolean;
    exchange?: string;
    exchangeType?: 'topic' | 'direct' | 'fanout' | 'headers' | (string & {});
    exchangeArguments?: Record<string, string>;
    routingKey?: string;
    wildcards?: boolean;
    maxConnectionAttempts?: number;
};

property pendingEventListeners

protected pendingEventListeners: {
    event: keyof RmqEvents;
    callback: RmqEvents[keyof RmqEvents];
}[];

property queue

protected queue: string;

property queueOptions

protected queueOptions: Record<string, any>;

property replyQueue

protected replyQueue: string;

property responseEmitter

protected responseEmitter: EventEmitter;

method close

close: () => Promise<void>;

method connect

connect: () => Promise<any>;

method consumeChannel

consumeChannel: (channel: Channel) => Promise<void>;

method convertConnectionToPromise

convertConnectionToPromise: () => Promise<any>;

method createChannel

createChannel: () => Promise<void>;

method createClient

createClient: () => AmqpConnectionManager;

method dispatchEvent

protected dispatchEvent: (packet: ReadPacket) => Promise<any>;

method handleMessage

handleMessage: {
    (packet: unknown, callback: (packet: WritePacket) => any): Promise<void>;
    (
        packet: unknown,
        options: Record<string, unknown>,
        callback: (packet: WritePacket<any>) => any
    ): Promise<void>;
};

method initializeSerializer

protected initializeSerializer: (options: RmqOptions['options']) => void;

method mergeDisconnectEvent

mergeDisconnectEvent: <T = any>(
    instance: any,
    source$: Observable<T>
) => Observable<T>;

method mergeHeaders

protected mergeHeaders: (
    requestHeaders?: Record<string, string>
) => Record<string, string> | undefined;

method on

on: <
    EventKey extends keyof RmqEvents = keyof RmqEvents,
    EventCallback extends RmqEvents[EventKey] = RmqEvents[EventKey]
>(
    event: EventKey,
    callback: EventCallback
) => void;

method parseMessageContent

protected parseMessageContent: (content: Buffer) => any;

method publish

protected publish: (
    message: ReadPacket,
    callback: (packet: WritePacket) => any
) => () => void;

method registerBlockedListener

registerBlockedListener: (client: AmqpConnectionManager) => void;

method registerDisconnectListener

registerDisconnectListener: (client: AmqpConnectionManager) => void;

method registerErrorListener

registerErrorListener: (client: AmqpConnectionManager) => void;

method registerUnblockedListener

registerUnblockedListener: (client: AmqpConnectionManager) => void;

method setupChannel

setupChannel: (channel: Channel, resolve: Function) => Promise<void>;

method unwrap

unwrap: <T>() => T;

class ClientsModule

class ClientsModule {}

method register

static register: (options: ClientsModuleOptions) => DynamicModule;

method registerAsync

static registerAsync: (options: ClientsModuleAsyncOptions) => DynamicModule;

class ClientTCP

class ClientTCP extends ClientProxy<TcpEvents, TcpStatus> {}

constructor

constructor(options: {
    host?: string;
    port?: number;
    serializer?: Serializer<any, any>;
    deserializer?: Deserializer<any, any>;
    tlsOptions?: ConnectionOptions;
    socketClass?: Type<TcpSocket>;
    maxBufferSize?: number;
});

property connectionPromise

protected connectionPromise: Promise<any>;

property host

protected readonly host: string;

property logger

protected readonly logger: Logger;

property maxBufferSize

protected readonly maxBufferSize?: number;

property pendingEventListeners

protected pendingEventListeners: {
    event: keyof TcpEvents;
    callback: TcpEvents[keyof TcpEvents];
}[];

property port

protected readonly port: number;

property socket

protected socket: TcpSocket;

property socketClass

protected readonly socketClass: Type<TcpSocket>;

property tlsOptions

protected readonly tlsOptions?: ConnectionOptions;

method close

close: () => void;

method connect

connect: () => Promise<any>;

method createSocket

createSocket: () => TcpSocket;

method dispatchEvent

protected dispatchEvent: (packet: ReadPacket) => Promise<any>;

method handleClose

handleClose: () => void;

method handleError

handleError: (err: any) => void;

method handleResponse

handleResponse: (buffer: unknown) => Promise<void>;

method on

on: <
    EventKey extends keyof TcpEvents = keyof TcpEvents,
    EventCallback extends TcpEvents[EventKey] = TcpEvents[EventKey]
>(
    event: EventKey,
    callback: EventCallback
) => void;

method publish

protected publish: (
    partialPacket: ReadPacket,
    callback: (packet: WritePacket) => any
) => () => void;

method registerCloseListener

registerCloseListener: (socket: TcpSocket) => void;

method registerConnectListener

registerConnectListener: (socket: TcpSocket) => void;

method registerErrorListener

registerErrorListener: (socket: TcpSocket) => void;

method unwrap

unwrap: <T>() => T;

class JsonSocket

class JsonSocket extends TcpSocket {}

constructor

constructor(socket: any, options?: JsonSocketOptions);

method handleData

protected handleData: (dataRaw: Buffer | string) => void;

method handleSend

protected handleSend: (message: any, callback?: (err?: any) => void) => void;

class KafkaContext

class KafkaContext extends BaseRpcContext<KafkaContextArgs> {}

constructor

constructor(args: KafkaContextArgs);

method getConsumer

getConsumer: () => Consumer;

Returns the Kafka consumer reference.

method getHeartbeat

getHeartbeat: () => () => Promise<void>;

Returns the Kafka heartbeat callback.

method getMessage

getMessage: () => KafkaMessage;

Returns the reference to the original message.

method getPartition

getPartition: () => number;

Returns the partition.

method getProducer

getProducer: () => Producer;

Returns the Kafka producer reference,

method getTopic

getTopic: () => string;

Returns the name of the topic.

class KafkaParser

class KafkaParser {}

constructor

constructor(config?: KafkaParserConfig);

property keepBinary

protected readonly keepBinary: boolean;

method decode

decode: (value: Buffer) => object | string | null | Buffer;

method parse

parse: <T = any>(data: any) => T;

class KafkaReplyPartitionAssigner

class KafkaReplyPartitionAssigner {}

constructor

constructor(clientKafka: ClientKafka, config: { cluster: Cluster });

property name

readonly name: string;

property version

readonly version: number;

method assign

assign: (group: {
    members: GroupMember[];
    topics: string[];
}) => Promise<GroupMemberAssignment[]>;

This process can result in imbalanced assignments
Parameter members
array of members, e.g: [{ memberId: 'test-5f93f5a3' }]
Parameter topics
Parameter userData
Returns
{array} object partitions per topic per member

method decodeMember

decodeMember: (member: GroupMember) => {
    memberId: string;
    previousAssignment: any;
};

method getPreviousAssignment

getPreviousAssignment: () => { [key: string]: number };

method protocol

protocol: (subscription: { topics: string[]; userData: Buffer }) => GroupState;

class KafkaRetriableException

class KafkaRetriableException extends RpcException {}

Exception that instructs Kafka driver to instead of introspecting error processing flow and sending serialized error message to the consumer, force bubble it up to the "eachMessage" callback of the underlying "kafkajs" package (even if interceptors are applied, or an observable stream is returned from the message handler).
A transient exception that if retried may succeed.

method getError

getError: () => string | object;

class MqttContext

class MqttContext extends BaseRpcContext<MqttContextArgs> {}

constructor

constructor(args: MqttContextArgs);

method getPacket

getPacket: () => Record<string, any>;

Returns the reference to the original MQTT packet.

method getTopic

getTopic: () => string;

Returns the name of the topic.

class MqttRecord

class MqttRecord<TData = any> {}

constructor

constructor(data: {}, options?: MqttRecordOptions);

property data

readonly data: {};

property options

options?: MqttRecordOptions;

class MqttRecordBuilder

class MqttRecordBuilder<TData> {}

constructor

constructor(data?: {});

method build

build: () => MqttRecord;

method setData

setData: (data: TData) => this;

method setDup

setDup: (dup: MqttRecordOptions['dup']) => this;

method setProperties

setProperties: (properties: MqttRecordOptions['properties']) => this;

method setQoS

setQoS: (qos: MqttRecordOptions['qos']) => this;

method setRetain

setRetain: (retain: MqttRecordOptions['retain']) => this;

class NatsContext

class NatsContext extends BaseRpcContext<NatsContextArgs> {}

constructor

constructor(args: NatsContextArgs);

method getHeaders

getHeaders: () => any;

Returns message headers (if exist).

method getSubject

getSubject: () => string;

Returns the name of the subject.

class NatsRecord

class NatsRecord<TData = any, THeaders = any> {}

constructor

constructor(data: {}, headers?: {});

property data

readonly data: {};

property headers

readonly headers?: {};

class NatsRecordBuilder

class NatsRecordBuilder<TData> {}

constructor

constructor(data?: {});

method build

build: () => NatsRecord;

method setData

setData: (data: TData) => this;

method setHeaders

setHeaders: <THeaders = any>(headers: THeaders) => this;

class NestMicroservice

class NestMicroservice
    extends NestApplicationContext<NestMicroserviceOptions>
    implements INestMicroservice {}

constructor

constructor(
    container: NestContainer,
    config: any,
    graphInspector: GraphInspector,
    applicationConfig: ApplicationConfig
);

property logger

protected readonly logger: Logger;

property status

readonly status: any;

Returns an observable that emits status changes.

method close

close: () => Promise<any>;

Terminates the application.
Returns
{Promise}

method closeApplication

protected closeApplication: () => Promise<any>;

method createServer

createServer: (config: CompleteMicroserviceOptions) => void;

method dispose

protected dispose: () => Promise<void>;

method init

init: () => Promise<this>;

method listen

listen: () => Promise<any>;

Starts the microservice.
Returns
{void}

method on

on: (event: string | number | symbol, callback: Function) => any;

Registers an event listener for the given event.
Parameter event
Event name
Parameter callback
Callback to be executed when the event is emitted

method registerListeners

registerListeners: () => void;

method registerModules

registerModules: () => Promise<any>;

method resolveAsyncOptions

protected resolveAsyncOptions: (
    config: AsyncMicroserviceOptions
) => MicroserviceOptions;

method setIsInitHookCalled

setIsInitHookCalled: (isInitHookCalled: boolean) => void;

Sets the flag indicating that the init hook was called.
Parameter isInitHookCalled
Value to set

method setIsInitialized

setIsInitialized: (isInitialized: boolean) => void;

Sets the flag indicating that the application is initialized.
Parameter isInitialized
Value to set

method setIsTerminated

setIsTerminated: (isTerminated: boolean) => void;

Sets the flag indicating that the application is terminated.
Parameter isTerminated
Value to set

method unwrap

unwrap: <T>() => T;

Returns an instance of the underlying server/broker instance, or a group of servers if there are more than one.

method useGlobalFilters

useGlobalFilters: (...filters: ExceptionFilter[]) => this;

Registers global exception filters (will be used for every pattern handler).
Parameter filters

method useGlobalGuards

useGlobalGuards: (...guards: CanActivate[]) => this;

method useGlobalInterceptors

useGlobalInterceptors: (...interceptors: NestInterceptor[]) => this;

Registers global interceptors (will be used for every pattern handler).
Parameter interceptors

method useGlobalPipes

useGlobalPipes: (...pipes: PipeTransform<any>[]) => this;

Registers global pipes (will be used for every pattern handler).
Parameter pipes

method useWebSocketAdapter

useWebSocketAdapter: (adapter: WebSocketAdapter) => this;

Registers a web socket adapter that will be used for Gateways. Use to override the default socket.io library.
Parameter adapter
Returns
{this}

class RedisContext

class RedisContext extends BaseRpcContext<RedisContextArgs> {}

constructor

constructor(args: RedisContextArgs);

method getChannel

getChannel: () => string;

Returns the name of the channel.

class RmqContext

class RmqContext extends BaseRpcContext<RmqContextArgs> {}

constructor

constructor(args: RmqContextArgs);

method getChannelRef

getChannelRef: () => any;

Returns the reference to the original RMQ channel.

method getMessage

getMessage: () => Record<string, any>;

Returns the original message (with properties, fields, and content).

method getPattern

getPattern: () => string;

Returns the name of the pattern.

class RmqRecord

class RmqRecord<TData = any> {}

constructor

constructor(data: {}, options?: RmqRecordOptions);

property data

readonly data: {};

property options

options?: RmqRecordOptions;

class RmqRecordBuilder

class RmqRecordBuilder<TData> {}

constructor

constructor(data?: {});

method build

build: () => RmqRecord;

method setData

setData: (data: TData) => this;

method setOptions

setOptions: (options: RmqRecordOptions) => this;

class RpcException

class RpcException extends Error {}

constructor

constructor(error: string | object);

method getError

getError: () => string | object;

method initMessage

initMessage: () => void;

class Server

abstract class Server<
    EventsMap extends Record<string, Function> = Record<string, Function>,
    Status extends string = string
> {}

property deserializer

protected deserializer: ConsumerDeserializer;

property logger

protected readonly logger: LoggerService;

property messageHandlers

protected readonly messageHandlers: Map<string, MessageHandler<any, any, any>>;

property onProcessingEndHook

protected onProcessingEndHook: (
    transportId: Transport | symbol,
    context: BaseRpcContext
) => void;

property onProcessingStartHook

protected onProcessingStartHook: (
    transportId: Transport | symbol,
    context: BaseRpcContext,
    done: () => Promise<any>
) => void;

property serializer

protected serializer: ConsumerSerializer;

property status

readonly status: Observable<Status>;

Returns an observable that emits status changes.

property transportId

transportId?: symbol | Transport;

Unique transport identifier.

method addHandler

addHandler: (
    pattern: any,
    callback: MessageHandler,
    isEventHandler?: boolean,
    extras?: Record<string, any>
) => void;

method close

abstract close: () => any;

Method called when server is being terminated.

method getHandlerByPattern

getHandlerByPattern: (pattern: string) => MessageHandler | null;

method getHandlers

getHandlers: () => Map<string, MessageHandler>;

method getOptionsProp

getOptionsProp: {
    <
        Options extends
            | {
                  url?: string;
                  maxSendMessageLength?: number;
                  maxReceiveMessageLength?: number;
                  maxMetadataSize?: number;
                  keepalive?: {
                      keepaliveTimeMs?: number;
                      keepaliveTimeoutMs?: number;
                      keepalivePermitWithoutCalls?: number;
                      http2MaxPingsWithoutData?: number;
                      http2MinTimeBetweenPingsMs?: number;
                      http2MinPingIntervalWithoutDataMs?: number;
                      http2MaxPingStrikes?: number;
                  };
                  channelOptions?: ChannelOptions;
                  credentials?: any;
                  protoPath?: string | string[];
                  package: string | string[];
                  protoLoader?: string;
                  packageDefinition?: any;
                  gracefulShutdown?: boolean;
                  onLoadPackageDefinition?: (pkg: any, server: any) => void;
                  loader?: {
                      keepCase?: boolean;
                      alternateCommentMode?: boolean;
                      longs?: Function;
                      enums?: Function;
                      bytes?: Function;
                      defaults?: boolean;
                      arrays?: boolean;
                      objects?: boolean;
                      oneofs?: boolean;
                      json?: boolean;
                      includeDirs?: string[];
                  };
              }
            | {
                  postfixId?: string;
                  client?: KafkaConfig;
                  consumer?: ConsumerConfig;
                  run?: Omit<ConsumerRunConfig, 'eachBatch' | 'eachMessage'>;
                  subscribe?: Omit<ConsumerSubscribeTopics, 'topics'>;
                  producer?: ProducerConfig;
                  send?: Omit<ProducerRecord, 'topic' | 'messages'>;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
                  parser?: KafkaParserConfig;
                  producerOnlyMode?: boolean;
              }
            | (MqttClientOptions & {
                  url?: string;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
                  subscribeOptions?: {
                      qos: QoS;
                      nl?: boolean;
                      rap?: boolean;
                      rh?: number;
                  };
                  userProperties?: Record<string, string | string[]>;
              })
            | {
                  [key: string]: any;
                  headers?: Record<string, string>;
                  authenticator?: any;
                  debug?: boolean;
                  ignoreClusterUpdates?: boolean;
                  inboxPrefix?: string;
                  encoding?: string;
                  name?: string;
                  user?: string;
                  pass?: string;
                  maxPingOut?: number;
                  maxReconnectAttempts?: number;
                  reconnectTimeWait?: number;
                  reconnectJitter?: number;
                  reconnectJitterTLS?: number;
                  reconnectDelayHandler?: any;
                  servers?: string | string[];
                  nkey?: any;
                  reconnect?: boolean;
                  pedantic?: boolean;
                  tls?: any;
                  queue?: string;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
                  userJWT?: string;
                  nonceSigner?: any;
                  userCreds?: any;
                  useOldRequestStyle?: boolean;
                  pingInterval?: number;
                  preserveBuffers?: boolean;
                  waitOnFirstConnect?: boolean;
                  verbose?: boolean;
                  noEcho?: boolean;
                  noRandomize?: boolean;
                  timeout?: number;
                  token?: string;
                  yieldTime?: number;
                  tokenHandler?: any;
                  gracefulShutdown?: boolean;
                  gracePeriod?: number;
              }
            | ({
                  host?: string;
                  port?: number;
                  retryAttempts?: number;
                  retryDelay?: number;
                  wildcards?: boolean;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
              } & IORedisOptions)
            | {
                  urls?: string[] | RmqUrl[];
                  queue?: string;
                  prefetchCount?: number;
                  isGlobalPrefetchCount?: boolean;
                  queueOptions?: AmqplibQueueOptions;
                  socketOptions?: AmqpConnectionManagerSocketOptions;
                  noAck?: boolean;
                  consumerTag?: string;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
                  replyQueue?: string;
                  persistent?: boolean;
                  headers?: Record<string, string>;
                  noAssert?: boolean;
                  exchange?: string;
                  exchangeType?:
                      | 'topic'
                      | 'direct'
                      | 'fanout'
                      | 'headers'
                      | (string & {});
                  exchangeArguments?: Record<string, string>;
                  routingKey?: string;
                  wildcards?: boolean;
                  maxConnectionAttempts?: number;
              }
            | Record<string, any>
            | {
                  host?: string;
                  port?: number;
                  retryAttempts?: number;
                  retryDelay?: number;
                  serializer?: Serializer<any, any>;
                  tlsOptions?: TlsOptions;
                  deserializer?: Deserializer<any, any>;
                  socketClass?: Type<TcpSocket>;
                  maxBufferSize?: number;
              },
        Attribute extends keyof Options
    >(
        obj: Options,
        prop: Attribute
    ): Options[Attribute];
    <
        Options extends
            | {
                  url?: string;
                  maxSendMessageLength?: number;
                  maxReceiveMessageLength?: number;
                  maxMetadataSize?: number;
                  keepalive?: {
                      keepaliveTimeMs?: number;
                      keepaliveTimeoutMs?: number;
                      keepalivePermitWithoutCalls?: number;
                      http2MaxPingsWithoutData?: number;
                      http2MinTimeBetweenPingsMs?: number;
                      http2MinPingIntervalWithoutDataMs?: number;
                      http2MaxPingStrikes?: number;
                  };
                  channelOptions?: ChannelOptions;
                  credentials?: any;
                  protoPath?: string | string[];
                  package: string | string[];
                  protoLoader?: string;
                  packageDefinition?: any;
                  gracefulShutdown?: boolean;
                  onLoadPackageDefinition?: (pkg: any, server: any) => void;
                  loader?: {
                      keepCase?: boolean;
                      alternateCommentMode?: boolean;
                      longs?: Function;
                      enums?: Function;
                      bytes?: Function;
                      defaults?: boolean;
                      arrays?: boolean;
                      objects?: boolean;
                      oneofs?: boolean;
                      json?: boolean;
                      includeDirs?: string[];
                  };
              }
            | {
                  postfixId?: string;
                  client?: KafkaConfig;
                  consumer?: ConsumerConfig;
                  run?: Omit<ConsumerRunConfig, 'eachBatch' | 'eachMessage'>;
                  subscribe?: Omit<ConsumerSubscribeTopics, 'topics'>;
                  producer?: ProducerConfig;
                  send?: Omit<ProducerRecord, 'topic' | 'messages'>;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
                  parser?: KafkaParserConfig;
                  producerOnlyMode?: boolean;
              }
            | (MqttClientOptions & {
                  url?: string;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
                  subscribeOptions?: {
                      qos: QoS;
                      nl?: boolean;
                      rap?: boolean;
                      rh?: number;
                  };
                  userProperties?: Record<string, string | string[]>;
              })
            | {
                  [key: string]: any;
                  headers?: Record<string, string>;
                  authenticator?: any;
                  debug?: boolean;
                  ignoreClusterUpdates?: boolean;
                  inboxPrefix?: string;
                  encoding?: string;
                  name?: string;
                  user?: string;
                  pass?: string;
                  maxPingOut?: number;
                  maxReconnectAttempts?: number;
                  reconnectTimeWait?: number;
                  reconnectJitter?: number;
                  reconnectJitterTLS?: number;
                  reconnectDelayHandler?: any;
                  servers?: string | string[];
                  nkey?: any;
                  reconnect?: boolean;
                  pedantic?: boolean;
                  tls?: any;
                  queue?: string;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
                  userJWT?: string;
                  nonceSigner?: any;
                  userCreds?: any;
                  useOldRequestStyle?: boolean;
                  pingInterval?: number;
                  preserveBuffers?: boolean;
                  waitOnFirstConnect?: boolean;
                  verbose?: boolean;
                  noEcho?: boolean;
                  noRandomize?: boolean;
                  timeout?: number;
                  token?: string;
                  yieldTime?: number;
                  tokenHandler?: any;
                  gracefulShutdown?: boolean;
                  gracePeriod?: number;
              }
            | ({
                  host?: string;
                  port?: number;
                  retryAttempts?: number;
                  retryDelay?: number;
                  wildcards?: boolean;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
              } & IORedisOptions)
            | {
                  urls?: string[] | RmqUrl[];
                  queue?: string;
                  prefetchCount?: number;
                  isGlobalPrefetchCount?: boolean;
                  queueOptions?: AmqplibQueueOptions;
                  socketOptions?: AmqpConnectionManagerSocketOptions;
                  noAck?: boolean;
                  consumerTag?: string;
                  serializer?: Serializer<any, any>;
                  deserializer?: Deserializer<any, any>;
                  replyQueue?: string;
                  persistent?: boolean;
                  headers?: Record<string, string>;
                  noAssert?: boolean;
                  exchange?: string;
                  exchangeType?:
                      | 'topic'
                      | 'direct'
                      | 'fanout'
                      | 'headers'
                      | (string & {});
                  exchangeArguments?: Record<string, string>;
                  routingKey?: string;
                  wildcards?: boolean;
                  maxConnectionAttempts?: number;
              }
            | Record<string, any>
            | {
                  host?: string;
                  port?: number;
                  retryAttempts?: number;
                  retryDelay?: number;
                  serializer?: Serializer<any, any>;
                  tlsOptions?: TlsOptions;
                  deserializer?: Deserializer<any, any>;
                  socketClass?: Type<TcpSocket>;
                  maxBufferSize?: number;
              },
        Attribute extends keyof Options,
        DefaultValue extends Options[Attribute] = Options[Attribute]
    >(
        obj: Options,
        prop: Attribute,
        defaultValue: DefaultValue
    ): Required<Options>[Attribute];
};

method getRouteFromPattern

protected getRouteFromPattern: (pattern: string) => string;

Transforms the server Pattern to valid type and returns a route for him.
Parameter pattern
server pattern
Returns
string

method handleError

protected handleError: (error: string) => void;

method handleEvent

handleEvent: (
    pattern: string,
    packet: ReadPacket,
    context: BaseRpcContext
) => Promise<any>;

method initializeDeserializer

protected initializeDeserializer: (options: ClientOptions['options']) => void;

method initializeSerializer

protected initializeSerializer: (options: ClientOptions['options']) => void;

method listen

abstract listen: (callback: (...optionalParams: unknown[]) => any) => any;

Method called when server is being initialized.
Parameter callback
Function to be called upon initialization

method loadPackage

protected loadPackage: <T = any>(
    name: string,
    ctx: string,
    loader?: Function
) => T;

method normalizePattern

protected normalizePattern: (pattern: MsPattern) => string;

method on

abstract on: <
    EventKey extends keyof EventsMap = keyof EventsMap,
    EventCallback extends EventsMap[EventKey] = EventsMap[EventKey]
>(
    event: EventKey,
    callback: EventCallback
) => any;

Registers an event listener for the given event.
Parameter event
Event name
Parameter callback
Callback to be executed when the event is emitted

method send

send: (
    stream$: Observable<any>,
    respond: (data: WritePacket) => Promise<unknown> | void
) => Subscription;

method setOnProcessingEndHook

setOnProcessingEndHook: (
    hook: (transportId: Transport | symbol, context: unknown) => void
) => void;

Sets a hook that will be called when processing ends.

method setOnProcessingStartHook

setOnProcessingStartHook: (
    hook: (
        transportId: Transport | symbol,
        context: unknown,
        done: () => Promise<any>
    ) => void
) => void;

Sets a hook that will be called when processing starts.

method setTransportId

setTransportId: (transportId: Transport | symbol) => void;

Sets the transport identifier.
Parameter transportId
Unique transport identifier.

method transformToObservable

transformToObservable: {
    <T>(resultOrDeferred: Observable<T> | Promise<T>): Observable<T>;
    <T>(resultOrDeferred: T): Observable<T>;
};

method unwrap

abstract unwrap: <T>() => T;

Returns an instance of the underlying server/broker instance, or a group of servers if there are more than one.

class ServerGrpc

class ServerGrpc extends Server<never, never> {}

constructor

constructor(options: {
    url?: string;
    maxSendMessageLength?: number;
    maxReceiveMessageLength?: number;
    maxMetadataSize?: number;
    keepalive?: {
        keepaliveTimeMs?: number;
        keepaliveTimeoutMs?: number;
        keepalivePermitWithoutCalls?: number;
        http2MaxPingsWithoutData?: number;
        http2MinTimeBetweenPingsMs?: number;
        http2MinPingIntervalWithoutDataMs?: number;
        http2MaxPingStrikes?: number;
    };
    channelOptions?: ChannelOptions;
    credentials?: any;
    protoPath?: string | string[];
    package: string | string[];
    protoLoader?: string;
    packageDefinition?: any;
    gracefulShutdown?: boolean;
    onLoadPackageDefinition?: (pkg: any, server: any) => void;
    loader?: {
        keepCase?: boolean;
        alternateCommentMode?: boolean;
        longs?: Function;
        enums?: Function;
        bytes?: Function;
        defaults?: boolean;
        arrays?: boolean;
        objects?: boolean;
        oneofs?: boolean;
        json?: boolean;
        includeDirs?: string[];
    };
});

property grpcClient

protected grpcClient: any;

property status

readonly status: never;

property transportId

transportId: TransportId;

property url

protected readonly url: string;

method addHandler

addHandler: (
    pattern: unknown,
    callback: MessageHandler,
    isEventHandler?: boolean
) => void;

method bindEvents

bindEvents: () => Promise<void>;

method close

close: () => Promise<void>;

method createClient

createClient: () => Promise<any>;

method createPattern

createPattern: (
    service: string,
    methodName: string,
    streaming: GrpcMethodStreamingType
) => string;

Will create a string of a JSON serialized format
Parameter service
name of the service which should be a match to gRPC service definition name
Parameter methodName
name of the method which is coming after rpc keyword
Parameter streaming
GrpcMethodStreamingType parameter which should correspond to stream keyword in gRPC service request part

method createRequestStreamMethod

createRequestStreamMethod: (
    methodHandler: Function,
    isResponseStream: boolean
) => (
    call: GrpcCall,
    callback: (err: unknown, value: unknown) => void
) => Promise<void>;

method createService

createService: (grpcService: any, name: string) => Promise<{}>;

Will create service mapping from gRPC generated Object to handlers defined with or annotations
Parameter grpcService
Parameter name

method createServiceMethod

createServiceMethod: (
    methodHandler: Function,
    protoNativeHandler: any,
    streamType: GrpcMethodStreamingType
) => Function;

Will return async function which will handle gRPC call with Rx streams or as a direct call passthrough
Parameter methodHandler
Parameter protoNativeHandler
Parameter streamType

method createStreamCallMethod

createStreamCallMethod: (
    methodHandler: Function,
    isResponseStream: boolean
) => (
    call: GrpcCall,
    callback: (err: unknown, value: unknown) => void
) => Promise<void>;

method createStreamServiceMethod

createStreamServiceMethod: (methodHandler: Function) => Function;

method createUnaryServiceMethod

createUnaryServiceMethod: (methodHandler: Function) => Function;

method deserialize

deserialize: (obj: any) => any;

method getKeepaliveOptions

getKeepaliveOptions: () => {};

method getMessageHandler

getMessageHandler: (
    serviceName: string,
    methodName: string,
    streaming: GrpcMethodStreamingType,
    grpcMethod: { path?: string }
) => MessageHandler;

method getServiceNames

getServiceNames: (grpcPkg: any) => { name: string; service: any }[];

Will return all of the services along with their fully namespaced names as an array of objects. This method initiates recursive scan of grpcPkg object

method listen

listen: (
    callback: (err?: unknown, ...optionalParams: unknown[]) => void
) => Promise<void>;

method loadProto

loadProto: () => any;

method lookupPackage

lookupPackage: (root: any, packageName: string) => any;

method on

on: <
    EventKey extends string | number | symbol = string | number | symbol,
    EventCallback = any
>(
    event: EventKey,
    callback: EventCallback
) => void;

method start

start: (callback?: () => void) => Promise<void>;

method unwrap

unwrap: <T>() => T;

class ServerKafka

class ServerKafka extends Server<never, KafkaStatus> {}

constructor

constructor(options: {
    postfixId?: string;
    client?: KafkaConfig;
    consumer?: ConsumerConfig;
    run?: Omit<ConsumerRunConfig, 'eachBatch' | 'eachMessage'>;
    subscribe?: Omit<ConsumerSubscribeTopics, 'topics'>;
    producer?: ProducerConfig;
    send?: Omit<ProducerRecord, 'topic' | 'messages'>;
    serializer?: Serializer<any, any>;
    deserializer?: Deserializer<any, any>;
    parser?: KafkaParserConfig;
    producerOnlyMode?: boolean;
});

property brokers

protected brokers: string[] | BrokersFunction;

property client

protected client: Kafka;

property clientId

protected clientId: string;

property consumer

protected consumer: Consumer;

property groupId

protected groupId: string;

property logger

protected logger: Logger;

property options

protected readonly options: {
    postfixId?: string;
    client?: KafkaConfig;
    consumer?: ConsumerConfig;
    run?: Omit<ConsumerRunConfig, 'eachBatch' | 'eachMessage'>;
    subscribe?: Omit<ConsumerSubscribeTopics, 'topics'>;
    producer?: ProducerConfig;
    send?: Omit<ProducerRecord, 'topic' | 'messages'>;
    serializer?: Serializer<any, any>;
    deserializer?: Deserializer<any, any>;
    parser?: KafkaParserConfig;
    producerOnlyMode?: boolean;
};

property parser

protected parser: KafkaParser;

property producer

protected producer: Producer;

property transportId

transportId: TransportId;

method assignCorrelationIdHeader

assignCorrelationIdHeader: (
    correlationId: string,
    outgoingMessage: Message
) => void;

method assignErrorHeader

assignErrorHeader: (
    outgoingResponse: OutgoingResponse,
    outgoingMessage: Message
) => void;

method assignIsDisposedHeader

assignIsDisposedHeader: (
    outgoingResponse: OutgoingResponse,
    outgoingMessage: Message
) => void;

method assignReplyPartition

assignReplyPartition: (
    replyPartition: string | null | undefined,
    outgoingMessage: Message
) => void;

method bindEvents

bindEvents: (consumer: Consumer) => Promise<void>;

method close

close: () => Promise<void>;

method createClient

createClient: <T = any>() => T;

method getMessageHandler

getMessageHandler: () => (payload: EachMessagePayload) => Promise<any>;

method getPublisher

getPublisher: (
    replyTopic: string,
    replyPartition: string,
    correlationId: string,
    context: KafkaContext
) => (data: any) => Promise<RecordMetadata[]>;

method handleEvent

handleEvent: (
    pattern: string,
    packet: ReadPacket,
    context: KafkaContext
) => Promise<any>;

method handleMessage

handleMessage: (payload: EachMessagePayload) => Promise<any>;

method initializeDeserializer

protected initializeDeserializer: (options: KafkaOptions['options']) => void;

method initializeSerializer

protected initializeSerializer: (options: KafkaOptions['options']) => void;

method listen

listen: (
    callback: (err?: unknown, ...optionalParams: unknown[]) => void
) => Promise<void>;

method on

on: <
    EventKey extends string | number | symbol = string | number | symbol,
    EventCallback = any
>(
    event: EventKey,
    callback: EventCallback
) => void;

method registerConsumerEventListeners

protected registerConsumerEventListeners: () => void;

method registerProducerEventListeners

protected registerProducerEventListeners: () => void;

method sendMessage

sendMessage: (
    message: OutgoingResponse,
    replyTopic: string,
    replyPartition: string | undefined | null,
    correlationId: string,
    context: KafkaContext
) => Promise<RecordMetadata[]>;

method start

start: (callback: () => void) => Promise<void>;

method unwrap

unwrap: <T>() => T;

class ServerMqtt

class ServerMqtt extends Server<MqttEvents, MqttStatus> {}

constructor

constructor(
    options: MqttClientOptions & {
        url?: string;
        serializer?: Serializer<any, any>;
        deserializer?: Deserializer<any, any>;
        subscribeOptions?: {
            qos: QoS;
            nl?: boolean;
            rap?: boolean;
            rh?: number;
        };
        userProperties?: Record<string, string | string[]>;
    }
);

property mqttClient

protected mqttClient: any;

property pendingEventListeners

protected pendingEventListeners: {
    event: keyof MqttEvents;
    callback: MqttEvents[keyof MqttEvents];
}[];

property transportId

transportId: TransportId;

property url

protected readonly url: string;

method bindEvents

bindEvents: (mqttClient: MqttClient) => void;

method close

close: () => void;

method createMqttClient

createMqttClient: () => MqttClient;

method getHandlerByPattern

getHandlerByPattern: (pattern: string) => MessageHandler | null;

method getMessageHandler

getMessageHandler: (
    pub: MqttClient
) => (
    channel: string,
    buffer: Buffer,
    originalPacket?: Record<string, any>
) => Promise<any>;

method getPublisher

getPublisher: (client: MqttClient, context: MqttContext, id: string) => any;

method getReplyPattern

getReplyPattern: (pattern: string) => string;

method getRequestPattern

getRequestPattern: (pattern: string) => string;

method handleMessage

handleMessage: (
    channel: string,
    buffer: Buffer,
    pub: MqttClient,
    originalPacket?: Record<string, any>
) => Promise<any>;

method initializeSerializer

protected initializeSerializer: (options: MqttOptions['options']) => void;

method listen

listen: (
    callback: (err?: unknown, ...optionalParams: unknown[]) => void
) => Promise<void>;

method matchMqttPattern

matchMqttPattern: (pattern: string, topic: string) => boolean;

method on

on: <
    EventKey extends keyof MqttEvents = keyof MqttEvents,
    EventCallback extends MqttEvents[EventKey] = MqttEvents[EventKey]
>(
    event: EventKey,
    callback: EventCallback
) => void;

method parseMessage

parseMessage: (content: any) => ReadPacket & PacketId;

method registerCloseListener

registerCloseListener: (client: MqttClient) => void;

method registerConnectListener

registerConnectListener: (client: MqttClient) => void;

method registerDisconnectListener

registerDisconnectListener: (client: MqttClient) => void;

method registerErrorListener

registerErrorListener: (client: MqttClient) => void;

method registerReconnectListener

registerReconnectListener: (client: MqttClient) => void;

method removeHandlerKeySharedPrefix

removeHandlerKeySharedPrefix: (handlerKey: string) => string;

method start

start: (callback: (err?: unknown, ...optionalParams: unknown[]) => void) => void;

method unwrap

unwrap: <T>() => T;

class ServerNats

class ServerNats<
    E extends NatsEvents = NatsEvents,
    S extends NatsStatus = NatsStatus
> extends Server<E, S> {}

constructor

constructor(options: {
    [key: string]: any;
    headers?: Record<string, string>;
    authenticator?: any;
    debug?: boolean;
    ignoreClusterUpdates?: boolean;
    inboxPrefix?: string;
    encoding?: string;
    name?: string;
    user?: string;
    pass?: string;
    maxPingOut?: number;
    maxReconnectAttempts?: number;
    reconnectTimeWait?: number;
    reconnectJitter?: number;
    reconnectJitterTLS?: number;
    reconnectDelayHandler?: any;
    servers?: string | string[];
    nkey?: any;
    reconnect?: boolean;
    pedantic?: boolean;
    tls?: any;
    queue?: string;
    serializer?: Serializer<any, any>;
    deserializer?: Deserializer<any, any>;
    userJWT?: string;
    nonceSigner?: any;
    userCreds?: any;
    useOldRequestStyle?: boolean;
    pingInterval?: number;
    preserveBuffers?: boolean;
    waitOnFirstConnect?: boolean;
    verbose?: boolean;
    noEcho?: boolean;
    noRandomize?: boolean;
    timeout?: number;
    token?: string;
    yieldTime?: number;
    tokenHandler?: any;
    gracefulShutdown?: boolean;
    gracePeriod?: number;
});

property statusEventEmitter

protected statusEventEmitter: EventEmitter<{
    disconnect: [data?: string | number | undefined];
    reconnect: [data?: string | number | undefined];
    update: [
        data?:
            | string
            | number
            | import('../events/nats.events').ServersChangedEvent
            | undefined
    ];
}>;

property transportId

transportId: TransportId;

method bindEvents

bindEvents: (client: Client) => void;

method close

close: () => Promise<void>;

method createNatsClient

createNatsClient: () => Promise<Client>;

method getMessageHandler

getMessageHandler: (channel: string) => Function;

method getPublisher

getPublisher: (
    natsMsg: NatsMsg,
    id: string,
    ctx: NatsContext
) => (response: any) => any;

method handleMessage

handleMessage: (channel: string, natsMsg: NatsMsg) => Promise<any>;

method handleStatusUpdates

handleStatusUpdates: (client: Client) => Promise<void>;

method initializeDeserializer

protected initializeDeserializer: (options: NatsOptions['options']) => void;

method initializeSerializer

protected initializeSerializer: (options: NatsOptions['options']) => void;

method listen

listen: (
    callback: (err?: unknown, ...optionalParams: unknown[]) => void
) => Promise<void>;

method on

on: <
    EventKey extends keyof E = keyof E,
    EventCallback extends E[EventKey] = E[EventKey]
>(
    event: EventKey,
    callback: EventCallback
) => void;

method start

start: (callback: (err?: unknown, ...optionalParams: unknown[]) => void) => void;

method unwrap

unwrap: <T>() => T;

class ServerRedis

class ServerRedis extends Server<RedisEvents, RedisStatus> {}

constructor

constructor(
    options: {
        host?: string;
        port?: number;
        retryAttempts?: number;
        retryDelay?: number;
        wildcards?: boolean;
        serializer?: Serializer<any, any>;
        deserializer?: Deserializer<any, any>;
    } & IORedisOptions
);

property isManuallyClosed

protected isManuallyClosed: boolean;

property options

protected readonly options: {
    host?: string;
    port?: number;
    retryAttempts?: number;
    retryDelay?: number;
    wildcards?: boolean;
    serializer?: Serializer<any, any>;
    deserializer?: Deserializer<any, any>;
} & IORedisOptions;

property pendingEventListeners

protected pendingEventListeners: {
    event: keyof RedisEvents;
    callback: RedisEvents[keyof RedisEvents];
}[];

property pubClient

protected pubClient: any;

property subClient

protected subClient: any;

property transportId

transportId: TransportId;

property wasInitialConnectionSuccessful

protected wasInitialConnectionSuccessful: boolean;

method bindEvents

bindEvents: (subClient: Redis, pubClient: Redis) => void;

method close

close: () => Promise<void>;

method createRedisClient

createRedisClient: () => Redis;

method createRetryStrategy

createRetryStrategy: (times: number) => undefined | number | void;

method getClientOptions

getClientOptions: () => Partial<RedisOptions['options']>;

method getMessageHandler

getMessageHandler: (
    pub: Redis
) => (channel: string, pattern: string, buffer: string) => Promise<any>;

method getPublisher

getPublisher: (
    pub: Redis,
    pattern: any,
    id: string,
    ctx: RedisContext
) => (response: any) => any;

method getReplyPattern

getReplyPattern: (pattern: string) => string;

method getRequestPattern

getRequestPattern: (pattern: string) => string;

method handleMessage

handleMessage: (
    channel: string,
    buffer: string,
    pub: Redis,
    pattern: string
) => Promise<any>;

method listen

listen: (
    callback: (err?: unknown, ...optionalParams: unknown[]) => void
) => void;

method on

on: <
    EventKey extends keyof RedisEvents = keyof RedisEvents,
    EventCallback extends RedisEvents[EventKey] = RedisEvents[EventKey]
>(
    event: EventKey,
    callback: EventCallback
) => void;

method parseMessage

parseMessage: (content: any) => Record<string, any>;

method registerEndListener

registerEndListener: (client: {
    on: (event: string, fn: () => void) => void;
}) => void;

method registerErrorListener

registerErrorListener: (client: any) => void;

method registerReadyListener

registerReadyListener: (client: {
    on: (event: string, fn: () => void) => void;
}) => void;

method registerReconnectListener

registerReconnectListener: (client: {
    on: (event: string, fn: () => void) => void;
}) => void;

method start

start: (callback?: () => void) => void;

method unwrap

unwrap: <T>() => T;

class ServerRMQ

class ServerRMQ extends Server<RmqEvents, RmqStatus> {}

constructor

constructor(options: {
    urls?: string[] | RmqUrl[];
    queue?: string;
    prefetchCount?: number;
    isGlobalPrefetchCount?: boolean;
    queueOptions?: AmqplibQueueOptions;
    socketOptions?: AmqpConnectionManagerSocketOptions;
    noAck?: boolean;
    consumerTag?: string;
    serializer?: Serializer<any, any>;
    deserializer?: Deserializer<any, any>;
    replyQueue?: string;
    persistent?: boolean;
    headers?: Record<string, string>;
    noAssert?: boolean;
    exchange?: string;
    exchangeType?: 'topic' | 'direct' | 'fanout' | 'headers' | (string & {});
    exchangeArguments?: Record<string, string>;
    routingKey?: string;
    wildcards?: boolean;
    maxConnectionAttempts?: number;
});

property channel

protected channel: any;

property connectionAttempts

protected connectionAttempts: number;

property noAck

protected readonly noAck: boolean;

property options

protected readonly options: {
    urls?: string[] | RmqUrl[];
    queue?: string;
    prefetchCount?: number;
    isGlobalPrefetchCount?: boolean;
    queueOptions?: AmqplibQueueOptions;
    socketOptions?: AmqpConnectionManagerSocketOptions;
    noAck?: boolean;
    consumerTag?: string;
    serializer?: Serializer<any, any>;
    deserializer?: Deserializer<any, any>;
    replyQueue?: string;
    persistent?: boolean;
    headers?: Record<string, string>;
    noAssert?: boolean;
    exchange?: string;
    exchangeType?: 'topic' | 'direct' | 'fanout' | 'headers' | (string & {});
    exchangeArguments?: Record<string, string>;
    routingKey?: string;
    wildcards?: boolean;
    maxConnectionAttempts?: number;
};

property pendingEventListeners

protected pendingEventListeners: {
    event: keyof RmqEvents;
    callback: RmqEvents[keyof RmqEvents];
}[];

property queue

protected readonly queue: string;

property queueOptions

protected readonly queueOptions: any;

property server

protected server: any;

property transportId

transportId: TransportId;

property urls

protected readonly urls: string[] | RmqUrl[];

property wildcardHandlers

protected readonly wildcardHandlers: Map<string, MessageHandler<any, any, any>>;

method close

close: () => Promise<void>;

method createClient

createClient: <T = any>() => T;

method getHandlerByPattern

getHandlerByPattern: (pattern: string) => MessageHandler | null;

method handleEvent

handleEvent: (
    pattern: string,
    packet: ReadPacket,
    context: RmqContext
) => Promise<any>;

method handleMessage

handleMessage: (message: Record<string, any>, channel: any) => Promise<void>;

method initializeSerializer

protected initializeSerializer: (options: RmqOptions['options']) => void;

method initializeWildcardHandlersIfExist

protected initializeWildcardHandlersIfExist: () => void;

method listen

listen: (
    callback: (err?: unknown, ...optionalParams: unknown[]) => void
) => Promise<void>;

method on

on: <
    EventKey extends keyof RmqEvents = keyof RmqEvents,
    EventCallback extends RmqEvents[EventKey] = RmqEvents[EventKey]
>(
    event: EventKey,
    callback: EventCallback
) => void;

method sendMessage

sendMessage: <T = any>(
    message: T,
    replyTo: any,
    correlationId: string,
    context: RmqContext
) => void;

method setupChannel

setupChannel: (channel: Channel, callback: Function) => Promise<void>;

method start

start: (
    callback?: (err?: unknown, ...optionalParams: unknown[]) => void
) => Promise<void>;

method unwrap

unwrap: <T>() => T;

class ServerTCP

class ServerTCP extends Server<TcpEvents, TcpStatus> {}

constructor

constructor(options: {
    host?: string;
    port?: number;
    retryAttempts?: number;
    retryDelay?: number;
    serializer?: Serializer<any, any>;
    tlsOptions?: TlsOptions;
    deserializer?: Deserializer<any, any>;
    socketClass?: Type<TcpSocket>;
    maxBufferSize?: number;
});

property host

protected readonly host: string;

property isManuallyTerminated

protected isManuallyTerminated: boolean;

property maxBufferSize

protected readonly maxBufferSize?: number;

property pendingEventListeners

protected pendingEventListeners: {
    event: keyof TcpEvents;
    callback: TcpEvents[keyof TcpEvents];
}[];

property port

protected readonly port: number;

property retryAttemptsCount

protected retryAttemptsCount: number;

property server

protected server: NetSocket;

property socketClass

protected readonly socketClass: Type<TcpSocket>;

property tlsOptions

protected tlsOptions?: TlsOptions;

property transportId

transportId: TransportId;

method bindHandler

bindHandler: (socket: Socket) => void;

method close

close: () => void;

method getSocketInstance

protected getSocketInstance: (socket: Socket) => TcpSocket;

method handleClose

handleClose: () => undefined | number | NodeJS.Timer;

method handleMessage

handleMessage: (socket: TcpSocket, rawMessage: unknown) => Promise<any>;

method init

protected init: () => void;

method listen

listen: (
    callback: (err?: unknown, ...optionalParams: unknown[]) => void
) => void;

method on

on: <
    EventKey extends keyof TcpEvents = keyof TcpEvents,
    EventCallback extends TcpEvents[EventKey] = TcpEvents[EventKey]
>(
    event: EventKey,
    callback: EventCallback
) => void;

method registerCloseListener

protected registerCloseListener: (socket: net.Server) => void;

method registerErrorListener

protected registerErrorListener: (socket: net.Server) => void;

method registerListeningListener

protected registerListeningListener: (socket: net.Server) => void;

method unwrap

unwrap: <T>() => T;

class TcpContext

class TcpContext extends BaseRpcContext<TcpContextArgs> {}

constructor

constructor(args: TcpContextArgs);

method getPattern

getPattern: () => string;

Returns the name of the pattern.

method getSocketRef

getSocketRef: () => TcpSocket;

Returns the underlying JSON socket.

class TcpSocket

abstract class TcpSocket {}

constructor

constructor(socket: Socket);

property netSocket

readonly netSocket: Socket;

property socket

readonly socket: Socket;

method connect

connect: (port: number, host: string) => this;

method emitMessage

protected emitMessage: (data: string) => void;

method end

end: () => this;

method handleData

protected abstract handleData: (data: Buffer | string) => any;

method handleSend

protected abstract handleSend: (
    message: any,
    callback?: (err?: any) => void
) => any;

method on

on: (event: string, callback: (err?: any) => void) => this;

method once

once: (event: string, callback: (err?: any) => void) => this;

method sendMessage

sendMessage: (message: any, callback?: (err?: any) => void) => void;

Interfaces

interface ClientGrpc

interface ClientGrpc {}

method getClientByServiceName

getClientByServiceName: <T = any>(name: string) => T;

Returns an instance of the given gRPC client.
Parameter name
Service name
Returns
gRPC client

method getService

getService: <T extends object>(name: string) => T;

Returns an instance of the given gRPC service.
Parameter name
Service name
Returns
gRPC service

interface ClientKafkaProxy

interface ClientKafkaProxy extends Omit<ClientProxy<never, KafkaStatus>, 'on'> {}

property consumer

consumer: Consumer | null;

Reference to the Kafka consumer instance.

property producer

producer: Producer | null;

Reference to the Kafka producer instance.

method commitOffsets

commitOffsets: (
    topicPartitions: TopicPartitionOffsetAndMetadata[]
) => Promise<void>;

Commits the given offsets.
Parameter topicPartitions
Array of topic partitions with their offsets and metadata

method subscribeToResponseOf

subscribeToResponseOf: (pattern: unknown) => void;

Subscribes to messages that match the pattern. Required for message-driven communication style between microservices. You can't use send without subscribing to the message pattern first.
Parameter pattern
Pattern to subscribe to

interface ClientsModuleOptionsFactory

interface ClientsModuleOptionsFactory {}

method createClientOptions

createClientOptions: () => Promise<ClientProvider> | ClientProvider;

interface ClientsProviderAsyncOptions

interface ClientsProviderAsyncOptions extends Pick<ModuleMetadata, 'imports'> {}

property extraProviders

extraProviders?: Provider[];

property inject

inject?: any[];

property name

name: string | symbol;

property useClass

useClass?: Type<ClientsModuleOptionsFactory>;

property useExisting

useExisting?: Type<ClientsModuleOptionsFactory>;

property useFactory

useFactory?: (...args: any[]) => Promise<ClientProvider> | ClientProvider;

interface CustomClientOptions

interface CustomClientOptions {}

property customClass

customClass: Type<ClientProxy>;

property options

options?: Record<string, any>;

interface CustomStrategy

interface CustomStrategy {}

property options

options?: Record<string, any>;

property strategy

strategy: CustomTransportStrategy;

interface CustomTransportStrategy

interface CustomTransportStrategy {}

property transportId

transportId?: TransportId;

Unique transport identifier.

method close

close: () => any;

Method called when the transport is being terminated.

method listen

listen: (callback: (...optionalParams: unknown[]) => any) => any;

Method called when the transport is being initialized.
Parameter callback
Function to be called upon initialization

interface Deserializer

interface Deserializer<TInput = any, TOutput = any> {}

method deserialize

deserialize: (
    value: TInput,
    options?: Record<string, any>
) => TOutput | Promise<TOutput>;

interface GrpcOptions

interface GrpcOptions {}

property options

options: {
    url?: string;
    maxSendMessageLength?: number;
    maxReceiveMessageLength?: number;
    maxMetadataSize?: number;
    keepalive?: {
        keepaliveTimeMs?: number;
        keepaliveTimeoutMs?: number;
        keepalivePermitWithoutCalls?: number;
        http2MaxPingsWithoutData?: number;
        http2MinTimeBetweenPingsMs?: number;
        http2MinPingIntervalWithoutDataMs?: number;
        http2MaxPingStrikes?: number;
    };
    channelOptions?: ChannelOptions;
    credentials?: any;
    protoPath?: string | string[];
    package: string | string[];
    protoLoader?: string;
    packageDefinition?: any;
    gracefulShutdown?: boolean;
    onLoadPackageDefinition?: (pkg: any, server: any) => void;
    loader?: {
        keepCase?: boolean;
        alternateCommentMode?: boolean;
        longs?: Function;
        enums?: Function;
        bytes?: Function;
        defaults?: boolean;
        arrays?: boolean;
        objects?: boolean;
        oneofs?: boolean;
        json?: boolean;
        includeDirs?: string[];
    };
};

property transport

transport?: Transport.GRPC;

interface JsonSocketOptions

interface JsonSocketOptions {}

property maxBufferSize

maxBufferSize?: number;

interface KafkaOptions

interface KafkaOptions {}

property options

options?: {
    /**
     * Defaults to `"-server"` on server side and `"-client"` on client side.
     */
    postfixId?: string;
    client?: KafkaConfig;
    consumer?: ConsumerConfig;
    /**
     * Options passed to KafkaJS consumer.run().
     * Note: `partitionsConsumedConcurrently` (KafkaJS parameter) controls
     * concurrent processing at the partition level (not topic level).
     */
    run?: Omit<ConsumerRunConfig, 'eachBatch' | 'eachMessage'>;
    subscribe?: Omit<ConsumerSubscribeTopics, 'topics'>;
    producer?: ProducerConfig;
    send?: Omit<ProducerRecord, 'topic' | 'messages'>;
    serializer?: Serializer;
    deserializer?: Deserializer;
    parser?: KafkaParserConfig;
    producerOnlyMode?: boolean;
};

property transport

transport?: Transport.KAFKA;

interface KafkaParserConfig

interface KafkaParserConfig {}

property keepBinary

keepBinary?: boolean;

interface MessageHandler

interface MessageHandler<TInput = any, TContext = any, TResult = any> {}

property extras

extras?: Record<string, any>;

property isEventHandler

isEventHandler?: boolean;

property next

next?: (
    data: TInput,
    ctx?: TContext
) => Promise<Observable<TResult>> | Promise<TResult>;

call signature

(data: TInput, ctx?: TContext): Promise<Observable<TResult>> | Promise<TResult>;

interface MqttOptions

interface MqttOptions {}

property options

options?: MqttClientOptions & {
    url?: string;
    serializer?: Serializer;
    deserializer?: Deserializer;
    subscribeOptions?: {
        /**
         * The QoS
         */
        qos: QoS;
        nl?: boolean;
        rap?: boolean;
        rh?: number;
    };
    userProperties?: Record<string, string | string[]>;
};

property transport

transport?: Transport.MQTT;

interface MqttRecordOptions

interface MqttRecordOptions {}

property dup

dup?: boolean;

Whether or not mark a message as duplicate

property properties

properties?: {
    payloadFormatIndicator?: boolean;
    messageExpiryInterval?: number;
    topicAlias?: number;
    responseTopic?: string;
    correlationData?: Buffer;
    userProperties?: Record<string, string | string[]>;
    subscriptionIdentifier?: number;
    contentType?: string;
};

property qos

qos?: 0 | 1 | 2;

The QoS

property retain

retain?: boolean;

The retain flag

interface MsObjectPattern

interface MsObjectPattern {}

index signature

[key: string]: MsFundamentalPattern | MsObjectPattern;

interface NatsOptions

interface NatsOptions {}

property options

options?: {
    headers?: Record<string, string>;
    authenticator?: any;
    debug?: boolean;
    ignoreClusterUpdates?: boolean;
    inboxPrefix?: string;
    encoding?: string;
    name?: string;
    user?: string;
    pass?: string;
    maxPingOut?: number;
    maxReconnectAttempts?: number;
    reconnectTimeWait?: number;
    reconnectJitter?: number;
    reconnectJitterTLS?: number;
    reconnectDelayHandler?: any;
    servers?: string[] | string;
    nkey?: any;
    reconnect?: boolean;
    pedantic?: boolean;
    tls?: any;
    queue?: string;
    serializer?: Serializer;
    deserializer?: Deserializer;
    userJWT?: string;
    nonceSigner?: any;
    userCreds?: any;
    useOldRequestStyle?: boolean;
    pingInterval?: number;
    preserveBuffers?: boolean;
    waitOnFirstConnect?: boolean;
    verbose?: boolean;
    noEcho?: boolean;
    noRandomize?: boolean;
    timeout?: number;
    token?: string;
    yieldTime?: number;
    tokenHandler?: any;
    gracefulShutdown?: boolean;
    gracePeriod?: number;
    [key: string]: any;
};

property transport

transport?: Transport.NATS;

interface PacketId

interface PacketId {}

property id

id: string;

interface ReadPacket

interface ReadPacket<T = any> {}

property data

data: T;

property pattern

pattern: any;

interface RedisOptions

interface RedisOptions {}

property options

options?: {
    host?: string;
    port?: number;
    retryAttempts?: number;
    retryDelay?: number;
    /**
     * Use `psubscribe`/`pmessage` to enable wildcards in the patterns
     */
    wildcards?: boolean;
    serializer?: Serializer;
    deserializer?: Deserializer;
} & IORedisOptions;

property transport

transport?: Transport.REDIS;

interface RequestContext

interface RequestContext<TData = any, TContext extends BaseRpcContext = any> {}

property context

context?: TContext;

property data

data: TData;

property pattern

pattern: string | Record<string, any>;

method getContext

getContext: () => TContext;

method getData

getData: () => TData;

method getPattern

getPattern: () => string | Record<string, any>;

interface RmqOptions

interface RmqOptions {}

property options

options?: {
    /**
     * An array of connection URLs to try in order.
     */
    urls?: string[] | RmqUrl[];
    /**
     * The name of the queue.
     */
    queue?: string;
    /**
     * A prefetch count for this channel. The count given is the maximum number of messages sent over the channel that can be awaiting acknowledgement;
     * once there are count messages outstanding, the server will not send more messages on this channel until one or more have been acknowledged.
     */
    prefetchCount?: number;
    /**
     * Sets the per-channel behavior for prefetching messages.
     */
    isGlobalPrefetchCount?: boolean;
    /**
     * Amqplib queue options.
     * @see https://amqp-node.github.io/amqplib/channel_api.html#channel_assertQueue
     */
    queueOptions?: AmqplibQueueOptions;
    /**
     * AMQP Connection Manager socket options.
     */
    socketOptions?: AmqpConnectionManagerSocketOptions;
    /**
     * If true, the broker won’t expect an acknowledgement of messages delivered to this consumer; i.e., it will dequeue messages as soon as they’ve been sent down the wire.
     * @default false
     */
    noAck?: boolean;
    /**
     * A name which the server will use to distinguish message deliveries for the consumer; mustn’t be already in use on the channel. It’s usually easier to omit this, in which case the server will create a random name and supply it in the reply.
     */
    consumerTag?: string;
    /**
     * A serializer for the message payload.
     */
    serializer?: Serializer;
    /**
     * A deserializer for the message payload.
     */
    deserializer?: Deserializer;
    /**
     * A reply queue for the producer.
     * @default 'amq.rabbitmq.reply-to'
     */
    replyQueue?: string;
    /**
     * If truthy, the message will survive broker restarts provided it’s in a queue that also survives restarts.
     */
    persistent?: boolean;
    /**
     * Additional headers to be sent with every message.
     * Applies only to the producer configuration.
     */
    headers?: Record<string, string>;
    /**
     * When false, a queue will not be asserted before consuming.
     * @default false
     */
    noAssert?: boolean;
    /**
     * Name for the exchange. Defaults to the queue name when "wildcards" is set to true.
     * @default ''
     */
    exchange?: string;
    /**
     * Type of the exchange.
     * Accepts the AMQP standard types ('direct', 'fanout', 'topic', 'headers') or any custom exchange type name provided as a string literal.
     * @default 'topic'
     */
    exchangeType?: 'direct' | 'fanout' | 'topic' | 'headers' | (string & {});
    /**
     * Exchange arguments
     */
    exchangeArguments?: Record<string, string>;
    /**
     * Additional routing key for the topic exchange.
     */
    routingKey?: string;
    /**
     * Set to true only if you want to use Topic Exchange for routing messages to queues.
     * Enabling this will allow you to use wildcards (*, #) as message and event patterns.
     * @see https://www.rabbitmq.com/tutorials/tutorial-five-python#topic-exchange
     * @default false
     */
    wildcards?: boolean;
    /**
     * Maximum number of connection attempts.
     * Applies only to the consumer configuration.
     * -1 === infinite
     * @default -1
     */
    maxConnectionAttempts?: number;
};

property transport

transport?: Transport.RMQ;

interface RmqRecordOptions

interface RmqRecordOptions {}

property appId

appId?: string;

property BCC

BCC?: string | string[];

property CC

CC?: string | string[];

property contentEncoding

contentEncoding?: string;

property contentType

contentType?: string;

property deliveryMode

deliveryMode?: boolean | number;

property expiration

expiration?: string | number;

property headers

headers?: Record<string, string>;

property mandatory

mandatory?: boolean;

property messageId

messageId?: string;

property persistent

persistent?: boolean;

property priority

priority?: number;

property timestamp

timestamp?: number;

property type

type?: string;

property userId

userId?: string;

interface Serializer

interface Serializer<TInput = any, TOutput = any> {}

method serialize

serialize: (value: TInput, options?: Record<string, any>) => TOutput;

interface TcpClientOptions

interface TcpClientOptions {}

property options

options?: {
    host?: string;
    port?: number;
    serializer?: Serializer;
    deserializer?: Deserializer;
    tlsOptions?: ConnectionOptions;
    socketClass?: Type<TcpSocket>;
    /**
     * Maximum buffer size in characters (default: 128MB in characters, i.e., (512 * 1024 * 1024) / 4).
     * This limit prevents memory exhaustion when receiving large TCP messages.
     */
    maxBufferSize?: number;
};

property transport

transport: Transport.TCP;

interface TcpOptions

interface TcpOptions {}

property options

options?: {
    host?: string;
    port?: number;
    retryAttempts?: number;
    retryDelay?: number;
    serializer?: Serializer;
    tlsOptions?: TlsOptions;
    deserializer?: Deserializer;
    socketClass?: Type<TcpSocket>;
    /**
     * Maximum buffer size in characters (default: 128MB in characters, i.e., (512 * 1024 * 1024) / 4).
     * This limit prevents memory exhaustion when receiving large TCP messages.
     */
    maxBufferSize?: number;
};

property transport

transport?: Transport.TCP;

interface WritePacket

interface WritePacket<T = any> {}

property err

err?: any;

property isDisposed

isDisposed?: boolean;

property response

response?: T;

property status

status?: string;

Enums

enum GrpcMethodStreamingType

enum GrpcMethodStreamingType {
    NO_STREAMING = 'no_stream',
    RX_STREAMING = 'rx_stream',
    PT_STREAMING = 'pt_stream',
}

member NO_STREAMING

NO_STREAMING = 'no_stream'

member PT_STREAMING

PT_STREAMING = 'pt_stream'

member RX_STREAMING

RX_STREAMING = 'rx_stream'

enum KafkaHeaders

enum KafkaHeaders {
    ACKNOWLEDGMENT = 'kafka_acknowledgment',
    BATCH_CONVERTED_HEADERS = 'kafka_batchConvertedHeaders',
    CONSUMER = 'kafka_consumer',
    CORRELATION_ID = 'kafka_correlationId',
    DELIVERY_ATTEMPT = 'kafka_deliveryAttempt',
    DLT_EXCEPTION_FQCN = 'kafka_dlt-exception-fqcn',
    DLT_EXCEPTION_MESSAGE = 'kafka_dlt-exception-message',
    DLT_EXCEPTION_STACKTRACE = 'kafka_dlt-exception-stacktrace',
    DLT_ORIGINAL_OFFSET = 'kafka_dlt-original-offset',
    DLT_ORIGINAL_PARTITION = 'kafka_dlt-original-partition',
    DLT_ORIGINAL_TIMESTAMP = 'kafka_dlt-original-timestamp',
    DLT_ORIGINAL_TIMESTAMP_TYPE = 'kafka_dlt-original-timestamp-type',
    DLT_ORIGINAL_TOPIC = 'kafka_dlt-original-topic',
    GROUP_ID = 'kafka_groupId',
    MESSAGE_KEY = 'kafka_messageKey',
    NATIVE_HEADERS = 'kafka_nativeHeaders',
    OFFSET = 'kafka_offset',
    PARTITION_ID = 'kafka_partitionId',
    PREFIX = 'kafka_',
    RAW_DATA = 'kafka_data',
    RECEIVED = 'kafka_received',
    RECEIVED_MESSAGE_KEY = 'kafka_receivedMessageKey',
    RECEIVED_PARTITION_ID = 'kafka_receivedPartitionId',
    RECEIVED_TIMESTAMP = 'kafka_receivedTimestamp',
    RECEIVED_TOPIC = 'kafka_receivedTopic',
    RECORD_METADATA = 'kafka_recordMetadata',
    REPLY_PARTITION = 'kafka_replyPartition',
    REPLY_TOPIC = 'kafka_replyTopic',
    TIMESTAMP = 'kafka_timestamp',
    TIMESTAMP_TYPE = 'kafka_timestampType',
    TOPIC = 'kafka_topic',
    NEST_ERR = 'kafka_nest-err',
    NEST_IS_DISPOSED = 'kafka_nest-is-disposed',
}

See Also
- https://docs.spring.io/spring-kafka/api/org/springframework/kafka/support/KafkaHeaders.html

member ACKNOWLEDGMENT

ACKNOWLEDGMENT = 'kafka_acknowledgment'

member BATCH_CONVERTED_HEADERS

BATCH_CONVERTED_HEADERS = 'kafka_batchConvertedHeaders'

member CONSUMER

CONSUMER = 'kafka_consumer'

member CORRELATION_ID

CORRELATION_ID = 'kafka_correlationId'

member DELIVERY_ATTEMPT

DELIVERY_ATTEMPT = 'kafka_deliveryAttempt'

member DLT_EXCEPTION_FQCN

DLT_EXCEPTION_FQCN = 'kafka_dlt-exception-fqcn'

member DLT_EXCEPTION_MESSAGE

DLT_EXCEPTION_MESSAGE = 'kafka_dlt-exception-message'

member DLT_EXCEPTION_STACKTRACE

DLT_EXCEPTION_STACKTRACE = 'kafka_dlt-exception-stacktrace'

member DLT_ORIGINAL_OFFSET

DLT_ORIGINAL_OFFSET = 'kafka_dlt-original-offset'

member DLT_ORIGINAL_PARTITION

DLT_ORIGINAL_PARTITION = 'kafka_dlt-original-partition'

member DLT_ORIGINAL_TIMESTAMP

DLT_ORIGINAL_TIMESTAMP = 'kafka_dlt-original-timestamp'

member DLT_ORIGINAL_TIMESTAMP_TYPE

DLT_ORIGINAL_TIMESTAMP_TYPE = 'kafka_dlt-original-timestamp-type'

member DLT_ORIGINAL_TOPIC

DLT_ORIGINAL_TOPIC = 'kafka_dlt-original-topic'

member GROUP_ID

GROUP_ID = 'kafka_groupId'

member MESSAGE_KEY

MESSAGE_KEY = 'kafka_messageKey'

member NATIVE_HEADERS

NATIVE_HEADERS = 'kafka_nativeHeaders'

member NEST_ERR

NEST_ERR = 'kafka_nest-err'

member NEST_IS_DISPOSED

NEST_IS_DISPOSED = 'kafka_nest-is-disposed'

member OFFSET

OFFSET = 'kafka_offset'

member PARTITION_ID

PARTITION_ID = 'kafka_partitionId'

member PREFIX

PREFIX = 'kafka_'

member RAW_DATA

RAW_DATA = 'kafka_data'

member RECEIVED

RECEIVED = 'kafka_received'

member RECEIVED_MESSAGE_KEY

RECEIVED_MESSAGE_KEY = 'kafka_receivedMessageKey'

member RECEIVED_PARTITION_ID

RECEIVED_PARTITION_ID = 'kafka_receivedPartitionId'

member RECEIVED_TIMESTAMP

RECEIVED_TIMESTAMP = 'kafka_receivedTimestamp'

member RECEIVED_TOPIC

RECEIVED_TOPIC = 'kafka_receivedTopic'

member RECORD_METADATA

RECORD_METADATA = 'kafka_recordMetadata'

member REPLY_PARTITION

REPLY_PARTITION = 'kafka_replyPartition'

member REPLY_TOPIC

REPLY_TOPIC = 'kafka_replyTopic'

member TIMESTAMP

TIMESTAMP = 'kafka_timestamp'

member TIMESTAMP_TYPE

TIMESTAMP_TYPE = 'kafka_timestampType'

member TOPIC

TOPIC = 'kafka_topic'

enum KafkaStatus

const enum KafkaStatus {
    DISCONNECTED = 'disconnected',
    CONNECTED = 'connected',
    CRASHED = 'crashed',
    STOPPED = 'stopped',
    REBALANCING = 'rebalancing',
}

member CONNECTED

CONNECTED = 'connected'

member CRASHED

CRASHED = 'crashed'

member DISCONNECTED

DISCONNECTED = 'disconnected'

member REBALANCING

REBALANCING = 'rebalancing'

member STOPPED

STOPPED = 'stopped'

enum MqttStatus

const enum MqttStatus {
    DISCONNECTED = 'disconnected',
    RECONNECTING = 'reconnecting',
    CONNECTED = 'connected',
    CLOSED = 'closed',
}

member CLOSED

CLOSED = 'closed'

member CONNECTED

CONNECTED = 'connected'

member DISCONNECTED

DISCONNECTED = 'disconnected'

member RECONNECTING

RECONNECTING = 'reconnecting'

enum NatsStatus

const enum NatsStatus {
    DISCONNECTED = 'disconnected',
    RECONNECTING = 'reconnecting',
    CONNECTED = 'connected',
}

member CONNECTED

CONNECTED = 'connected'

member DISCONNECTED

DISCONNECTED = 'disconnected'

member RECONNECTING

RECONNECTING = 'reconnecting'

enum RedisStatus

const enum RedisStatus {
    DISCONNECTED = 'disconnected',
    RECONNECTING = 'reconnecting',
    CONNECTED = 'connected',
}

member CONNECTED

CONNECTED = 'connected'

member DISCONNECTED

DISCONNECTED = 'disconnected'

member RECONNECTING

RECONNECTING = 'reconnecting'

enum RmqStatus

const enum RmqStatus {
    DISCONNECTED = 'disconnected',
    CONNECTED = 'connected',
    BLOCKED = 'blocked',
    UNBLOCKED = 'unblocked',
}

member BLOCKED

BLOCKED = 'blocked'

member CONNECTED

CONNECTED = 'connected'

member DISCONNECTED

DISCONNECTED = 'disconnected'

member UNBLOCKED

UNBLOCKED = 'unblocked'

enum TcpStatus

const enum TcpStatus {
    DISCONNECTED = 'disconnected',
    CONNECTED = 'connected',
}

member CONNECTED

CONNECTED = 'connected'

member DISCONNECTED

DISCONNECTED = 'disconnected'

enum Transport

enum Transport {
    TCP = 0,
    REDIS = 1,
    NATS = 2,
    MQTT = 3,
    GRPC = 4,
    RMQ = 5,
    KAFKA = 6,
}

member GRPC

GRPC = 4

member KAFKA

KAFKA = 6

member MQTT

MQTT = 3

member NATS

NATS = 2

member REDIS

REDIS = 1

member RMQ

RMQ = 5

member TCP

TCP = 0

Type Aliases

type AsyncMicroserviceOptions

type AsyncMicroserviceOptions = {
    inject: InjectionToken[];
    useFactory: (...args: any[]) => MicroserviceOptions;
};

type AsyncOptions

type AsyncOptions<T extends object> = {
    inject: InjectionToken[];
    useFactory: (...args: any[]) => T;
};

type ClientOptions

type ClientOptions =
    | RedisOptions
    | NatsOptions
    | MqttOptions
    | GrpcOptions
    | KafkaOptions
    | TcpClientOptions
    | RmqOptions;

type ClientProvider

type ClientProvider = ClientOptions | CustomClientOptions;

type ClientProviderOptions

type ClientProviderOptions = ClientProvider & {
    name: string | symbol;
};

type ClientsModuleAsyncOptions

type ClientsModuleAsyncOptions =
    | Array<ClientsProviderAsyncOptions>
    | {
          clients: Array<ClientsProviderAsyncOptions>;
          isGlobal?: boolean;
      };

type ClientsModuleOptions

type ClientsModuleOptions =
    | Array<ClientProviderOptions>
    | {
          clients: Array<ClientProviderOptions>;
          isGlobal?: boolean;
      };

type ConsumerDeserializer

type ConsumerDeserializer = Deserializer<any, IncomingRequest | IncomingEvent>;

type ConsumerSerializer

type ConsumerSerializer = Serializer<OutgoingResponse, any>;

type IncomingEvent

type IncomingEvent = ReadPacket;

type IncomingRequest

type IncomingRequest = ReadPacket & PacketId;

type IncomingResponse

type IncomingResponse = WritePacket & PacketId;

type MicroserviceOptions

type MicroserviceOptions =
    | GrpcOptions
    | TcpOptions
    | RedisOptions
    | NatsOptions
    | MqttOptions
    | RmqOptions
    | KafkaOptions
    | CustomStrategy;

type MqttEvents

type MqttEvents = {
    connect: OnPacketCallback;
    reconnect: VoidCallback;
    disconnect: OnPacketCallback;
    close: VoidCallback;
    offline: VoidCallback;
    end: VoidCallback;
    error: OnErrorCallback;
    packetreceive: OnPacketCallback;
    packetsend: OnPacketCallback;
};

MQTT events map for the MQTT client. Key is the event name and value is the corresponding callback function.

type MsFundamentalPattern

type MsFundamentalPattern = string | number;

type MsPattern

type MsPattern = MsObjectPattern | MsFundamentalPattern;

type NatsEvents

type NatsEvents = {
    disconnect: DefaultCallback;
    reconnect: DefaultCallback;
    update: (data?: string | number | ServersChangedEvent) => any;
};

Nats events map for the Nats client. Key is the event name and value is the corresponding callback function.

type OutgoingEvent

type OutgoingEvent = ReadPacket;

type OutgoingRequest

type OutgoingRequest = ReadPacket & PacketId;

type OutgoingResponse

type OutgoingResponse = WritePacket & PacketId;

type PatternMetadata

type PatternMetadata = Record<string, any> | string;

type ProducerDeserializer

type ProducerDeserializer = Deserializer<any, IncomingResponse>;

type ProducerSerializer

type ProducerSerializer = Serializer<OutgoingEvent | OutgoingRequest, any>;

type RedisEvents

type RedisEvents = {
    connect: VoidCallback;
    ready: VoidCallback;
    error: OnErrorCallback;
    close: VoidCallback;
    reconnecting: VoidCallback;
    end: VoidCallback;
    warning: OnWarningCallback;
};

Redis events map for the Redis client. Key is the event name and value is the corresponding callback function.

type RmqEvents

type RmqEvents = {
    error: OnErrorCallback;
    disconnect: VoidCallback;
    connect: VoidCallback;
    blocked: OnBlockedCallback;
    unblocked: VoidCallback;
};

RabbitMQ events map for the ampqlip client. Key is the event name and value is the corresponding callback function.

type TcpEvents

type TcpEvents = {
    error: OnErrorCallback;
    connect: VoidCallback;
    end: VoidCallback;
    close: VoidCallback;
    timeout: VoidCallback;
    drain: VoidCallback;
    lookup: OnLookupCallback;
};

TCP events map for the net TCP socket. Key is the event name and value is the corresponding callback function.

type TransportId

type TransportId = Transport | symbol;

Package Files (67)

Dependencies (2)

Dev Dependencies (2)

Peer Dependencies (13)

Badge

To add a badge like this oneto your package's README, use the codes available below.

You may also use Shields.io to create a custom badge linking to https://www.jsdocs.io/package/@nestjs/microservices.

Markdown

[![jsDocs.io](https://img.shields.io/badge/jsDocs.io-reference-blue)](https://www.jsdocs.io/package/@nestjs/microservices)

HTML

<a href="https://www.jsdocs.io/package/@nestjs/microservices"><img src="https://img.shields.io/badge/jsDocs.io-reference-blue" alt="jsDocs.io"></a>

Updated 1 second ago.
Package analyzed in 11700 ms.
Missing or incorrect documentation? Open an issue for this package.

@nestjs/microservices

Install

Overview

Index

Variables

Functions

Classes

Interfaces

Enums

Type Aliases

Variables

variable CONTEXT

variable EventPattern

variable GrpcService

variable MessagePattern

Functions

function Client

Parameter metadata

function createGrpcMethodMetadata

function Ctx

function getGrpcPackageDefinition

function GrpcMethod

function GrpcStreamCall

Parameter service

Parameter service

Parameter method

function GrpcStreamMethod

Parameter service

Parameter service

Parameter method

function KafkaLogger

function Payload

Parameter pipes

Parameter propertyKey

Parameter pipes

Classes

class BaseRpcContext

constructor

property args

method getArgByIndex

Parameter index

method getArgs

class BaseRpcExceptionFilter

method catch

method handleUnknownError

method isError

class ClientGrpcProxy

constructor

property clients

property grpcClients

property logger

property options

property status

property url

method close

method connect

method createClientByServiceName

method createClients

method createServiceMethod

method createStreamServiceMethod

method createUnaryServiceMethod

method dispatchEvent

method getClient

method getClientByServiceName

method getKeepaliveOptions

method getService

method loadProto

method lookupPackage

method on

method publish

method send

method unwrap

class ClientKafka

constructor

property brokers

property client

property clientId

property consumer

property consumerAssignments

property groupId