Refactor stream inteceptor

Summary:
There was a frustrating issue previously where the UI would jump everytime a new frame came in with bloks data. This would occur constantly as the client was generating new frames every 10 seconds. The issue was because the previous appraoch to speed up the augmentation was to display the raw client raw and then asynchronously process a new frame with the added derived components.

The issue was the raw client frame has less nodes so we would breifly display this, and because the reduciton trace was cache on the next tick we would show this causing a jump.

Now we have a new solution:

1. We use an event emitter to push frames / metadata into the interceptor
2. The interceptor should always produce 1 frame, this can be the same or augmented, for open source its just the same
3. We use react query to check if we already have the reduction trace cached, if so then we are able to quickly process the whole thing, if not its split into 2 parts.

React query is also nice since we can add a cache time for the reduciton traces which are very large and not useful one your screen is gone. This should help keep memory usage under control a bit.

one other benefit to this approach is i can remove the error handling stuff from the main plugin code, since the blok augmentation is not on the critical path and it can retry internally and push out a new frame whenever.

Reviewed By: antonk52

Differential Revision: D49272152

fbshipit-source-id: e5539d36231a32754e8612014195449b9faafdb1

desktop/plugins/public/ui-debugger/DesktopTypes.tsx

@@ -19,6 +19,7 @@ import {
   Metadata,
   SnapshotInfo,
 } from './ClientTypes';
+import TypedEmitter from 'typed-emitter';
 
 export type LiveClientState = {
   snapshotInfo: SnapshotInfo | null;
@@ -127,13 +128,22 @@ export type StreamState =
       clearCallBack: () => Promise<void>;
     };
 
-export interface StreamInterceptor {
+export type DesktopFrame = {
-  transformNodes(
+  nodes: Map<Id, ClientNode>;
-    nodes: Map<Id, ClientNode>,
+  snapshot?: SnapshotInfo;
-  ): Promise<[Map<Id, ClientNode>, Metadata[]]>;
+  frameTime: number;
+};
 
-  transformMetadata(metadata: Metadata): Promise<Metadata>;
+export type StreamInterceptorEventEmitter = TypedEmitter<{
-}
+  /* one of these event will be emitted when frame comes from client */
+  frameReceived: (frame: DesktopFrame) => void;
+  /* at leat one these events will be emitted in reponse to frame received from client */
+  frameUpdated: (frame: DesktopFrame) => void;
+  /* one of these events will be emitted when metadata comes from client */
+  metadataReceived: (metadata: Metadata[]) => void;
+  /* at leat one these events will be emitted in reponse to frame received from client */
+  metadataUpdated: (metadata: Metadata[]) => void;
+}>;
 
 export class StreamInterceptorError extends Error {
   title: string;

desktop/plugins/public/ui-debugger/fb-stubs/StreamInterceptor.tsx

@@ -8,25 +8,21 @@
  */
 
 import {DeviceOS} from 'flipper-plugin';
-import {Id, Metadata, ClientNode} from '../ClientTypes';
+import {StreamInterceptorEventEmitter} from '../DesktopTypes';
-import {StreamInterceptor} from '../DesktopTypes';
 
-export function getStreamInterceptor(_: DeviceOS): StreamInterceptor {
+/**
-  return new NoOpStreamInterceptor();
+ * Stream inteceptors have the change to modify the frame or metata early in the pipeline
-}
+ */
-
+export function addInterceptors(
-class NoOpStreamInterceptor implements StreamInterceptor {
+  _deviceOS: DeviceOS,
-  init() {
+  eventEmitter: StreamInterceptorEventEmitter,
-    return null;
+) {
-  }
+  //no-op impmentation for open source
-
+  eventEmitter.on('frameReceived', async (frame) => {
-  async transformNodes(
+    eventEmitter.emit('frameUpdated', frame);
-    nodes: Map<Id, ClientNode>,
+  });
-  ): Promise<[Map<Id, ClientNode>, Metadata[]]> {
+
-    return [nodes, []];
+  eventEmitter.on('metadataReceived', async (metadata) => {
-  }
+    eventEmitter.emit('metadataUpdated', metadata);
-
+  });
-  async transformMetadata(metadata: Metadata): Promise<Metadata> {
-    return metadata;
-  }
 }

desktop/plugins/public/ui-debugger/index.tsx

@@ -23,29 +23,27 @@ import {
 import {
   UIState,
   NodeSelection,
-  StreamInterceptorError,
   StreamState,
   ReadOnlyUIState,
   LiveClientState,
   WireFrameMode,
   AugmentedFrameworkEvent,
+  StreamInterceptorEventEmitter,
 } from './DesktopTypes';
-import {getStreamInterceptor} from './fb-stubs/StreamInterceptor';
+import EventEmitter from 'eventemitter3';
+import {addInterceptors} from './fb-stubs/StreamInterceptor';
 import {prefetchSourceFileLocation} from './components/fb-stubs/IDEContextMenu';
 import {checkFocusedNodeStillActive} from './plugin/ClientDataUtils';
 import {uiActions} from './plugin/uiActions';
 import {first} from 'lodash';
 import {getNode} from './utils/map';
 
-type PendingData = {
-  metadata: Record<MetadataId, Metadata>;
-  frame: FrameScanEvent | null;
-};
-
 export function plugin(client: PluginClient<Events>) {
   const rootId = createState<Id | undefined>(undefined);
   const metadata = createState<Map<MetadataId, Metadata>>(new Map());
-  const streamInterceptor = getStreamInterceptor(client.device.os);
+
+  const streamInterceptor = new EventEmitter() as StreamInterceptorEventEmitter;
+  addInterceptors(client.device.os, streamInterceptor);
   const snapshot = createState<SnapshotInfo | null>(null);
   const nodesAtom = createState<Map<Id, ClientNode>>(new Map());
   const frameworkEvents = createDataSource<AugmentedFrameworkEvent>([], {
@@ -67,17 +65,20 @@ export function plugin(client: PluginClient<Events>) {
     nodes: new Map(),
   };
 
+  let lastProcessedFrameTime = 0;
+
+  const _uiActions = uiActions(
+    uiState,
+    nodesAtom,
+    snapshot,
+    mutableLiveClientData,
+  );
+
   const perfEvents = createDataSource<PerformanceStatsEvent, 'txId'>([], {
     key: 'txId',
     limit: 10 * 1024,
   });
 
-  //this holds pending any pending data that needs to be applied in the event of a stream interceptor error
-  //while in the error state more metadata or a more recent frame may come in so both cases need to apply the same darta
-  const pendingData: PendingData = {frame: null, metadata: {}};
-
-  let lastFrameTime = 0;
-
   client.onMessage('init', (event) => {
     console.log('[ui-debugger] init');
     rootId.set(event.rootId);
@@ -103,82 +104,20 @@ export function plugin(client: PluginClient<Events>) {
     console.log('[ui-debugger] disconnected');
   });
 
-  async function processMetadata(
-    incomingMetadata: Record<MetadataId, Metadata>,
-  ) {
-    try {
-      const mappedMeta = await Promise.all(
-        Object.values(incomingMetadata).map((metadata) =>
-          streamInterceptor.transformMetadata(metadata),
-        ),
-      );
-
-      metadata.update((draft) => {
-        for (const metadata of mappedMeta) {
-          draft.set(metadata.id, metadata);
-        }
-      });
-      return true;
-    } catch (error) {
-      for (const metadata of Object.values(incomingMetadata)) {
-        pendingData.metadata[metadata.id] = metadata;
-      }
-      handleStreamError('Metadata', error);
-      return false;
-    }
-  }
-
-  function handleStreamError(source: 'Frame' | 'Metadata', error: any) {
-    if (error instanceof StreamInterceptorError) {
-      const retryCallback = async () => {
-        uiState.streamState.set({state: 'RetryingAfterError'});
-
-        if (!(await processMetadata(pendingData.metadata))) {
-          //back into error state, dont proceed
-          return;
-        }
-        if (pendingData.frame != null) {
-          if (!(await processFrame(pendingData.frame))) {
-            //back into error state, dont proceed
-            return;
-          }
-        }
-
-        uiState.streamState.set({state: 'Ok'});
-        pendingData.frame = null;
-        pendingData.metadata = {};
-      };
-
-      uiState.streamState.set({
-        state: 'StreamInterceptorRetryableError',
-        retryCallback: retryCallback,
-        error: error,
-      });
-    } else {
-      console.error(
-        `[ui-debugger] Unexpected Error processing ${source}`,
-        error,
-      );
-
-      uiState.streamState.set({
-        state: 'FatalError',
-        error: error,
-        clearCallBack: async () => {
-          uiState.streamState.set({state: 'Ok'});
-          nodesAtom.set(new Map());
-          frameworkEvents.clear();
-          snapshot.set(null);
-        },
-      });
-    }
-  }
-
   client.onMessage('metadataUpdate', async (event) => {
     if (!event.attributeMetadata) {
       return;
     }
+    const metadata = Object.values(event.attributeMetadata);
+    streamInterceptor.emit('metadataReceived', metadata);
+  });
 
-    await processMetadata(event.attributeMetadata);
+  streamInterceptor.on('metadataUpdated', (updatedMetadata) => {
+    metadata.update((draft) => {
+      for (const meta of updatedMetadata) {
+        draft.set(meta.id, meta);
+      }
+    });
   });
 
   /**
@@ -209,37 +148,26 @@ export function plugin(client: PluginClient<Events>) {
   });
 
   const processFrame = async (frameScan: FrameScanEvent) => {
-    try {
+    const nodes = new Map(frameScan.nodes.map((node) => [node.id, {...node}]));
-      const nodes = new Map(
-        frameScan.nodes.map((node) => [node.id, {...node}]),
-      );
-      if (frameScan.frameTime > lastFrameTime) {
-        applyFrameData(nodes, frameScan.snapshot);
-        lastFrameTime = frameScan.frameTime;
-      }
-      applyFrameworkEvents(frameScan, nodes);
-      lastFrameTime = frameScan.frameTime;
 
-      const [processedNodes, additionalMetadata] =
+    streamInterceptor.emit('frameReceived', {
-        await streamInterceptor.transformNodes(nodes);
+      frameTime: frameScan.frameTime,
-
+      snapshot: frameScan.snapshot,
-      metadata.update((draft) => {
+      nodes: nodes,
-        for (const metadata of additionalMetadata) {
+    });
-          draft.set(metadata.id, metadata);
+    applyFrameworkEvents(frameScan, nodes);
-        }
-      });
-
-      if (frameScan.frameTime >= lastFrameTime) {
-        applyFrameData(processedNodes, frameScan.snapshot);
-        lastFrameTime = frameScan.frameTime;
-      }
-    } catch (error) {
-      pendingData.frame = frameScan;
-      handleStreamError('Frame', error);
-      return false;
-    }
   };
 
+  streamInterceptor.on('frameUpdated', (frame) => {
+    if (frame.frameTime > lastProcessedFrameTime) {
+      applyFrameData(frame.nodes, frame.snapshot);
+      lastProcessedFrameTime = frame.frameTime;
+      const selectedNode = uiState.selectedNode.get();
+      if (selectedNode != null)
+        _uiActions.ensureAncestorsExpanded(selectedNode.id);
+    }
+  });
+
   function applyFrameworkEvents(
     frameScan: FrameScanEvent,
     nodes: Map<Id, ClientNode>,
@@ -295,7 +223,6 @@ export function plugin(client: PluginClient<Events>) {
     }, HighlightTime);
   }
 
-  //todo deal with racecondition, where bloks screen is fetching, takes time then you go back get more recent frame then bloks screen comes and overrites it
   function applyFrameData(
     nodes: Map<Id, ClientNode>,
     snapshotInfo: SnapshotInfo | undefined,
@@ -318,6 +245,7 @@ export function plugin(client: PluginClient<Events>) {
       }
     }, 0);
   }
+
   client.onMessage('subtreeUpdate', (subtreeUpdate) => {
     processFrame({
       frameTime: subtreeUpdate.txId,
@@ -331,7 +259,7 @@ export function plugin(client: PluginClient<Events>) {
   return {
     rootId,
     uiState: uiState as ReadOnlyUIState,
-    uiActions: uiActions(uiState, nodesAtom, snapshot, mutableLiveClientData),
+    uiActions: _uiActions,
     nodes: nodesAtom,
     frameworkEvents,
     frameworkEventMetadata,

desktop/plugins/public/ui-debugger/package.json

@@ -20,10 +20,12 @@
     "async": "2.3.0",
     "@tanstack/react-virtual": "3.0.0-beta.54",
     "ts-retry-promise": "^0.7.0",
-    "memoize-weak": "^1.0.2"
+    "memoize-weak": "^1.0.2",
+    "eventemitter3": "^4.0.7"
   },
   "devDependencies": {
-    "@types/async": "3.2.20"
+    "@types/async": "3.2.20",
+    "typed-emitter": "^2.1.0"
   },
   "bugs": {
     "url": "https://github.com/facebook/flipper/issues"

desktop/plugins/public/yarn.lock

@@ -981,7 +981,7 @@ estree-walker@^2.0.1:
   resolved "https://registry.yarnpkg.com/estree-walker/-/estree-walker-2.0.2.tgz#52f010178c2a4c117a7757cfe942adb7d2da4cac"
   integrity sha512-Rfkk/Mp/DL7JVje3u18FxFujQlTNR2q6QfMSMB7AvCBx91NGj/ba3kCfza0f6dVDbw7YlRf/nDrn7pQrCCyQ/w==
 
-eventemitter3@^4.0.1:
+eventemitter3@^4.0.1, eventemitter3@^4.0.7:
   version "4.0.7"
   resolved "https://registry.yarnpkg.com/eventemitter3/-/eventemitter3-4.0.7.tgz#2de9b68f6528d5644ef5c59526a1b4a07306169f"
   integrity sha512-8guHBZCwKnFhYdHr2ysuRWErTwhoN2X8XELRlrRwpmfeY2jjuUN4taQMsULKUVo1K4DvZl+0pgfyoysHxvmvEw==
@@ -2132,6 +2132,13 @@ rollup@^2.70.1:
   optionalDependencies:
     fsevents "~2.3.2"
 
+rxjs@^7.5.2:
+  version "7.8.1"
+  resolved "https://registry.yarnpkg.com/rxjs/-/rxjs-7.8.1.tgz#6f6f3d99ea8044291efd92e7c7fcf562c4057543"
+  integrity sha512-AA3TVj+0A2iuIoQkWEK/tqFjBq2j+6PO6Y0zJcvzLAFhEFIO3HL0vls9hWLncZbAAbK0mar7oZ4V079I/qPMxg==
+  dependencies:
+    tslib "^2.1.0"
+
 safe-regex@^1.1.0:
   version "1.1.0"
   resolved "https://registry.yarnpkg.com/safe-regex/-/safe-regex-1.1.0.tgz#40a3669f3b077d1e943d44629e157dd48023bf2e"
@@ -2349,6 +2356,18 @@ ts-retry-promise@^0.7.0:
   resolved "https://registry.yarnpkg.com/ts-retry-promise/-/ts-retry-promise-0.7.0.tgz#08f2dcbbf5d2981495841cb63389a268324e8147"
   integrity sha512-x6yWZXC4BfXy4UyMweOFvbS1yJ/Y5biSz/mEPiILtJZLrqD3ZxIpzVOGGgifHHdaSe3WxzFRtsRbychI6zofOg==
 
+tslib@^2.1.0:
+  version "2.6.2"
+  resolved "https://registry.yarnpkg.com/tslib/-/tslib-2.6.2.tgz#703ac29425e7b37cd6fd456e92404d46d1f3e4ae"
+  integrity sha512-AEYxH93jGFPn/a2iVAwW87VuUIkR1FVUKB77NwMF7nBTDkDrrT/Hpt/IrCJ0QXhW27jTBDcf5ZY7w6RiqTMw2Q==
+
+typed-emitter@^2.1.0:
+  version "2.1.0"
+  resolved "https://registry.yarnpkg.com/typed-emitter/-/typed-emitter-2.1.0.tgz#ca78e3d8ef1476f228f548d62e04e3d4d3fd77fb"
+  integrity sha512-g/KzbYKbH5C2vPkaXGu8DJlHrGKHLsM25Zg9WuC9pMGfuvT+X25tZQWo5fK1BjBm8+UrVE9LDCvaY0CQk+fXDA==
+  optionalDependencies:
+    rxjs "^7.5.2"
+
 unicode-substring@^1.0.0:
   version "1.0.0"
   resolved "https://registry.yarnpkg.com/unicode-substring/-/unicode-substring-1.0.0.tgz#659fb839078e7bee84b86c27210ac4db215bf885"