import { assert , bufferToString , unreachableCase } from "@fluidframework/common-utils" ; import { ISequencedDocumentMessage , MessageType } from "@fluidframework/protocol-definitions" ; import { IChannelAttributes , IFluidDataStoreRuntime , IChannelStorageService , } from "@fluidframework/datastore-definitions" ; import { IFluidSerializer , SharedObject } from "@fluidframework/shared-object-base" ; import { ISummaryTreeWithStats } from "@fluidframework/runtime-definitions" ; import { SummaryTreeBuilder } from "@fluidframework/runtime-utils" ; import { v4 as uuid } from "uuid" ; import { ConsensusCallback , ConsensusResult , IConsensusOrderedCollection , IOrderedCollection , IConsensusOrderedCollectionEvents , } from "./interfaces" ; const snapshotFileNameData = "header" ; const snapshotFileNameTracking = "jobTracking" ; interface IConsensusOrderedCollectionValue < T > { readonly acquireId : string ; readonly value : T ; } /**
 * An operation for consensus ordered collection
 */ interface IConsensusOrderedCollectionAddOperation { opName : "add" ; value : string ; } interface IConsensusOrderedCollectionAcquireOperation { opName : "acquire" ; acquireId : string ; } interface IConsensusOrderedCollectionCompleteOperation { opName : "complete" ; acquireId : string ; } interface IConsensusOrderedCollectionReleaseOperation { opName : "release" ; acquireId : string ; } type IConsensusOrderedCollectionOperation = IConsensusOrderedCollectionAddOperation | IConsensusOrderedCollectionAcquireOperation | IConsensusOrderedCollectionCompleteOperation | IConsensusOrderedCollectionReleaseOperation ; /** The type of the resolve function to call after the local operation is ack'd */ type PendingResolve < T > = ( value : IConsensusOrderedCollectionValue < T > | undefined ) => void ; /**
 * For job tracking, we need to keep track of which client "owns" a given value.
 * Key is the acquireId from when it was acquired
 * Value is the acquired value, and the id of the client who acquired it, or undefined for unattached client
 */ type JobTrackingInfo < T > = Map < string , { value : T ; clientId : string | undefined ; } > ; const idForLocalUnattachedClient = undefined ; /**
 * Implementation of a consensus collection shared object
 *
 * Implements the shared object's communication, and the semantics around the
 * release/complete mechanism following acquire.
 *
 * Generally not used directly. A derived type will pass in a backing data type
 * IOrderedCollection that will define the deterministic add/acquire order and snapshot ability.
 */ export class ConsensusOrderedCollection < T = any > extends SharedObject < IConsensusOrderedCollectionEvents < T > > implements IConsensusOrderedCollection < T > { /**
     * The set of values that have been acquired but not yet completed or released
     */ private jobTracking : JobTrackingInfo < T > = new Map ( ) ; /**
     * Constructs a new consensus collection. If the object is non-local an id and service interfaces will
     * be provided
     */ protected constructor ( id , runtime , attributes , private readonly data ) { super ( id , runtime , attributes ) ; runtime . getQuorum ( ) . on ( "removeMember" , ( clientId ) => { assert ( ! ! clientId , 0x067 ) ; this . removeClient ( clientId ) ; } ) ; } /**
     * Add a value to the consensus collection.
     */ public async add ( value ) { const valueSer = this . serializeValue ( value , this . serializer ) ; if ( ! this . isAttached ( ) ) { const addValue = this . deserializeValue ( valueSer , this . serializer ) as T ; this . addCore ( addValue ) ; return ; } await this . submit < IConsensusOrderedCollectionAddOperation > ( { opName : "add" , value : valueSer } ) ; } /**
     * Remove a value from the consensus collection.  If the collection is empty, returns false.
     * Otherwise calls callback with the value
     */ public async acquire ( callback ) { const result = await this . acquireInternal ( ) ; if ( result === undefined ) { return false ; } const res = await callback ( result . value ) ; switch ( res ) { case ConsensusResult . Complete : await this . complete ( result . acquireId ) ; break ; case ConsensusResult . Release : this . release ( result . acquireId ) ; this . emit ( "localRelease" , result . value , true ) ; break ; default : unreachableCase ( res ) ; } return true ; } /**
     * Wait for a value to be available and acquire it from the consensus collection
     */ public async waitAndAcquire ( callback ) { do { if ( this . data . size ( ) === 0 ) { await this . newAckBasedPromise < T > ( ( resolve ) => { this . once ( "add" , resolve ) ; } ) ; } } while ( ! ( await this . acquire ( callback ) ) ) ; } protected summarizeCore ( serializer ) { this . removeClient ( idForLocalUnattachedClient ) ; const builder = new SummaryTreeBuilder ( ) ; let blobContent = this . serializeValue ( this . data . asArray ( ) , serializer ) ; builder . addBlob ( snapshotFileNameData , blobContent ) ; blobContent = this . serializeValue ( Array . from ( this . jobTracking . entries ( ) ) , serializer ) ; builder . addBlob ( snapshotFileNameTracking , blobContent ) ; return builder . getSummaryTree ( ) ; } protected isActive ( ) { return this . runtime . connected && this . runtime . deltaManager . active ; } protected async complete ( acquireId ) { if ( ! this . isAttached ( ) ) { this . completeCore ( acquireId ) ; return ; } if ( this . isActive ( ) ) { await this . submit < IConsensusOrderedCollectionCompleteOperation > ( { opName : "complete" , acquireId } ) ; } } protected completeCore ( acquireId ) { const rec = this . jobTracking . get ( acquireId ) ; if ( rec !== undefined ) { this . jobTracking . delete ( acquireId ) ; this . emit ( "complete" , rec . value ) ; } } protected release ( acquireId ) { if ( ! this . isAttached ( ) ) { this . releaseCore ( acquireId ) ; return ; } if ( this . isActive ( ) ) { this . submit < IConsensusOrderedCollectionReleaseOperation > ( { opName : "release" , acquireId } ) . catch ( ( error ) => { this . runtime . logger . sendErrorEvent ( { eventName : "ConsensusQueue_release" } , error ) ; } ) ; } } protected releaseCore ( acquireId ) { const rec = this . jobTracking . get ( acquireId ) ; if ( rec !== undefined ) { this . jobTracking . delete ( acquireId ) ; this . data . add ( rec . value ) ; this . emit ( "add" , rec . value , false ) ; } } /**
     * {@inheritDoc @fluidframework/shared-object-base#SharedObject.loadCore}
     */ protected async loadCore ( storage ) { assert ( this . jobTracking . size === 0 , 0x068 ) ; const blob = await storage . readBlob ( snapshotFileNameTracking ) ; const rawContentTracking = bufferToString ( blob , "utf8" ) ; const content = this . deserializeValue ( rawContentTracking , this . serializer ) ; this . jobTracking = new Map ( content ) as JobTrackingInfo < T > ; assert ( this . data . size ( ) === 0 , 0x069 ) ; const blob2 = await storage . readBlob ( snapshotFileNameData ) ; const rawContentData = bufferToString ( blob2 , "utf8" ) ; const content2 = this . deserializeValue ( rawContentData , this . serializer ) as T [ ] ; this . data . loadFrom ( content2 ) ; } protected registerCore ( ) { return ; } protected onDisconnect ( ) { for ( const [ , { value , clientId } ] of this . jobTracking ) { if ( clientId === this . runtime . clientId ) { this . emit ( "localRelease" , value , false ) ; } } } protected processCore ( message , local , localOpMetadata ) { if ( message . type === MessageType . Operation ) { const op = message . contents ; let value ; switch ( op . opName ) { case "add" : this . addCore ( this . deserializeValue ( op . value , this . serializer ) as T ) ; break ; case "acquire" : value = this . acquireCore ( op . acquireId , message . clientId ) ; break ; case "complete" : this . completeCore ( op . acquireId ) ; break ; case "release" : this . releaseCore ( op . acquireId ) ; break ; default : unreachableCase ( op ) ; } if ( local ) { const resolve = localOpMetadata as PendingResolve < T > ; resolve ( value ) ; } } } private async submit < TMessage extends IConsensusOrderedCollectionOperation > ( message ) { assert ( this . isAttached ( ) , 0x06a ) ; return this . newAckBasedPromise < IConsensusOrderedCollectionValue < T > | undefined > ( ( resolve ) => { this . submitLocalMessage ( message , resolve ) ; } ) . catch ( ( error ) => undefined ) ; } private addCore ( value ) { this . data . add ( value ) ; this . emit ( "add" , value , true ) ; } private acquireCore ( acquireId , clientId ? ) { if ( this . data . size ( ) === 0 ) { return undefined ; } const value = this . data . remove ( ) ; const value2 = { acquireId , value } ; this . jobTracking . set ( value2 . acquireId , { value , clientId } ) ; this . emit ( "acquire" , value , clientId ) ; return value2 ; } private async acquireInternal ( ) { if ( ! this . isAttached ( ) ) { return this . acquireCore ( uuid ( ) , idForLocalUnattachedClient ) ; } return this . submit < IConsensusOrderedCollectionAcquireOperation > ( { opName : "acquire" , acquireId : uuid ( ) } ) ; } private removeClient ( clientIdToRemove ? ) { const added = [ ] ; for ( const [ acquireId , { value , clientId } ] of this . jobTracking ) { if ( clientId === clientIdToRemove ) { this . jobTracking . delete ( acquireId ) ; this . data . add ( value ) ; added . push ( value ) ; } } added . map ( ( value ) => this . emit ( "add" , value , false ) ) ; } private serializeValue ( value , serializer ) { return serializer . stringify ( value , this . handle ) ; } private deserializeValue ( content , serializer ) { return serializer . parse ( content ) ; } protected applyStashedOp ( ) { throw new Error ( "not implemented" ) ; } }