import { Kafka , EachMessagePayload , Consumer } from 'kafkajs' ; import { Nullable } from '@mds-core/mds-types' ; import logger from '@mds-core/mds-logger' ; import { isDefined } from '@mds-core/mds-utils' ; import { StreamConsumer } from '../stream-interface' ; import { getKafkaBrokers } from './helpers' ; export interface KafkaStreamConsumerOptions { clientId : string ; groupId : string ; } const createStreamConsumer = async ( topic , eachMessage , { clientId = 'client' , groupId = 'group' } = { } ) => { try { const kafka = new Kafka ( { clientId , brokers : getKafkaBrokers ( ) } ) ; const consumer = kafka . consumer ( { groupId } ) ; await consumer . connect ( ) ; await consumer . subscribe ( { topic } ) ; await consumer . run ( { eachMessage } ) ; return consumer ; } catch ( err ) { logger . error ( err ) ; } return null ; } ; const disconnectConsumer = async ( consumer ) => { if ( isDefined ( consumer ) ) { await consumer . disconnect ( ) ; } } ; export const KafkaStreamConsumer = ( topic , eachMessage , options ? ) : StreamConsumer => { let consumer = null ; return { initialize : async ( ) => { if ( ! consumer ) { consumer = await createStreamConsumer ( topic , eachMessage , options ) ; } } , shutdown : async ( ) => { await disconnectConsumer ( consumer ) ; consumer = null ; } } ; } ;