import { strict as assert } from "assert" ; import { EventEmitter } from "events" ; import * as core from "@fluidframework/server-services-core" ; import { TestContext } from "./testContext" ; export class TestConsumer implements core . IConsumer { private readonly emitter = new EventEmitter ( ) ; private pausedQueue : string [ ] = null ; private failOnCommit = false ; private readonly context = new TestContext ( ) ; constructor ( public groupId , public topic ) { } public setFailOnCommit ( value ) { this . failOnCommit = value ; } public isConnected ( ) { return true ; } /**
     * Returns the offset of the latest consumsed message
     */ public getLatestMessageOffset ( partitionId ) { return undefined ; } public async commitCheckpoint ( partitionId , queuedMessage ) { assert ( partitionId === 0 ) ; if ( this . failOnCommit ) { return Promise . reject ( new Error ( "TestConsumer set to fail on commit" ) ) ; } else { this . context . checkpoint ( queuedMessage ) ; return ; } } public getOffset ( ) { return this . context . offset ; } public async waitForOffset ( offset ) { return this . context . waitForOffset ( offset ) ; } public on ( event , listener ) { this . emitter . on ( event , listener as ( ... args : any [ ] ) => void ) ; return this ; } public once ( event , listener ) { this . emitter . once ( event , listener as ( ... args : any [ ] ) => void ) ; return this ; } public close ( ) { return Promise . resolve ( ) ; } public async pause ( ) { if ( ! this . pausedQueue ) { this . pausedQueue = [ ] ; } } public async resume ( ) { if ( ! this . pausedQueue ) { return ; } const pendingMessages = this . pausedQueue ; this . pausedQueue = null ; for ( const message of pendingMessages ) { this . emit ( message ) ; } } /**
     * Manually signal an error
     */ public emitError ( error ) { this . emitter . emit ( "error" , error ) ; } public emit ( message ) { if ( this . pausedQueue ) { this . pausedQueue . push ( message ) ; } else { this . emitter . emit ( "data" , message ) ; } } public rebalance ( ) { this . emitter . emit ( "rebalancing" ) ; this . emitter . emit ( "rebalanced" , [ { topic : this . topic , offset : 0 , partition : 0 } ] ) ; } } export class TestProducer implements core . IProducer { constructor ( private readonly kafka ) { } public isConnected ( ) { return true ; } public send ( messages , key ) { for ( const message of messages ) { this . kafka . addMessage ( message , key ) ; } return Promise . resolve ( ) ; } public close ( ) { return Promise . resolve ( ) ; } public on ( event , listener ) { return this ; } public once ( event , listener ) { return this ; } } /**
 * Test Kafka implementation. Allows for the creation of a joined producer/consumer pair.
 */ export class TestKafka { public static createdQueuedMessage ( offset , metadata ? ) { return { topic : "topic" , partition : 0 , offset , value : "" } ; } private readonly messages : core . IQueuedMessage [ ] = [ ] ; private offset = 0 ; private readonly consumers : TestConsumer [ ] = [ ] ; public createProducer ( ) { return new TestProducer ( this ) ; } public createConsumer ( ) { const consumer = new TestConsumer ( "test" , "test" ) ; this . consumers . push ( consumer ) ; return consumer ; } public getRawMessages ( ) { return this . messages ; } public addMessage ( message , topic ) { const offset = this . offset ++ ; const queuedMessage = TestKafka . createdQueuedMessage ( offset ) ; queuedMessage . value = message ; queuedMessage . topic = topic ; this . messages . push ( queuedMessage ) ; for ( const consumer of this . consumers ) { consumer . emit ( queuedMessage ) ; } } public getLastMessage ( ) { return this . getMessage ( this . messages . length - 1 ) ; } public getMessage ( index ) { return this . messages [ index ] . value as core . ISequencedOperationMessage ; } }