import { ITelemetryLogger , IEvent } from "@fluidframework/common-definitions" ; import { assert , performance , Deferred , TypedEventEmitter } from "@fluidframework/common-utils" ; import { DocumentDeltaConnection } from "@fluidframework/driver-base" ; import { DriverError } from "@fluidframework/driver-definitions" ; import { OdspError } from "@fluidframework/odsp-driver-definitions" ; import { loggerToMonitoringContext , LoggingError } from "@fluidframework/telemetry-utils" ; import { IClient , IConnect , INack , ISequencedDocumentMessage , ISignalMessage , } from "@fluidframework/protocol-definitions" ; import { v4 as uuid } from "uuid" ; import { IOdspSocketError , IGetOpsResponse , IFlushOpsResponse } from "./contracts" ; import { EpochTracker } from "./epochTracker" ; import { errorObjectFromSocketError } from "./odspError" ; import { pkgVersion } from "./packageVersion" ; const protocolVersions = [ "^0.4.0" , "^0.3.0" , "^0.2.0" , "^0.1.0" ] ; const feature_get_ops = "api_get_ops" ; const feature_flush_ops = "api_flush_ops" ; export interface FlushResult { lastPersistedSequenceNumber ? : number ; retryAfter ? : number ; } const socketReferenceBufferTime = 2000 ; export interface ISocketEvents extends IEvent { ( event , listener ) ; } class SocketReference extends TypedEventEmitter < ISocketEvents > { private references : number = 1 ; private delayDeleteTimeout : ReturnType < typeof setTimeout > | undefined ; private _socket : SocketIOClient . Socket | undefined ; private isPendingInitialConnection = true ; private static readonly socketIoSockets : Map < string , SocketReference > = new Map ( ) ; public static find ( key , logger ) { const socketReference = SocketReference . socketIoSockets . get ( key ) ; if ( socketReference && socketReference . disconnected ) { socketReference . closeSocket ( ) ; return undefined ; } if ( socketReference ) { socketReference . clearTimer ( ) ; socketReference . references ++ ; } return socketReference ; } param key param isFatalError public removeSocketIoReference ( isFatalError ) { assert ( this . references > 0 , 0x09f ) ; this . references -- ; this . isPendingInitialConnection = false ; if ( isFatalError || this . disconnected ) { this . closeSocket ( ) ; return ; } if ( this . references === 0 && this . delayDeleteTimeout === undefined ) { this . delayDeleteTimeout = setTimeout ( ( ) => { assert ( this . references === 0 , 0x0a0 ) ; this . closeSocket ( ) ; } , socketReferenceBufferTime ) ; } } public get socket ( ) { if ( ! this . _socket ) { throw new Error ( `Invalid socket for key "${ this . key }` ) ; } return this . _socket ; } public constructor ( public readonly key , socket ) { super ( ) ; this . _socket = socket ; assert ( ! SocketReference . socketIoSockets . has ( key ) , 0x220 ) ; SocketReference . socketIoSockets . set ( key , this ) ; socket . on ( "server_disconnect" , ( socketError ) => { const error = errorObjectFromSocketError ( socketError , "server_disconnect" ) ; error . canRetry = true ; this . isPendingInitialConnection = false ; this . emit ( "server_disconnect" , error ) ; this . closeSocket ( ) ; } ) ; } private clearTimer ( ) { if ( this . delayDeleteTimeout !== undefined ) { clearTimeout ( this . delayDeleteTimeout ) ; this . delayDeleteTimeout = undefined ; } } private closeSocket ( ) { if ( ! this . _socket ) { return ; } this . clearTimer ( ) ; assert ( SocketReference . socketIoSockets . get ( this . key ) === this , 0x0a1 ) ; SocketReference . socketIoSockets . delete ( this . key ) ; const socket = this . _socket ; this . _socket = undefined ; Promise . resolve ( ) . then ( ( ) => { socket . disconnect ( ) ; } ) ; } private get disconnected ( ) { if ( this . _socket === undefined ) { return true ; } if ( this . socket . connected ) { return false ; } return ! this . isPendingInitialConnection ; } } /**
 * Represents a connection to a stream of delta updates
 */ export class OdspDocumentDeltaConnection extends DocumentDeltaConnection { param tenantId param documentId param token param io param client param mode param url param telemetryLogger param timeoutMs param epochTracker param socketReferenceKeyPrefix public static async create ( tenantId , documentId , token , io , client , url , telemetryLogger , timeoutMs , epochTracker , socketReferenceKeyPrefix ) { const mc = loggerToMonitoringContext ( telemetryLogger ) ; const parsedUrl = new URL ( url ) ; const enableMultiplexing = ! parsedUrl . searchParams . has ( "documentId" ) && ! parsedUrl . searchParams . has ( "tenantId" ) ; const key = socketReferenceKeyPrefix ? `${ socketReferenceKeyPrefix },${ url }` : url ; const socketReferenceKey = enableMultiplexing ? key : `${ key },${ tenantId },${ documentId }` ; const socketReference = OdspDocumentDeltaConnection . getOrCreateSocketIoReference ( io , timeoutMs , socketReferenceKey , url , enableMultiplexing , tenantId , documentId , telemetryLogger ) ; const socket = socketReference . socket ; const connectMessage = { client , id : documentId , mode : client . mode , tenantId , token , versions : protocolVersions , nonce : uuid ( ) , epoch : epochTracker . fluidEpoch , relayUserAgent : [ client . details . environment , ` driverVersion:${ pkgVersion }` ] . join ( ";" ) } ; connectMessage . supportedFeatures = { } ; if ( mc . config . getBoolean ( "Fluid.Driver.Odsp.GetOpsEnabled" ) !== false ) { connectMessage . supportedFeatures [ feature_get_ops ] = true ; } const deltaConnection = new OdspDocumentDeltaConnection ( socket , documentId , socketReference , telemetryLogger , enableMultiplexing ) ; try { await deltaConnection . initialize ( connectMessage , timeoutMs ) ; await epochTracker . validateEpochFromPush ( deltaConnection . details ) ; } catch ( errorObject ) { if ( errorObject !== null && typeof errorObject === "object" ) { if ( errorObject . statusCode === 400 || errorObject . statusCode === 404 ) { errorObject . canRetry = true ; } } throw errorObject ; } return deltaConnection ; } private socketReference : SocketReference | undefined ; private readonly requestOpsNoncePrefix : string ; private pushCallCounter = 0 ; private readonly getOpsMap : Map < string , { start : number ; from : number ; to : number ; } > = new Map ( ) ; private flushOpNonce : string | undefined ; private flushDeferred : Deferred < FlushResult > | undefined ; /**
     * Error raising for socket.io issues
     */ protected createErrorObject ( handler , error ? , canRetry = true ) { if ( canRetry && Number . isInteger ( error ?. code ) && typeof error ?. message === "string" ) { return errorObjectFromSocketError ( error as IOdspSocketError , handler ) as DriverError ; } else { return super . createErrorObject ( handler , error , canRetry ) ; } } /**
     * Gets or create a socket io connection for the given key
     */ private static getOrCreateSocketIoReference ( io , timeoutMs , key , url , enableMultiplexing , tenantId , documentId , logger ) { const existingSocketReference = SocketReference . find ( key , logger ) ; if ( existingSocketReference ) { return existingSocketReference ; } const query = enableMultiplexing ? undefined : { documentId , tenantId } ; const socket = io ( url , { multiplex : false , query , reconnection : false , transports : [ "websocket" ] , timeout : timeoutMs } ) ; return new SocketReference ( key , socket ) ; } param socket param documentId param details param socketReferenceKey param enableMultiplexing private constructor ( socket , documentId , socketReference , logger , private readonly enableMultiplexing ? ) { super ( socket , documentId , logger ) ; this . socketReference = socketReference ; this . requestOpsNoncePrefix = `${ uuid ( ) }-` ; } param from param to returns public requestOps ( from , to ) { assert ( to > from , 0x272 ) ; if ( ( this . details as any ) . supportedFeatures ?. [ feature_get_ops ] !== true ) { return ; } this . pushCallCounter ++ ; const nonce = `${ this . requestOpsNoncePrefix }${ this . pushCallCounter }` ; const start = performance . now ( ) ; if ( this . getOpsMap . size >= 5 ) { let time = start ; let key ; for ( const [ keyCandidate , value ] of this . getOpsMap . entries ( ) ) { if ( value . start <= time || key === undefined ) { time = value . start ; key = keyCandidate ; } } const payloadToDelete = this . getOpsMap . get ( key ! ) ! ; this . logger . sendErrorEvent ( { eventName : "GetOpsTooMany" , nonce , from : payloadToDelete . from , to : payloadToDelete . to , length : payloadToDelete . to - payloadToDelete . from , duration : performance . now ( ) - payloadToDelete . start } ) ; this . getOpsMap . delete ( key ! ) ; } this . getOpsMap . set ( nonce , { start , from , to } ) ; this . socket . emit ( "get_ops" , this . clientId , { nonce , from , to : to - 1 } ) ; } public async flush ( ) { if ( ( this . details as any ) . supportedFeatures ?. [ feature_flush_ops ] !== true ) { this . logger . sendErrorEvent ( { eventName : "FlushOpsNotSupported" } ) ; throw new Error ( "flush() API is not supported by PUSH, required for single-commit summaries" ) ; } this . pushCallCounter ++ ; const nonce = `${ this . requestOpsNoncePrefix }${ this . pushCallCounter }` ; if ( this . flushDeferred !== undefined ) { this . logger . sendErrorEvent ( { eventName : "FlushOpsTooMany" } ) ; this . flushDeferred . reject ( "process involving flush() was cancelled OR unsupported concurrency" ) ; } this . socket . emit ( "flush_ops" , this . clientId , { nonce } ) ; this . flushOpNonce = nonce ; this . flushDeferred = new Deferred < FlushResult > ( ) ; return this . flushDeferred . promise ; } protected serverDisconnectHandler = ( error ) => { this . disposeCore ( true , error ) ; } ; protected async initialize ( connectMessage , timeout ) { if ( this . enableMultiplexing ) { this . earlyOpHandler = ( messageDocumentId , msgs ) => { if ( this . documentId === messageDocumentId ) { this . queuedMessages . push ( ... msgs ) ; } } ; this . earlySignalHandler = ( msg , messageDocumentId ? ) => { if ( messageDocumentId === undefined || messageDocumentId === this . documentId ) { this . queuedSignals . push ( msg ) ; } } ; } this . socketReference ! . once ( "server_disconnect" , this . serverDisconnectHandler ) ; this . socket . on ( "get_ops_response" , ( result ) => { const messages = result . messages ; const data = this . getOpsMap . get ( result . nonce ) ; if ( data !== undefined || result . nonce . indexOf ( this . requestOpsNoncePrefix ) === 0 ) { this . getOpsMap . delete ( result . nonce ) ; const common = { eventName : "GetOps" , nonce : data === undefined ? result . nonce : undefined , code : result . code , from : data ?. from , to : data ?. to , duration : data === undefined ? undefined : performance . now ( ) - data . start } ; if ( messages !== undefined && messages . length > 0 ) { this . logger . sendPerformanceEvent ( { ... common , first : messages [ 0 ] . sequenceNumber , last : messages [ messages . length - 1 ] . sequenceNumber , length : messages . length } ) ; this . emit ( "op" , this . documentId , messages ) ; } else { this . logger . sendPerformanceEvent ( { ... common , length : 0 } ) ; } } } ) ; this . socket . on ( "flush_ops_response" , ( result ) => { if ( this . flushOpNonce === result . nonce ) { const seq = result . lastPersistedSequenceNumber ; let category = "generic" ; if ( result . lastPersistedSequenceNumber === undefined || result . code !== 200 ) { switch ( result . code ) { case 409 : case 429 : category = "error" ; break ; case 204 : break ; default : category = "error" ; break ; } } this . logger . sendTelemetryEvent ( { eventName : "FlushResult" , code : result . code , sequenceNumber : seq , category } ) ; this . flushDeferred ! . resolve ( result ) ; this . flushDeferred = undefined ; this . flushOpNonce = undefined ; } } ) ; await super . initialize ( connectMessage , timeout ) ; } protected addTrackedListener ( event , listener ) { switch ( event ) { case "op" : super . addTrackedListener ( event , ( documentId , msgs ) => { if ( ! this . enableMultiplexing || this . documentId === documentId ) { listener ( documentId , msgs ) ; } } ) ; break ; case "signal" : super . addTrackedListener ( event , ( msg , documentId ? ) => { if ( ! this . enableMultiplexing || ! documentId || documentId === this . documentId ) { listener ( msg , documentId ) ; } } ) ; break ; case "nack" : super . addTrackedListener ( event , ( clientIdOrDocumentId , message ) => { if ( clientIdOrDocumentId . length === 0 || clientIdOrDocumentId === this . documentId || ( this . hasDetails && clientIdOrDocumentId === this . clientId ) ) { this . emit ( "nack" , clientIdOrDocumentId , message ) ; } } ) ; break ; default : super . addTrackedListener ( event , listener ) ; break ; } } /**
     * Disconnect from the websocket
     */ protected disconnect ( socketProtocolError , reason ) { const socket = this . socketReference ; assert ( socket !== undefined , 0x0a2 ) ; this . socketReference = undefined ; this . socket . off ( "server_disconnect" , this . serverDisconnectHandler ) ; if ( ! socketProtocolError && this . hasDetails ) { this . socket . emit ( "disconnect_document" , this . clientId , this . documentId ) ; } socket . removeSocketIoReference ( socketProtocolError ) ; this . emit ( "disconnect" , reason ) ; } }