import { IClient , IClientJoin , IDocumentMessage , IDocumentSystemMessage , MessageType , } from "@fluidframework/protocol-definitions" ; import * as core from "@fluidframework/server-services-core" ; export class KafkaOrdererConnection implements core . IOrdererConnection { public static async create ( producer , tenantId , documentId , client , maxMessageSize , clientId , serviceConfiguration ) { return new KafkaOrdererConnection ( producer , tenantId , documentId , clientId , client , maxMessageSize , serviceConfiguration ) ; } constructor ( private readonly producer , public readonly tenantId , public readonly documentId , public readonly clientId , private readonly client , public readonly maxMessageSize , public readonly serviceConfiguration ) { } /**
     * Sends the client join op for this connection
     */ public async connect ( clientJoinMessageServerMetadata ? ) { const clientDetail = { clientId : this . clientId , detail : this . client } ; const operation = { clientSequenceNumber : - 1 , contents : null , data : JSON . stringify ( clientDetail ) , referenceSequenceNumber : - 1 , traces : this . serviceConfiguration . enableTraces ? [ ] : undefined , type : MessageType . ClientJoin , serverMetadata : clientJoinMessageServerMetadata } ; const message = { clientId : null , documentId : this . documentId , operation , tenantId : this . tenantId , timestamp : Date . now ( ) , type : core . RawOperationType } ; return this . submitRawOperation ( [ message ] ) ; } /**
     * Orders the provided list of messages. The messages in the array are guaranteed to be ordered sequentially
     * so long as their total size fits under the maxMessageSize.
     */ public async order ( messages ) { const rawMessages = messages . map ( ( message ) => { const rawMessage = { clientId : this . clientId , documentId : this . documentId , operation : message , tenantId : this . tenantId , timestamp : Date . now ( ) , type : core . RawOperationType } ; return rawMessage ; } ) ; return this . submitRawOperation ( rawMessages ) ; } /**
     * Sends the client leave op for this connection
     */ public async disconnect ( clientLeaveMessageServerMetadata ? ) { const operation = { clientSequenceNumber : - 1 , contents : null , data : JSON . stringify ( this . clientId ) , referenceSequenceNumber : - 1 , traces : this . serviceConfiguration . enableTraces ? [ ] : undefined , type : MessageType . ClientLeave , serverMetadata : clientLeaveMessageServerMetadata } ; const message = { clientId : null , documentId : this . documentId , operation , tenantId : this . tenantId , timestamp : Date . now ( ) , type : core . RawOperationType } ; return this . submitRawOperation ( [ message ] ) ; } public once ( event , listener ) { this . producer . once ( event , listener ) ; } private async submitRawOperation ( messages ) { if ( this . serviceConfiguration . enableTraces ) { messages . forEach ( ( message ) => { const operation = message . operation ; if ( operation && operation . traces === undefined ) { operation . traces = [ ] ; } else if ( operation && operation . traces && operation . traces . length > 1 ) { operation . traces . push ( { action : "end" , service : "alfred" , timestamp : Date . now ( ) } ) ; } } ) ; } return this . producer . send ( messages , this . tenantId , this . documentId ) ; } } export class KafkaOrderer implements core . IOrderer { public static async create ( producer , tenantId , documentId , maxMessageSize , serviceConfiguration ) { return new KafkaOrderer ( producer , tenantId , documentId , maxMessageSize , serviceConfiguration ) ; } constructor ( private readonly producer , private readonly tenantId , private readonly documentId , private readonly maxMessageSize , private readonly serviceConfiguration ) { } public async connect ( socket , clientId , client ) { const connection = KafkaOrdererConnection . create ( this . producer , this . tenantId , this . documentId , client , this . maxMessageSize , clientId , this . serviceConfiguration ) ; return connection ; } public close ( ) { return Promise . resolve ( ) ; } } export class KafkaOrdererFactory { private readonly ordererMap = new Map < string , Promise < core . IOrderer > > ( ) ; constructor ( private readonly producer , private readonly maxMessageSize , private readonly serviceConfiguration ) { } public async create ( tenantId , documentId ) { const fullId = `${ tenantId }/${ documentId }` ; let orderer = this . ordererMap . get ( fullId ) ; if ( orderer === undefined ) { orderer = KafkaOrderer . create ( this . producer , tenantId , documentId , this . maxMessageSize , this . serviceConfiguration ) ; this . ordererMap . set ( fullId , orderer ) ; } return orderer ; } public delete ( tenantId , documentId ) { this . ordererMap . delete ( `${ tenantId }/${ documentId }` ) ; } }