import { IConsumer , IPartitionLambdaFactory , IResources , IResourcesFactory , } from "@fluidframework/server-services-core" ; import * as moniker from "moniker" ; import { Provider } from "nconf" ; import { KafkaNodeConsumer } from "./kafkaNodeConsumer" ; export interface IKafkaResources extends IResources { lambdaFactory : IPartitionLambdaFactory ; consumer : IConsumer ; config : Provider ; } export class KafkaResources implements IKafkaResources { constructor ( public lambdaFactory , public consumer , public config ) { } public async dispose ( ) { const consumerClosedP = this . consumer . close ( ) ; await Promise . all ( [ consumerClosedP ] ) ; } } export class KafkaResourcesFactory implements IResourcesFactory < KafkaResources > { constructor ( private readonly name , private readonly lambdaModule ) { } public async create ( config ) { const plugin = require ( this . lambdaModule ) ; const lambdaFactory = await plugin . create ( config ) as IPartitionLambdaFactory ; const kafkaEndpoint = config . get ( "kafka:lib:endpoint" ) ; const zookeeperEndpoint = config . get ( "zookeeper:endpoint" ) ; const numberOfPartitions = config . get ( "kafka:lib:numberOfPartitions" ) ; const replicationFactor = config . get ( "kafka:lib:replicationFactor" ) ; const streamConfig = config . get ( `lambdas:${ this . name }` ) ; const groupId = streamConfig . group ; const receiveTopic = streamConfig . topic ; const clientId = moniker . choose ( ) ; const consumer = new KafkaNodeConsumer ( { kafkaHost : kafkaEndpoint } , clientId , groupId , receiveTopic , zookeeperEndpoint , numberOfPartitions , replicationFactor ) ; return new KafkaResources ( lambdaFactory , consumer , config ) ; } }