import { IClientJoin , IDocumentMessage , IDocumentSystemMessage , ISequencedDocumentAugmentedMessage , ISequencedDocumentMessage , MessageType , ScopeType , } from "@fluidframework/protocol-definitions" ; import { BoxcarType , IBoxcarMessage , IQueuedMessage , IRawOperationMessage , ISequencedOperationMessage , RawOperationType , SequencedOperationType , } from "@fluidframework/server-services-core" ; import hash = require ( "string-hash" ) ; export class KafkaMessageFactory { private readonly offsets : number [ ] = [ ] ; constructor ( public topic = "test" , partitions = 1 , private readonly stringify = true , private readonly tenantId = null , private readonly documentId = null ) { for ( let i = 0 ; i < partitions ; i ++ ) { this . offsets . push ( 0 ) ; } } public sequenceMessage ( value , key ) { const partition = this . getPartition ( key ) ; const offset = this . offsets [ partition ] ++ ; const message = { offset , partition , topic : this . topic , value : this . stringify ? JSON . stringify ( value ) : ( { contents : [ value ] , documentId : this . documentId , tenantId : this . tenantId , type : BoxcarType } as IBoxcarMessage ) } ; return message ; } public getHeadOffset ( key ) { return this . offsets [ this . getPartition ( key ) ] - 1 ; } private getPartition ( key ) { return hash ( key ) % this . offsets . length ; } } export class MessageFactory { private clientSequenceNumber = 0 ; private sequenceNumber = 0 ; constructor ( private readonly documentId , private readonly clientId , private readonly tenantId = "test" ) { } public createDocumentMessage ( type = MessageType . Operation , referenceSequenceNumber = 0 ) { const operation = { clientSequenceNumber : ++ this . clientSequenceNumber , contents : null , metadata : undefined , referenceSequenceNumber , traces : [ ] , type } ; return operation ; } public create ( type = MessageType . Operation , referenceSequenceNumber = 0 , timestamp = Date . now ( ) ) { const operation = this . createDocumentMessage ( type , referenceSequenceNumber ) ; return this . createRawOperation ( operation , timestamp , this . clientId ) ; } public createJoin ( timestamp = Date . now ( ) , serverMetadata = undefined ) { const joinMessage = { clientId : this . clientId , detail : { mode : "write" , permission : [ ] , scopes : [ ScopeType . DocRead , ScopeType . DocWrite , ScopeType . SummaryWrite ] , details : { capabilities : { interactive : true } } , user : null } } ; const operation = { clientSequenceNumber : - 1 , contents : null , data : JSON . stringify ( joinMessage ) , referenceSequenceNumber : - 1 , traces : [ ] , type : MessageType . ClientJoin , serverMetadata } ; return this . createRawOperation ( operation , timestamp , null ) ; } public createLeave ( timestamp = Date . now ( ) ) { const operation = { clientSequenceNumber : - 1 , contents : null , data : JSON . stringify ( this . clientId ) , referenceSequenceNumber : - 1 , traces : [ ] , type : MessageType . ClientLeave } ; return this . createRawOperation ( operation , timestamp , null ) ; } public createRawOperation ( operation , timestamp , clientId ) { const objectMessage = { clientId , documentId : this . documentId , operation , tenantId : this . tenantId , timestamp , type : RawOperationType } ; return objectMessage ; } public createSave ( ) { const operation = { clientSequenceNumber : this . clientSequenceNumber ++ , contents : "Test Save" , referenceSequenceNumber : 0 , traces : [ ] , type : MessageType . Save } ; const sequencedOperation = { clientId : this . clientId , clientSequenceNumber : operation . clientSequenceNumber , contents : operation . contents , metadata : operation . metadata , minimumSequenceNumber : 0 , origin : undefined , referenceSequenceNumber : operation . referenceSequenceNumber , sequenceNumber : this . sequenceNumber ++ , term : 1 , timestamp : Date . now ( ) , traces : [ ] , type : operation . type } ; const message = { documentId : this . documentId , operation : sequencedOperation , tenantId : this . tenantId , type : SequencedOperationType } ; return message ; } public createSequencedOperation ( referenceSequenceNumber = 0 ) { const operation = this . createDocumentMessage ( MessageType . Operation , referenceSequenceNumber ) ; const sequencedOperation = { clientId : this . clientId , clientSequenceNumber : operation . clientSequenceNumber , contents : operation . contents , metadata : operation . metadata , minimumSequenceNumber : 0 , origin : undefined , referenceSequenceNumber : operation . referenceSequenceNumber , sequenceNumber : this . sequenceNumber ++ , term : 1 , timestamp : Date . now ( ) , traces : [ ] , type : operation . type } ; const message = { documentId : this . documentId , operation : sequencedOperation , tenantId : this . tenantId , type : SequencedOperationType } ; return message ; } public createSummarize ( referenceSequenceNumber , handle ) { const operation = this . createDocumentMessage ( MessageType . Summarize , referenceSequenceNumber ) ; const sequencedOperation = { clientId : this . clientId , clientSequenceNumber : operation . clientSequenceNumber , contents : `{"handle": "${ handle }" ,"head":null,"message":"","parents":[]}` , metadata : operation . metadata , minimumSequenceNumber : 0 , origin : undefined , referenceSequenceNumber : operation . referenceSequenceNumber , sequenceNumber : this . sequenceNumber ++ , term : 1 , timestamp : Date . now ( ) , traces : [ ] , type : operation . type , additionalContent : "" } ; const message = { documentId : this . documentId , operation : sequencedOperation , tenantId : this . tenantId , type : SequencedOperationType } ; return message ; } public createNoClient ( referenceSequenceNumber = 0 ) { const operation = this . createDocumentMessage ( MessageType . NoClient , referenceSequenceNumber ) ; const sequencedOperation = { clientId : this . clientId , clientSequenceNumber : operation . clientSequenceNumber , contents : operation . contents , metadata : operation . metadata , minimumSequenceNumber : this . sequenceNumber , origin : undefined , referenceSequenceNumber : this . sequenceNumber , sequenceNumber : this . sequenceNumber ++ , term : 1 , timestamp : Date . now ( ) , traces : [ ] , type : operation . type , additionalContent : "" } ; const message = { documentId : this . documentId , operation : sequencedOperation , tenantId : this . tenantId , type : SequencedOperationType } ; return message ; } public createSummaryAck ( handle ) { const operation = this . createDocumentMessage ( MessageType . SummaryAck , 0 ) ; const sequencedOperation = { clientId : null , clientSequenceNumber : - 1 , contents : { handle , summaryProposal : { summarySequenceNumber : 1 } } , metadata : operation . metadata , minimumSequenceNumber : 0 , origin : undefined , referenceSequenceNumber : - 1 , sequenceNumber : this . sequenceNumber ++ , term : 1 , timestamp : Date . now ( ) , traces : [ ] , type : operation . type , additionalContent : "" } ; const message = { documentId : this . documentId , operation : sequencedOperation , tenantId : this . tenantId , type : SequencedOperationType } ; return message ; } }