@nestjs/microservices

@nestjs/microservices

Version 12.2.0
Published 7 days ago
557 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: {
    <TEventTypes extends Record<string, any>>(
        topic: keyof TEventTypes
    ): EventDataMethodDecorator<TEventTypes, keyof TEventTypes>;
    <TEventTypes extends Record<string, any>>(
        topic: keyof TEventTypes,
        transport: Transport | symbol
    ): EventDataMethodDecorator<TEventTypes, keyof TEventTypes>;
    <TEventTypes extends Record<string, any>>(
        topic: keyof TEventTypes,
        extras: Record<string, any>
    ): EventDataMethodDecorator<TEventTypes, keyof TEventTypes>;
    <TEventTypes extends Record<string, any>>(
        topic: keyof TEventTypes,
        transport: Transport | symbol,
        extras: Record<string, any>
    ): EventDataMethodDecorator<TEventTypes, keyof TEventTypes>;
    <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;
    (propertyKey: string, options: ParameterDecoratorOptions): ParameterDecorator;
    (options: ParameterDecoratorOptions): 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.
Microservice message pattern payload parameter decorator. Extracts a property from the payload object with additional options.
For example, extracting a single param with schema:
create(@Payload('data', { schema: z.string() }) data: string)
Parameter propertyKey
name of single property to extract from the message payload
Parameter options
options object containing additional configuration for the decorator, such as pipes and schema
Microservice message pattern payload parameter decorator.
For example, passing schema as options:
create(@Payload({ schema: z.object({ data: z.string() }) }) payload)
Parameter options
options object containing additional configuration for the decorator, such as pipes and schema

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> {}

property logger

protected static readonly logger: Logger;

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>() => Promise<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 handleClose

handleClose: () => void;

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[]>;
        maxConnectionAttempts?: number;
    }
);

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[]>;
    maxConnectionAttempts?: number;
};

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 assertTopicLength

protected assertTopicLength: (topic: string) => void;

A topic's length is encoded in two bytes. mqtt.js only fails on a longer topic once it has written the start of the packet, so the broker would read the packets sent after it as the rest of that packet.

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 handleClose

handleClose: () => void;

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.js').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: Error | null, natsMsg: NatsMsg) => Promise<any>;

method dispatchEvent

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

method handleClose

handleClose: () => void;

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 onDispatchHook

protected onDispatchHook?: (packet: ReadPacket) => void;

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 createPacket

protected createPacket: <TInput>(pattern: any, data: TInput) => ReadPacket;

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[]>;
                  maxConnectionAttempts?: number;
              })
            | {
                  [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;
                  incompleteMessageTimeout?: number;
                  maxSendBufferSize?: 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[]>;
                  maxConnectionAttempts?: number;
              })
            | {
                  [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;
                  incompleteMessageTimeout?: number;
                  maxSendBufferSize?: 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 setOnDispatchHook

setOnDispatchHook: (hook: (packet: ReadPacket) => void) => void;

Sets a hook that is called with every outgoing packet - requests and events alike - right before it is handed to the transport. The hook may assign packet.metadata; it must not replace pattern or data.
The client-side counterpart of Server#setOnProcessingStartHook: what one side attaches here, the other reads from the RPC context, which is what lets instrumentation carry a trace id across a hop without touching the application's payloads.

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: () => Promise<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 handleClose

handleClose: () => void;

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;
    incompleteMessageTimeout?: number;
    maxSendBufferSize?: number;
});

property connectionPromise

protected connectionPromise: Promise<any>;

property host

protected readonly host: string;

property incompleteMessageTimeout

protected readonly incompleteMessageTimeout?: number;

property logger

protected readonly logger: Logger;

property maxBufferSize

protected readonly maxBufferSize?: number;

property maxSendBufferSize

protected readonly maxSendBufferSize?: 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 GrpcAbortedException

class GrpcAbortedException extends GrpcException {}

constructor

constructor(message?: string);

class GrpcAlreadyExistsException

class GrpcAlreadyExistsException extends GrpcException {}

constructor

constructor(message?: string);

class GrpcCancelledException

class GrpcCancelledException extends GrpcException {}

constructor

constructor(message?: string);

class GrpcDataLossException

class GrpcDataLossException extends GrpcException {}

constructor

constructor(message?: string);

class GrpcDeadlineExceededException

class GrpcDeadlineExceededException extends GrpcException {}

constructor

constructor(message?: string);

class GrpcException

class GrpcException extends Error {}

constructor

constructor(message: string, code?: GrpcStatus);

method getCode

getCode: () => GrpcStatus;

method getError

getError: () => GrpcExceptionBody;

class GrpcExceptionFilter

class GrpcExceptionFilter implements RpcExceptionFilter {}

method catch

catch: (
    exception: unknown,
    host: ArgumentsHost
) => Observable<GrpcExceptionBody>;

class GrpcFailedPreconditionException

class GrpcFailedPreconditionException extends GrpcException {}

constructor

constructor(message?: string);

class GrpcInternalException

class GrpcInternalException extends GrpcException {}

constructor

constructor(message?: string);

class GrpcInvalidArgumentException

class GrpcInvalidArgumentException extends GrpcException {}

constructor

constructor(message?: string);

class GrpcNotFoundException

class GrpcNotFoundException extends GrpcException {}

constructor

constructor(message?: string);

class GrpcOutOfRangeException

class GrpcOutOfRangeException extends GrpcException {}

constructor

constructor(message?: string);

class GrpcPermissionDeniedException

class GrpcPermissionDeniedException extends GrpcException {}

constructor

constructor(message?: string);

class GrpcResourceExhaustedException

class GrpcResourceExhaustedException extends GrpcException {}

constructor

constructor(message?: string);

class GrpcUnauthenticatedException

class GrpcUnauthenticatedException extends GrpcException {}

constructor

constructor(message?: string);

class GrpcUnavailableException

class GrpcUnavailableException extends GrpcException {}

constructor

constructor(message?: string);

class GrpcUnimplementedException

class GrpcUnimplementedException extends GrpcException {}

constructor

constructor(message?: string);

class GrpcUnknownException

class GrpcUnknownException extends GrpcException {}

constructor

constructor(message?: string);

class IdentityDeserializer

class IdentityDeserializer implements Deserializer {}

method deserialize

deserialize: (value: any) => any;

class IncomingRequestDeserializer

class IncomingRequestDeserializer implements ConsumerDeserializer {}

method deserialize

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

method isExternal

isExternal: (value: any) => boolean;

method mapToSchema

mapToSchema: (
    value: any,
    options?: Record<string, any>
) => IncomingRequest | IncomingEvent;

class IncomingResponseDeserializer

class IncomingResponseDeserializer implements ProducerDeserializer {}

method deserialize

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

method isExternal

isExternal: (value: any) => boolean;

method mapToSchema

mapToSchema: (value: any) => IncomingResponse;

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;

property maxMessageSize

protected readonly maxMessageSize: number;

method decode

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

method decodeAsString

protected decodeAsString: (value: Buffer) => string | null;

method getMessageSize

protected getMessageSize: (data: any) => number;

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: Record<string, unknown>;
};

method getPreviousAssignment

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

method protocol

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

class KafkaRequestDeserializer

class KafkaRequestDeserializer extends IncomingRequestDeserializer {}

method mapToSchema

mapToSchema: (
    data: KafkaRequest,
    options?: Record<string, any>
) => IncomingRequest | IncomingEvent;

class KafkaResponseDeserializer

class KafkaResponseDeserializer implements Deserializer<any, IncomingResponse> {}

method deserialize

deserialize: (message: any, options?: Record<string, any>) => IncomingResponse;

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 NatsRequestJSONDeserializer

class NatsRequestJSONDeserializer extends IncomingRequestDeserializer {}

method deserialize

deserialize: (
    value: NatsMsg,
    options?: Record<string, any>
) =>
    | IncomingRequest
    | IncomingEvent
    | Promise<IncomingRequest>
    | Promise<IncomingEvent>;

class NatsResponseJSONDeserializer

class NatsResponseJSONDeserializer extends IncomingResponseDeserializer {}

method deserialize

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

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 getTransportServer

getTransportServer: () => ITransportServer;

Returns the underlying transport server instance. Use this to close only the transport (port/connection) without triggering the full application shutdown lifecycle.

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 registerPreRequestHook

registerPreRequestHook: (...hooks: PreRequestHook[]) => this;

Registers a global preRequest hook (executed before all enhancers for every pattern handler).
Parameter hooks

method resolveAsyncOptions

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

method setIsConnectedToApplication

setIsConnectedToApplication: (isConnectedToApplication: boolean) => void;

Sets the flag indicating that the microservice is connected to a hybrid application, which sets up the container they share (gateways, clients and lifecycle hooks).
Parameter isConnectedToApplication
Value to set

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.

method getMetadata

getMetadata: () => PacketMetadata | undefined;

Returns the metadata the client attached to the packet, if any.

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
> implements ITransportServer {}

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 propagatesEventHandlerErrors

readonly propagatesEventHandlerErrors: boolean;

Whether this transport hands an event handler failure to its client library, which then reports it. The base handleEvent connects the stream without subscribing, so nothing observes the failure and Nest logs it instead. ServerKafka awaits the stream and sets this to true.

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 assertValidPacketId

protected assertValidPacketId: (id: unknown) => void;

Throws unless the id of an incoming request is a number or a short string.
Ids are client-controlled and every reply to the request carries a copy, so a long string or a nested value would be copied and serialized once per reply, which a handler that streams its response multiplies.
Parameter id
request id

method close

abstract close: () => any;

Method called when server is being terminated.

method createProcessingEndHookRunner

protected createProcessingEndHookRunner: (context: BaseRpcContext) => () => void;

Returns a function that runs the processing end hook exactly once.
An event handler can fail either by rejecting or by returning a stream that errors, and ServerKafka#handleEvent awaits that stream, so both the finalize teardown and the catch block can be reached for a single event. The hook closes a span, so it must not run twice.

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[]>;
                  maxConnectionAttempts?: number;
              })
            | {
                  [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;
                  incompleteMessageTimeout?: number;
                  maxSendBufferSize?: number;
                  drainTimeout?: 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[]>;
                  maxConnectionAttempts?: number;
              })
            | {
                  [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;
                  incompleteMessageTimeout?: number;
                  maxSendBufferSize?: number;
                  drainTimeout?: number;
              },
        Attribute extends keyof Options,
        DefaultValue extends Options[Attribute] = Options[Attribute]
    >(
        obj: Options,
        prop: Attribute,
        defaultValue: DefaultValue
    ): Required<Options>[Attribute];
};

method getPatternAsString

protected getPatternAsString: (pattern: unknown) => string;

Returns the string representation of an incoming message pattern.
Patterns are client-controlled: serializing a deeply nested one makes JSON.stringify throw a RangeError, which must not escape the message handler (as an unhandled promise rejection that terminates the process).
Parameter pattern
client pattern
Returns
string

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 handleRequest

protected handleRequest: (
    context: BaseRpcContext,
    produce: () => Promise<Observable<any>>,
    respond: (data: WritePacket) => Promise<unknown> | void
) => void;

Handles a request-response message. produce returns the response stream, whose values are replied through respond, and the processing end hook runs once that stream settles (or once produce rejects).

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?: () => T
) => T | Promise<T>;

method loadPackageSynchronously

protected loadPackageSynchronously: <T = any>(
    name: string,
    ctx: string,
    loader?: () => T
) => 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 runWithProcessingHooks

protected runWithProcessingHooks: (
    context: BaseRpcContext,
    fn: (runEndHook: () => void) => Promise<void>,
    endHookContext?: BaseRpcContext
) => void;

Runs fn between the processing start and end hooks.
fn receives a runner that closes the span exactly once and attaches it to the teardown of whatever stream it produces. If fn rejects before that teardown can run, the span is closed here and the rejection travels on to the transport's client library.
endHookContext is for transports that report a different context to the end hook than to the start hook (gRPC passes the request).

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.

method warnIfEventHandlerReplies

protected warnIfEventHandlerReplies: (
    handler: MessageHandler,
    pattern: string
) => void;

Logs a deprecation warning the first time an event handler is about to reply to a request (a packet with an id, sent with ClientProxy#send). Event handlers still reply to requests, except on Kafka, but will not in the next major version.

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 propagatesEventHandlerErrors

readonly propagatesEventHandlerErrors: boolean;

property transportId

transportId: TransportId;

method addHandler

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

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>() => Promise<T>;

method getHandlerByPattern

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

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>;

Handles a message as an event: one for an @EventPattern() handler, or one without the reply headers (see handleMessage).
The returned promise settles once the stream the handler returns completes, and kafkajs only then moves on to the next message of the partition, so a handler whose stream does not complete holds the partition. A handler that fails (as when a guard or a pipe rejects the message) rejects into kafkajs, which does not commit the message and delivers it again: it retries it, and once its retries are exhausted it restarts the consumer from the last committed offset. Handlers of topics that other producers write to should complete their streams, and should not fail on messages they cannot process unless these are to be delivered again.

method handleMessage

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

Handles a message received by the consumer.
A message with the kafka_correlationId and kafka_replyTopic headers, which ClientKafka#send() adds, is a request: what its handler responds with, an error included, is published to the reply topic. Any other message is handled as an event (see handleEvent), also when its handler is a @MessagePattern() one, so that such handlers can consume topics written by producers that do not add these headers.

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[]>;
        maxConnectionAttempts?: number;
    }
);

property connectionAttempts

protected connectionAttempts: number;

property hasConnected

protected hasConnected: boolean;

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 assertTopicLength

protected assertTopicLength: (topic: string) => void;

A topic's length is encoded in two bytes, and "/reply" can take a topic the broker delivered past it. mqtt.js only fails on a longer topic once it has written the start of the packet, so the broker would read the packets sent after it as the rest of that packet.

method bindEvents

bindEvents: (mqttClient: MqttClient) => void;

method close

close: () => void;

method createMqttClient

createMqttClient: () => Promise<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,
    callback?: (err?: unknown, ...optionalParams: unknown[]) => void
) => 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.js').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) => Promise<void>;

method close

close: () => Promise<void>;

method createRedisClient

createRedisClient: () => Promise<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
) => Promise<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>() => Promise<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
) => Promise<unknown>;

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;
    incompleteMessageTimeout?: number;
    maxSendBufferSize?: number;
    drainTimeout?: number;
});

property disconnectNotifiers

protected readonly disconnectNotifiers: WeakMap<TcpSocket, Observable<void>>;

Notifiers that emit once a socket bound by "bindHandler" can no longer be written to. They replay, so that a response stream that only starts after the peer has gone ends right away as well.

property drainTimeout

protected readonly drainTimeout?: number;

property host

protected readonly host: string;

property incompleteMessageTimeout

protected readonly incompleteMessageTimeout?: number;

property isManuallyTerminated

protected isManuallyTerminated: boolean;

property maxBufferSize

protected readonly maxBufferSize?: number;

property maxSendBufferSize

protected readonly maxSendBufferSize?: number;

property openSockets

protected readonly openSockets: Set<net.Socket>;

Sockets accepted by this server that are still open. "net.Server#close" only stops the server from accepting new connections, so these are tracked separately and torn down on "close" - otherwise the process outlives the shutdown and handlers keep running on already established connections.

property pendingEventListeners

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

property port

protected readonly port: number;

property retryAttemptsCount

protected retryAttemptsCount: number;

property retryTimer

protected retryTimer?: 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 closeOpenSockets

protected closeOpenSockets: () => void;

Destroys every socket still open. Called on shutdown so that "close" does not leave the process alive, and so that no further messages are dispatched to handlers over connections established before the shutdown.

method endOnDisconnect

protected endOnDisconnect: (
    result: unknown,
    context: TcpContext,
    socket: TcpSocket
) => Observable<unknown>;

Turns a handler result into the response stream. As for HTTP, the stream ends once the peer disconnects only if the handler itself returned (or resolved to) an Observable: any other handler, and the interceptors around it, run to completion.

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 trackDisconnect

protected trackDisconnect: (socket: TcpSocket) => void;

Notes when the socket can no longer be written to: "TcpSocket" drops every message once the socket has closed or errored.

method trackOpenSocket

protected trackOpenSocket: (socket: Socket) => void;

Keeps a reference to an accepted socket so that it can be destroyed when the server is closed, and drops it again once it closes on its own.

method unwrap

unwrap: <T>() => T;

class TcpContext

class TcpContext extends BaseRpcContext<TcpContextArgs> {}

constructor

constructor(args: TcpContextArgs);

method getMetadata

getMetadata: () => PacketMetadata | undefined;

Returns the metadata the client attached to the packet, if any.

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 handleDataError

protected handleDataError: (
    e: any,
    { destroy }?: { destroy?: boolean | undefined }
) => void;

Surfaces a framing error on the underlying socket and closes it. Pass "destroy" for a peer that cannot be trusted to answer a FIN.

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 GrpcExceptionBody

interface GrpcExceptionBody {}

property code

code: GrpcStatus;

property message

message: string;

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 drainTimeout

drainTimeout?: number;

property incompleteMessageTimeout

incompleteMessageTimeout?: number;

property maxBufferSize

maxBufferSize?: number;

property maxSendBufferSize

maxSendBufferSize?: number;

property pauseOnBackpressure

pauseOnBackpressure?: boolean;

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;

property maxMessageSize

maxMessageSize?: number;

Maximum size of a message in bytes, after decompression: its key, its headers and, unless keepBinary is set, its value. A larger message is logged and skipped instead of being decoded. 16777216 (16 MiB)

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[]>;
    /**
     * Maximum number of connection attempts before the server gives up and
     * reports the failure to the bootstrap callback.
     * Applies only to the server configuration, and only to the initial
     * connection - once connected, reconnections are retried indefinitely.
     * -1 === infinite
     * @default -1
     */
    maxConnectionAttempts?: number;
};

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 metadata

metadata?: PacketMetadata;

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 | Promise<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;
    /**
     * How long a peer may stay silent in the middle of a packet before the
     * connection is dropped, in milliseconds (default: 30000). The timer is
     * refreshed on every read, so a slow but progressing transfer is never
     * interrupted; it only fires when a peer stops sending mid-packet, which
     * would otherwise pin the partial packet in memory indefinitely.
     * Set to 0 to disable.
     */
    incompleteMessageTimeout?: number;
    /**
     * Maximum number of bytes that may sit queued for a peer that is not
     * reading them, in bytes (default: 128MB). The client keeps reading
     * replies while the outgoing buffer is backed up, and the connection is
     * dropped once this limit is passed. Set to 0 to disable.
     */
    maxSendBufferSize?: 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;
    /**
     * How long a peer may stay silent in the middle of a packet before the
     * connection is dropped, in milliseconds (default: 30000). The timer is
     * refreshed on every read, so a slow but progressing transfer is never
     * interrupted; it only fires when a peer stops sending mid-packet, which
     * would otherwise pin the partial packet in memory indefinitely.
     * Set to 0 to disable.
     */
    incompleteMessageTimeout?: number;
    /**
     * Maximum number of response bytes that may sit queued for a single peer
     * that is not reading them, in bytes (default: 128MB). Reading from a peer
     * is suspended while its outgoing buffer is backed up, and the connection
     * is dropped once this limit is passed. Set to 0 to disable.
     */
    maxSendBufferSize?: number;
    /**
     * How long the responses queued for a peer may go without any of them
     * being written out, while reading from that peer is suspended, before
     * the connection is dropped, in milliseconds (default: 30000). The timer
     * is refreshed whenever queued responses have been written out in full,
     * so a peer that keeps reading is given more time, while one that stops
     * reading can no longer hold its responses, up to "maxSendBufferSize",
     * in memory indefinitely. Set to 0 to disable.
     */
    drainTimeout?: 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 GrpcStatus

enum GrpcStatus {
    OK = 0,
    CANCELLED = 1,
    UNKNOWN = 2,
    INVALID_ARGUMENT = 3,
    DEADLINE_EXCEEDED = 4,
    NOT_FOUND = 5,
    ALREADY_EXISTS = 6,
    PERMISSION_DENIED = 7,
    RESOURCE_EXHAUSTED = 8,
    FAILED_PRECONDITION = 9,
    ABORTED = 10,
    OUT_OF_RANGE = 11,
    UNIMPLEMENTED = 12,
    INTERNAL = 13,
    UNAVAILABLE = 14,
    DATA_LOSS = 15,
    UNAUTHENTICATED = 16,
}

member ABORTED

ABORTED = 10

member ALREADY_EXISTS

ALREADY_EXISTS = 6

member CANCELLED

CANCELLED = 1

member DATA_LOSS

DATA_LOSS = 15

member DEADLINE_EXCEEDED

DEADLINE_EXCEEDED = 4

member FAILED_PRECONDITION

FAILED_PRECONDITION = 9

member INTERNAL

INTERNAL = 13

member INVALID_ARGUMENT

INVALID_ARGUMENT = 3

member NOT_FOUND

NOT_FOUND = 5

member OK

OK = 0

member OUT_OF_RANGE

OUT_OF_RANGE = 11

member PERMISSION_DENIED

PERMISSION_DENIED = 7

member RESOURCE_EXHAUSTED

RESOURCE_EXHAUSTED = 8

member UNAUTHENTICATED

UNAUTHENTICATED = 16

member UNAVAILABLE

UNAVAILABLE = 14

member UNIMPLEMENTED

UNIMPLEMENTED = 12

member UNKNOWN

UNKNOWN = 2

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 EventDataMethodDecorator

type EventDataMethodDecorator<
    TEventTypes extends Record<string, any>,
    TKey extends keyof TEventTypes
> = (
    target: object,
    propertyKey: string | symbol,
    descriptor: TypedPropertyDescriptor<
        (data: TEventTypes[TKey], ...args: unknown[]) => any
    >
) => void;

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 | MsPatternMatch;

type MsPatternMatch

type MsPatternMatch = RegExp;

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 PacketMetadata

type PacketMetadata = Record<string, string>;

Out-of-band values that travel with a packet but are not part of its payload - a trace id, a tenant, a locale. Set on the client through ClientProxy#setOnDispatchHook, read on the server from the RPC context.

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 (77)

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 11637 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

Parameter propertyKey

Parameter options

Parameter options

Classes

class BaseRpcContext

constructor

property args

method getArgByIndex

Parameter index

method getArgs

class BaseRpcExceptionFilter

property logger

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