import { Kafka , Producer } from 'kafkajs' ; import { isArray } from 'util' ; import { Nullable } from '@mds-core/mds-types' ; import logger from '@mds-core/mds-logger' ; import { isDefined , ClientDisconnectedError , ExceptionMessages } from '@mds-core/mds-utils' ; import { StreamProducer } from '../stream-interface' ; import { getKafkaBrokers } from './helpers' ; export interface KafkaStreamProducerOptions { clientId : string ; } const createStreamProducer = async ( { clientId = 'writer' } = { } ) => { try { const kafka = new Kafka ( { clientId , brokers : getKafkaBrokers ( ) } ) ; const producer = kafka . producer ( ) ; await producer . connect ( ) ; return producer ; } catch ( err ) { logger . error ( err ) ; } return null ; } ; const disconnectProducer = async ( producer ) => { if ( isDefined ( producer ) ) { await producer . disconnect ( ) ; } } ; export const KafkaStreamProducer = < TMessage > ( topic , options ? ) : StreamProducer < TMessage > => { let producer = null ; return { initialize : async ( ) => { if ( ! producer ) { producer = await createStreamProducer ( options ) ; } } , write : async ( message ) => { if ( isDefined ( producer ) ) { const messages = ( isArray ( message ) ? message : [ message ] ) . map ( msg => { return { value : JSON . stringify ( msg ) } ; } ) ; await producer . send ( { topic , messages } ) ; return ; } throw new ClientDisconnectedError ( ExceptionMessages . INITIALIZE_CLIENT_MESSAGE ) ; } , shutdown : async ( ) => { await disconnectProducer ( producer ) ; producer = null ; } } ; } ;