CU-86epthegu Log consumer run options and log notice when batch messa…

dist/index.d.ts

@@ -0,0 +1,5 @@
+export * from './kafka.server';
+export * from './kafka.client';
+export * from './kafka.context';
+export * from './kafka.deserializer';
+export * from './kafka.decorators';

dist/kafka.client.d.ts

@@ -0,0 +1,17 @@
+import { ClientKafka, ReadPacket } from '@nestjs/microservices';
+import { Observable } from 'rxjs';
+export type NestjsKafkaBatchTopicMessage<TInput = any> = {
+    pattern: any;
+    data: TInput[];
+};
+export declare class KafkaBatchClient extends ClientKafka {
+    protected publish(): () => void;
+    emitBatch<TResult = any, TInput = any>(pattern: any, data: {
+        messages: TInput[];
+    }): Observable<TResult>;
+    protected dispatchBatchEvent<TInput = any>(packets: ReadPacket<{
+        messages: TInput[];
+    }>): Promise<any>;
+    emitBatchTopics<TResult = any, TInput = any>(topicMessages: NestjsKafkaBatchTopicMessage<TInput>[]): Observable<TResult>;
+    protected dispatchBatchTopics<TInput = any>(topicMessages: NestjsKafkaBatchTopicMessage<TInput>[]): Promise<any>;
+}

dist/kafka.context.d.ts

@@ -0,0 +1,26 @@
+import { BaseRpcContext } from '@nestjs/microservices';
+import type { Consumer, KafkaMessage, Offsets, OffsetsByTopicPartition } from 'kafkajs';
+type KafkaBatchContextArgs = [
+    messages: KafkaMessage[],
+    consumer: Consumer,
+    resolveOffset: (offset: string) => void,
+    heartbeat: () => Promise<void>,
+    pause: () => void,
+    commitOffsetsIfNecessary: (offsets?: Offsets) => Promise<void>,
+    uncommittedOffsets: () => OffsetsByTopicPartition,
+    isRunning: () => boolean,
+    isStale: () => boolean
+];
+export declare class KafkaBatchContext extends BaseRpcContext<KafkaBatchContextArgs> {
+    constructor(args: KafkaBatchContextArgs);
+    getMessages(): KafkaMessage[];
+    getConsumer(): Consumer;
+    getResolveOffset(): (offset: string) => void;
+    getHeartbeat(): () => Promise<void>;
+    getPause(): () => void;
+    getCommitOffsetsIfNecessary(): (offsets?: Offsets) => Promise<void>;
+    getUncommittedOffsets(): OffsetsByTopicPartition;
+    getIsRunning(): boolean;
+    getIsStale(): boolean;
+}
+export {};

dist/kafka.decorators.d.ts

@@ -0,0 +1 @@
+export declare const BatchProcessor: <T = string>(event: T) => <TFunction extends Function, Y>(target: object | TFunction, propertyKey?: string | symbol, descriptor?: TypedPropertyDescriptor<Y>) => void;

dist/kafka.deserializer.d.ts

@@ -0,0 +1,6 @@
+import { IncomingEvent, IncomingRequest } from '@nestjs/microservices';
+import { IncomingRequestDeserializer } from '@nestjs/microservices/deserializers/incoming-request.deserializer';
+import { KafkaMessage } from 'kafkajs';
+export declare class KafkaBatchDeserializer extends IncomingRequestDeserializer {
+    mapToSchema(data: KafkaMessage[], options: Record<string, any>): IncomingRequest | IncomingEvent;
+}

dist/kafka.server.d.ts

@@ -0,0 +1,22 @@
+import { KafkaOptions, ServerKafka } from '@nestjs/microservices';
+import { Consumer, EachBatchHandler, EachBatchPayload } from '@nestjs/microservices/external/kafka.interface';
+import { Logger } from '@nestjs/common';
+export interface BatchKafkaOptionsConfig extends Omit<KafkaOptions['options'], 'producer' | 'producerOnlyMode'> {
+    run?: BatchConsumerRunConfig;
+}
+export type BatchConsumerRunConfig = {
+    autoCommit?: boolean;
+    autoCommitInterval?: number;
+    autoCommitThreshold?: number;
+    eachBatchAutoResolve?: boolean;
+    partitionsConsumedConcurrently?: number;
+};
+export declare class KafkaBatchServer extends ServerKafka {
+    readonly options: BatchKafkaOptionsConfig;
+    logger: Logger;
+    constructor(options: BatchKafkaOptionsConfig);
+    bindEvents(consumer: Consumer): Promise<void>;
+    getMessageHandler(): EachBatchHandler;
+    handleMessage(payload: EachBatchPayload): Promise<any>;
+    protected initializeDeserializer(options: BatchKafkaOptionsConfig): void;
+}