import cloneDeep from "lodash/cloneDeep" ; import { IDocumentStorageService } from "@fluidframework/driver-definitions" ; import { ISequencedDocumentMessage , ISnapshotTree } from "@fluidframework/protocol-definitions" ; import { IChannel , IFluidDataStoreRuntime , IChannelFactory , IChannelAttributes , } from "@fluidframework/datastore-definitions" ; import { IFluidDataStoreContext , IGarbageCollectionData , ISummarizeResult , } from "@fluidframework/runtime-definitions" ; import { readAndParse } from "@fluidframework/driver-utils" ; import { CreateProcessingError } from "@fluidframework/container-utils" ; import { assert , Lazy } from "@fluidframework/common-utils" ; import { IFluidHandle } from "@fluidframework/core-interfaces" ; import { createServiceEndpoints , IChannelContext , summarizeChannel , summarizeChannelAsync , } from "./channelContext" ; import { ChannelDeltaConnection } from "./channelDeltaConnection" ; import { ISharedObjectRegistry } from "./dataStoreRuntime" ; import { ChannelStorageService } from "./channelStorageService" ; /**
 * Channel context for a locally created channel
 */ export abstract class LocalChannelContextBase implements IChannelContext { public channel : IChannel | undefined ; private attached = false ; protected readonly pending : ISequencedDocumentMessage [ ] = [ ] ; protected factory : IChannelFactory | undefined ; constructor ( protected readonly id , protected readonly registry , protected readonly runtime , private readonly servicesGetter ) { } public async getChannel ( ) { assert ( this . channel !== undefined , 0x207 ) ; return this . channel ; } public get isLoaded ( ) : boolean { return this . channel !== undefined ; } public setConnectionState ( connected , clientId ? ) { if ( this . attached && this . isLoaded ) { this . servicesGetter ( ) . value . deltaConnection . setConnectionState ( connected ) ; } } public processOp ( message , local , localOpMetadata ) { assert ( this . attached , 0x188 ) ; if ( this . isLoaded ) { this . servicesGetter ( ) . value . deltaConnection . process ( message , local , localOpMetadata ) ; } else { assert ( local === false , 0x189 ) ; this . pending . push ( message ) ; } } public reSubmit ( content , localOpMetadata ) { assert ( this . isLoaded , 0x18a ) ; assert ( this . attached , 0x18b ) ; this . servicesGetter ( ) . value . deltaConnection . reSubmit ( content , localOpMetadata ) ; } public applyStashedOp ( ) { throw new Error ( "no stashed ops on local channel" ) ; } param fullTree param trackState public async summarize ( fullTree = false , trackState = false ) { assert ( this . isLoaded && this . channel !== undefined , 0x18c ) ; return summarizeChannelAsync ( this . channel , fullTree , trackState ) ; } public getAttachSummary ( ) { assert ( this . isLoaded && this . channel !== undefined , 0x18d ) ; return summarizeChannel ( this . channel , true , false ) ; } public markAttached ( ) { if ( this . attached ) { throw new Error ( "Channel is already attached" ) ; } if ( this . isLoaded ) { assert ( ! ! this . channel , 0x192 ) ; this . channel . connect ( this . servicesGetter ( ) . value ) ; } this . attached = true ; } param fullGC public async getGCData ( fullGC = false ) { assert ( this . isLoaded && this . channel !== undefined , 0x193 ) ; return this . channel . getGCData ( fullGC ) ; } public updateUsedRoutes ( usedRoutes , gcTimestamp ? ) { } } export class RehydratedLocalChannelContext extends LocalChannelContextBase { private readonly services : Lazy < { readonly deltaConnection : ChannelDeltaConnection ; readonly objectStorage : ChannelStorageService ; } > ; private readonly dirtyFn : ( ) => void ; constructor ( id , registry , runtime , dataStoreContext , storageService , submitFn , dirtyFn , addedGCOutboundReferenceFn , private readonly snapshotTree ) { super ( id , registry , runtime , ( ) => this . services ) ; const blobMap = new Map < string , ArrayBufferLike > ( ) ; const clonedSnapshotTree = cloneDeep ( this . snapshotTree ) ; if ( this . isSnapshotInOldFormatAndCollectBlobs ( clonedSnapshotTree , blobMap ) ) { this . sanitizeSnapshot ( clonedSnapshotTree ) ; } this . services = new Lazy ( ( ) => { return createServiceEndpoints ( this . id , dataStoreContext . connected , submitFn , this . dirtyFn , addedGCOutboundReferenceFn , storageService , clonedSnapshotTree , blobMap ) ; } ) ; this . dirtyFn = ( ) => { dirtyFn ( id ) ; } ; } public async getChannel ( ) { if ( this . channel === undefined ) { this . channel = await this . loadChannel ( ) . catch ( ( err ) => { throw CreateProcessingError ( err , "rehydratedLocalChannelContextFailedToLoadChannel" , undefined ) ; } ) ; } return this . channel ; } private async loadChannel ( ) { assert ( ! this . isLoaded , 0x18e ) ; assert ( await this . services . value . objectStorage . contains ( ".attributes" ) , 0x190 ) ; const attributes = await readAndParse < IChannelAttributes > ( this . services . value . objectStorage , ".attributes" ) ; assert ( this . factory === undefined , 0x208 ) ; this . factory = this . registry . get ( attributes . type ) ; if ( this . factory === undefined ) { throw new Error ( `Channel Factory ${ attributes . type } not registered` ) ; } const channel = await this . factory . load ( this . runtime , this . id , this . services . value , attributes ) ; this . channel = channel ; for ( const message of this . pending ) { this . services . value . deltaConnection . process ( message , false , undefined ) ; } return this . channel ; } private isSnapshotInOldFormatAndCollectBlobs ( snapshotTree , blobMap ) { let sanitize = false ; const blobsContents = ( snapshotTree as any ) . blobsContents ; Object . entries ( blobsContents ) . forEach ( ( [ key , value ] ) => { blobMap . set ( key , value ) ; if ( snapshotTree . blobs [ key ] !== undefined ) { sanitize = true ; } } ) ; for ( const value of Object . values ( snapshotTree . trees ) ) { sanitize = sanitize || this . isSnapshotInOldFormatAndCollectBlobs ( value , blobMap ) ; } return sanitize ; } private sanitizeSnapshot ( snapshotTree ) { const blobMapInitial = new Map ( Object . entries ( snapshotTree . blobs ) ) ; for ( const [ blobName , blobId ] of blobMapInitial . entries ( ) ) { const blobValue = blobMapInitial . get ( blobId ) ; if ( blobValue === undefined ) { delete snapshotTree . blobs [ blobName ] ; } } for ( const value of Object . values ( snapshotTree . trees ) ) { this . sanitizeSnapshot ( value ) ; } } } export class LocalChannelContext extends LocalChannelContextBase { private readonly services : Lazy < { readonly deltaConnection : ChannelDeltaConnection ; readonly objectStorage : ChannelStorageService ; } > ; private readonly dirtyFn : ( ) => void ; constructor ( id , registry , type , runtime , dataStoreContext , storageService , submitFn , dirtyFn , addedGCOutboundReferenceFn ) { super ( id , registry , runtime , ( ) => this . services ) ; assert ( type !== undefined , 0x209 ) ; this . factory = registry . get ( type ) ; if ( this . factory === undefined ) { throw new Error ( `Channel Factory ${ type } not registered` ) ; } this . channel = this . factory . create ( runtime , id ) ; this . services = new Lazy ( ( ) => { return createServiceEndpoints ( this . id , dataStoreContext . connected , submitFn , this . dirtyFn , addedGCOutboundReferenceFn , storageService ) ; } ) ; this . dirtyFn = ( ) => { dirtyFn ( id ) ; } ; } }