-
Notifications
You must be signed in to change notification settings - Fork 536
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Wait to raise Connected event for Read connection until container is …
…caught up (#9377) The desired outcome is that regardless of connection mode (which is an implementation detail - we use read connection even if we have write permissions sometimes), the semantics of the "connected" event are consisted: It should mean that you're connected and caught up. This has been the case for write but not for read connections. This change is "dark" meaning no behavior change is expected with this PR, but there's a setting "Fluid.Container.CatchUpBeforeDeclaringConnected" which can be set to true to enable the new behavior. Once this is tested and permanently enabled, we should be able to deprecate `waitContainerToCatchUp` in favor of just responding to the `connected` event.
- Loading branch information
1 parent
4beded1
commit 0b5950d
Showing
6 changed files
with
532 additions
and
18 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,99 @@ | ||
| /*! | ||
| * Copyright (c) Microsoft Corporation and contributors. All rights reserved. | ||
| * Licensed under the MIT License. | ||
| */ | ||
|
|
||
| import { IDisposable, IEvent } from "@fluidframework/common-definitions"; | ||
| import { assert, TypedEventEmitter } from "@fluidframework/common-utils"; | ||
| import { IDeltaManager } from "@fluidframework/container-definitions"; | ||
| import { ISequencedDocumentMessage } from "@fluidframework/protocol-definitions"; | ||
|
|
||
| /** @see ICatchUpMonitor for usage */ | ||
| type CaughtUpListener = () => void; | ||
|
|
||
| /** @see ICatchUpMonitor for usage */ | ||
| export interface ICatchUpMonitorEvents extends IEvent { | ||
| (event: "caughtUp", listener: CaughtUpListener): void; | ||
| } | ||
|
|
||
| /** Monitor that emits an event when a Container has caught up to a given point in the op stream */ | ||
| export interface ICatchUpMonitor extends TypedEventEmitter<ICatchUpMonitorEvents>, IDisposable { } | ||
|
|
||
| /** | ||
| * Monitors a Container's DeltaManager, notifying listeners when all ops have been processed | ||
| * that were known at the time the monitor was created. | ||
| */ | ||
| export class CatchUpMonitor extends TypedEventEmitter<ICatchUpMonitorEvents> implements ICatchUpMonitor { | ||
| private readonly targetSeqNumber: number; | ||
| private caughtUp: boolean = false; | ||
|
|
||
| private readonly opHandler = (message: Pick<ISequencedDocumentMessage, "sequenceNumber">) => { | ||
| if (!this.caughtUp && message.sequenceNumber >= this.targetSeqNumber) { | ||
| this.caughtUp = true; | ||
| this.emit("caughtUp"); | ||
| } | ||
| }; | ||
|
|
||
| /** | ||
| * Create the CatchUpMonitor, setting the target sequence number to wait for based on DeltaManager's current state. | ||
| */ | ||
| constructor( | ||
| private readonly deltaManager: IDeltaManager<any, any>, | ||
| ) { | ||
| super(); | ||
|
|
||
| this.targetSeqNumber = this.deltaManager.lastKnownSeqNumber; | ||
|
|
||
| assert(this.targetSeqNumber >= this.deltaManager.lastSequenceNumber, | ||
| "Cannot wait for seqNumber below last processed sequence number"); | ||
|
|
||
| this.deltaManager.on("op", this.opHandler); | ||
|
|
||
| // Simulate the last processed op to set caughtUp in case we already are | ||
| this.opHandler({ sequenceNumber: this.deltaManager.lastSequenceNumber }); | ||
|
|
||
| // If a listener is added after we are already caught up, notify that new listener immediately | ||
| this.on("newListener", (event: string, listener) => { | ||
| if (event === "caughtUp") { | ||
| const caughtUpListener = listener as CaughtUpListener; | ||
| if (this.caughtUp) { | ||
| caughtUpListener(); | ||
| } | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| public disposed: boolean = false; | ||
| public dispose() { | ||
| if (this.disposed) { | ||
| return; | ||
| } | ||
| this.disposed = true; | ||
|
|
||
| this.removeAllListeners(); | ||
| this.deltaManager.off("op", this.opHandler); | ||
| } | ||
| } | ||
|
|
||
| /** Monitor that always notifies listeners immediately */ | ||
| export class ImmediateCatchUpMonitor extends TypedEventEmitter<ICatchUpMonitorEvents> implements ICatchUpMonitor { | ||
| constructor() { | ||
| super(); | ||
| this.on("newListener", (event: string, listener) => { | ||
| if (event === "caughtUp") { | ||
| const caughtUpListener = listener as CaughtUpListener; | ||
| caughtUpListener(); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| public disposed: boolean = false; | ||
| public dispose() { | ||
| if (this.disposed) { | ||
| return; | ||
| } | ||
| this.disposed = true; | ||
|
|
||
| this.removeAllListeners(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.