Remove batch manager with feature gate

api-report/driver-base.api.md

@@ -48,6 +48,8 @@ export class DocumentDeltaConnection extends TypedEventEmitter<IDocumentDeltaCon
     // (undocumented)
     protected earlySignalHandler: (msg: ISignalMessage) => void;
     // (undocumented)
+    protected emitMessages(type: string, messages: IDocumentMessage[][]): void;
+    // (undocumented)
     static readonly eventsAlwaysForwarded: string[];
     // (undocumented)
     static readonly eventsToForward: string[];
@@ -60,6 +62,8 @@ export class DocumentDeltaConnection extends TypedEventEmitter<IDocumentDeltaCon
     get initialMessages(): ISequencedDocumentMessage[];
     get initialSignals(): ISignalMessage[];
     // (undocumented)
+    protected readonly isBatchManagerDisabled: boolean;
+    // (undocumented)
     protected readonly logger: ITelemetryLogger;
     get maxMessageSize(): number;
     get mode(): ConnectionMode;
@@ -72,6 +76,8 @@ export class DocumentDeltaConnection extends TypedEventEmitter<IDocumentDeltaCon
     protected readonly socket: SocketIOClient.Socket;
     submit(messages: IDocumentMessage[]): void;
     // (undocumented)
+    protected submitCore(type: string, messages: IDocumentMessage[]): void;
+    // (undocumented)
     protected readonly submitManager: BatchManager<IDocumentMessage[]>;
     submitSignal(message: IDocumentMessage): void;
     get version(): string;

packages/drivers/driver-base/src/documentDeltaConnection.ts

@@ -3,7 +3,7 @@
  * Licensed under the MIT License.
  */
 
-import { assert , BatchManager, TypedEventEmitter } from "@fluidframework/common-utils";
+import { assert, BatchManager, TypedEventEmitter } from "@fluidframework/common-utils";
 import {
     IDocumentDeltaConnection,
     IDocumentDeltaConnectionEvents,
@@ -25,6 +25,9 @@ import {
 import { IDisposable, ITelemetryLogger } from "@fluidframework/common-definitions";
 import { ChildLogger } from "@fluidframework/telemetry-utils";
 
+// Local storage key to disable the BatchManager
+const batchManagerDisabledKey = "FluidDisableBatchManager";
+
 /**
  * Represents a connection to a stream of delta updates
  */
@@ -80,6 +83,7 @@ export class DocumentDeltaConnection
      */
     protected _disposed: boolean = false;
     protected readonly logger: ITelemetryLogger;
+    protected readonly isBatchManagerDisabled: boolean = false;
 
     public get details(): IConnected {
         if (!this._details) {
@@ -102,14 +106,7 @@ export class DocumentDeltaConnection
         this.logger = ChildLogger.create(logger, "DeltaConnection");
 
         this.submitManager = new BatchManager<IDocumentMessage[]>(
-            (submitType, work) => {
-                // Although the implementation here disconnects the socket and does not reuse it, other subclasses
-                // (e.g. OdspDocumentDeltaConnection) may reuse the socket.  In these cases, we need to avoid emitting
-                // on the still-live socket.
-                if (!this.disposed) {
-                    this.socket.emit(submitType, this.clientId, work);
-                }
-            });
+            (submitType, work) => this.emitMessages(submitType, work));
 
         this.on("newListener", (event, listener) => {
             assert(!this.disposed, 0x20a /* "register for event on disposed object" */);
@@ -138,6 +135,8 @@ export class DocumentDeltaConnection
                     });
             }
         });
+
+        this.isBatchManagerDisabled = DocumentDeltaConnection.disabledBatchManagerFeatureGate;
     }
 
     /**
@@ -259,14 +258,40 @@ export class DocumentDeltaConnection
         return this.details.initialClients;
     }
 
+    protected emitMessages(type: string, messages: IDocumentMessage[][]) {
+        // Although the implementation here disconnects the socket and does not reuse it, other subclasses
+        // (e.g. OdspDocumentDeltaConnection) may reuse the socket.  In these cases, we need to avoid emitting
+        // on the still-live socket.
+        if (!this.disposed) {
+            this.socket.emit(type, this.clientId, messages);
+        }
+    }
+
+    private static get disabledBatchManagerFeatureGate() {
+        try {
+            return localStorage !== undefined
+                && typeof localStorage === "object"
+                && localStorage.getItem(batchManagerDisabledKey) === "1";
+        } catch (e) { }
+        return false;
+    }
+
+    protected submitCore(type: string, messages: IDocumentMessage[]) {
+        if (this.isBatchManagerDisabled) {
+            this.emitMessages(type, [messages]);
+        } else {
+            this.submitManager.add(type, messages);
+        }
+    }
+
     /**
      * Submits a new delta operation to the server
      *
      * @param message - delta operation to submit
      */
     public submit(messages: IDocumentMessage[]): void {
         this.checkNotClosed();
-        this.submitManager.add("submitOp", messages);
+        this.submitCore("submitOp", messages);
     }
 
     /**
@@ -276,7 +301,7 @@ export class DocumentDeltaConnection
      */
     public submitSignal(message: IDocumentMessage): void {
         this.checkNotClosed();
-        this.submitManager.add("submitSignal", [message]);
+        this.submitCore("submitSignal", [message]);
     }
 
     /**

packages/drivers/local-driver/src/localDocumentDeltaConnection.ts

@@ -62,24 +62,31 @@ export class LocalDocumentDeltaConnection extends DocumentDeltaConnection {
           super(socket, documentId, new TelemetryNullLogger());
     }
 
+    protected submitCore(type: string, messages: IDocumentMessage[]) {
+        if (this.isBatchManagerDisabled) {
+            this.emitMessages(type, [messages]);
+        } else {
+            this.submitManager.add(type, messages);
+            this.submitManager.drain();
+        }
+    }
+
     /**
      * Submits a new delta operation to the server
      */
     public submit(messages: IDocumentMessage[]): void {
         // We use a promise resolve to force a turn break given message processing is sync
         // eslint-disable-next-line @typescript-eslint/no-floating-promises
         Promise.resolve().then(() => {
-            this.submitManager.add("submitOp", messages);
-            this.submitManager.drain();
+            this.submitCore("submitOp", messages);
         });
     }
 
     /**
      * Submits a new signal to the server
      */
     public submitSignal(message: any): void {
-        this.submitManager.add("submitSignal", message);
-        this.submitManager.drain();
+        this.submitCore("submitSignal", [message]);
     }
 
     /**