import { Module , DynamicModule , Global , Provider } from '@nestjs/common' ; import { KafkaService } from './kafka.service' ; import { KafkaModuleOption , KafkaModuleOptionsAsync , KafkaOptionsFactory } from './interfaces' ; import { KafkaModuleOptionsProvider } from './kafka-module-options.provider' ; import { KAFKA_MODULE_OPTIONS } from './constants' ; @ Global ( ) @ Module ( { } ) export class KafkaModule { static register ( options ) { const clients = ( options || [ ] ) . map ( ( item ) => ( { provide : item . name , useValue : new KafkaService ( item . options ) , } ) ) ; return { module : KafkaModule , providers : clients , exports : clients , } ; } public static registerAsync ( consumers , connectOptions ) { const clients = [ ] ; for ( const consumer of consumers ) { clients . push ( { provide : consumer , useFactory : async ( kafkaModuleOptionsProvider ) => { return new KafkaService ( kafkaModuleOptionsProvider . getOptionsByName ( consumer ) ) ; } , inject : [ KafkaModuleOptionsProvider ] , } ) ; } const createKafkaModuleOptionsProvider = this . createKafkaModuleOptionsProvider ( connectOptions ) ; return { module : KafkaModule , imports : connectOptions . imports || [ ] , providers : [ createKafkaModuleOptionsProvider , KafkaModuleOptionsProvider , ... clients , ] , exports : [ createKafkaModuleOptionsProvider , ... clients , ] , } ; } private static createKafkaModuleOptionsProvider ( options ) { if ( options . useFactory ) { return { provide : KAFKA_MODULE_OPTIONS , useFactory : options . useFactory , inject : options . inject || [ ] , } ; } return { provide : KAFKA_MODULE_OPTIONS , useFactory : async ( optionsFactory ) => await optionsFactory . creatKafkaModuleOptions ( ) , inject : [ options . useExisting || options . useClass ] , } ; } }