import { EventEmitter } from "events" ; import * as util from "util" ; import { BoxcarType , IBoxcarMessage , IPendingBoxcar , IProducer , PendingBoxcar , MaxBatchSize , } from "@fluidframework/server-services-core" ; import * as kafka from "kafka-node" ; import { ensureTopics } from "./kafkaTopics" ; /**
 * Kafka producer using the kafka-node library
 */ export class KafkaNodeProducer implements IProducer { private readonly messages = new Map < string , IPendingBoxcar [ ] > ( ) ; private client : kafka . KafkaClient ; private producer : kafka . Producer ; private sendPending : NodeJS . Immediate ; private readonly events = new EventEmitter ( ) ; private connecting = false ; private connected = false ; constructor ( private readonly clientOptions , clientId , private readonly topic , private readonly topicPartitions ? , private readonly topicReplicationFactor ? ) { clientOptions . clientId = clientId ; this . connect ( ) ; } public isConnected ( ) { return this . connected ; } /**
     * Sends the provided message to Kafka
     */ public send ( messages , tenantId , documentId ) { const key = `${ tenantId }/${ documentId }` ; if ( ! this . messages . has ( key ) ) { this . messages . set ( key , [ new PendingBoxcar ( tenantId , documentId ) ] ) ; } const boxcars = this . messages . get ( key ) ; if ( boxcars [ boxcars . length - 1 ] . messages . length + messages . length >= MaxBatchSize ) { boxcars . push ( new PendingBoxcar ( tenantId , documentId ) ) ; } const boxcar = boxcars [ boxcars . length - 1 ] ; boxcar . messages . push ( ... messages ) ; if ( this . connected && boxcar . messages . length >= MaxBatchSize ) { this . sendBoxcars ( boxcars ) ; this . messages . delete ( key ) ; } else { this . requestSend ( ) ; } return boxcar . deferred . promise ; } public async close ( ) { await util . promisify ( ( ( callback ) => this . producer . close ( callback ) ) as any ) ( ) ; await util . promisify ( ( ( callback ) => this . client . close ( callback ) ) as any ) ( ) ; } public on ( event , listener ) { this . events . on ( event , listener ) ; return this ; } public once ( event , listener ) { this . events . once ( event , listener ) ; return this ; } /**
     * Notifies of the need to send pending messages. We defer sending messages to batch together messages
     * to the same partition.
     */ private requestSend ( ) { if ( ! this . connected ) { return ; } if ( this . sendPending ) { return ; } this . sendPending = setImmediate ( ( ) => { this . sendPendingMessages ( ) ; this . sendPending = undefined ; } ) ; } /**
     * Sends all pending messages
     */ private sendPendingMessages ( ) { for ( const [ , value ] of this . messages ) { this . sendBoxcars ( value ) ; } this . messages . clear ( ) ; } private sendBoxcars ( boxcars ) { for ( const boxcar of boxcars ) { const boxcarMessage = { contents : boxcar . messages , documentId : boxcar . documentId , tenantId : boxcar . tenantId , type : BoxcarType } ; const stringifiedMessage = Buffer . from ( JSON . stringify ( boxcarMessage ) ) ; this . producer . send ( [ { key : boxcar . documentId , messages : stringifiedMessage , topic : this . topic } ] , ( error , data ) => { if ( error ) { this . handleError ( error ) ; boxcar . deferred . reject ( error ) ; } else { this . events . emit ( "produced" ) ; boxcar . deferred . resolve ( ) ; } } ) ; } } /**
     * Creates a connection to Kafka. Will reconnect on failure.
     */ private connect ( ) { if ( this . connected || this . connecting ) { return ; } this . connecting = true ; this . client = new kafka . KafkaClient ( this . clientOptions ) ; this . producer = new kafka . Producer ( this . client , { partitionerType : 3 } ) ; this . client . on ( "error" , ( error ) => { this . handleError ( error ) ; } ) ; this . producer . on ( "ready" , async ( ) => { try { await ensureTopics ( this . client , [ this . topic ] , this . topicPartitions , this . topicReplicationFactor ) ; this . connected = true ; this . connecting = false ; this . events . emit ( "connected" ) ; this . sendPendingMessages ( ) ; } catch ( error ) { this . handleError ( error ) ; } } ) ; this . producer . on ( "error" , ( error ) => { this . handleError ( error ) ; } ) ; } /**
     * Handles an error that requires a reconnect to Kafka
     */ private handleError ( error ) { if ( this . client ) { this . client . close ( ) ; this . client = undefined ; } this . connecting = this . connected = false ; this . events . emit ( "error" , error ) ; this . connect ( ) ; } }