import { inspect } from "util" ; import winston from "winston" ; import { IContextErrorData , IProducer } from "@fluidframework/server-services-core" ; import { KafkaNodeProducer } from "@fluidframework/server-services-ordering-kafkanode" ; import { RdkafkaProducer } from "@fluidframework/server-services-ordering-rdkafka" ; import { Lumberjack } from "@fluidframework/server-services-telemetry" ; export function createProducer ( type , kafkaEndPoint , clientId , topic , enableIdempotence ? , pollIntervalMs ? , numberOfPartitions ? , replicationFactor ? , sslCACertFilePath ? ) { let producer ; if ( type === "rdkafka" ) { producer = new RdkafkaProducer ( { kafka : [ kafkaEndPoint ] } , clientId , topic , { enableIdempotence , pollIntervalMs , numberOfPartitions , replicationFactor , sslCACertFilePath } ) ; producer . on ( "error" , ( error , errorData ) => { if ( errorData ?. restart ) { throw new Error ( error ) ; } else { winston . error ( "Kafka Producer emitted an error that is not configured to restart the process." ) ; winston . error ( inspect ( error ) ) ; Lumberjack . error ( "Kafka Producer emitted an error that is not configured to restart the process." , undefined , error ) ; } } ) ; } else { producer = new KafkaNodeProducer ( { kafkaHost : kafkaEndPoint } , clientId , topic , numberOfPartitions , replicationFactor ) ; producer . on ( "error" , ( error ) => { winston . error ( error ) ; Lumberjack . error ( error ) ; } ) ; } return producer ; }