Support receiving messages in Sandy plugins

Summary: This diffs adds the capability to listen to messages in Sandy plugins. Although API wise it looks more like the old `this.subscribe`, semantically it behaves like the `persistedStateReducer`; messages are queued if the plugin is enabled but not active.

Reviewed By: nikoant

Differential Revision: D22282711

fbshipit-source-id: 885faa702fe779ac8d593c1d224b2be13e688d47

desktop/app/src/Client.tsx

@@ -11,6 +11,8 @@ import {
   PluginDefinition,
   ClientPluginDefinition,
   isSandyPlugin,
+  FlipperPlugin,
+  FlipperDevicePlugin,
 } from './plugin';
 import BaseDevice, {OS} from './devices/BaseDevice';
 import {App} from './App';
@@ -135,7 +137,10 @@ export default class Client extends EventEmitter {
   messageBuffer: Record<
     string /*pluginKey*/,
     {
-      plugin: PluginDefinition;
+      plugin:
+        | typeof FlipperPlugin
+        | typeof FlipperDevicePlugin
+        | SandyPluginInstance;
       messages: Params[];
     }
   > = {};
@@ -456,11 +461,11 @@ export default class Client extends EventEmitter {
           this.store.getState().plugins.devicePlugins.get(params.api);
 
         let handled = false; // This is just for analysis
-        // TODO: support Sandy plugins T68683442
         if (
           persistingPlugin &&
-          !isSandyPlugin(persistingPlugin) &&
+          ((persistingPlugin as any).persistedStateReducer ||
-          persistingPlugin.persistedStateReducer
+            // only send messages to enabled sandy plugins
+            this.sandyPluginStates.has(params.api))
         ) {
           handled = true;
           const pluginKey = getPluginKey(
@@ -470,7 +475,8 @@ export default class Client extends EventEmitter {
           );
           if (!this.messageBuffer[pluginKey]) {
             this.messageBuffer[pluginKey] = {
-              plugin: persistingPlugin,
+              plugin: (this.sandyPluginStates.get(params.api) ??
+                persistingPlugin) as any,
               messages: [params],
             };
           } else {

desktop/app/src/utils/__tests__/messageQueueSandy.node.tsx

@@ -0,0 +1,639 @@
+/**
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ *
+ * @format
+ */
+
+import {FlipperDevicePlugin} from '../../plugin';
+import {createMockFlipperWithPlugin} from '../../test-utils/createMockFlipperWithPlugin';
+import {Store, Client, sleep} from '../../';
+import {
+  selectPlugin,
+  starPlugin,
+  selectClient,
+  selectDevice,
+} from '../../reducers/connections';
+import {processMessageQueue} from '../messageQueue';
+import {getPluginKey} from '../pluginUtils';
+import {TestIdler} from '../Idler';
+import {registerPlugins} from '../../reducers/plugins';
+import {
+  SandyPluginDefinition,
+  TestUtils,
+  FlipperClient,
+  SandyPluginInstance,
+} from 'flipper-plugin';
+
+type Events = {
+  inc: {
+    delta?: number;
+  };
+};
+
+function plugin(client: FlipperClient<Events, {}>) {
+  const state = {
+    count: 0,
+  };
+
+  client.onMessage('inc', (params) => {
+    state.count += params.delta || 1;
+  });
+
+  return {
+    state,
+  };
+}
+
+const TestPlugin = new SandyPluginDefinition(
+  TestUtils.createMockPluginDetails(),
+  {
+    plugin,
+    Component() {
+      return null;
+    },
+  },
+);
+
+function starTestPlugin(store: Store, client: Client) {
+  store.dispatch(
+    starPlugin({
+      plugin: TestPlugin,
+      selectedApp: client.query.app,
+    }),
+  );
+}
+
+function selectDeviceLogs(store: Store) {
+  store.dispatch(
+    selectPlugin({
+      selectedPlugin: 'DeviceLogs',
+      selectedApp: null,
+      deepLinkPayload: null,
+      selectedDevice: store.getState().connections.selectedDevice!,
+    }),
+  );
+}
+
+function selectTestPlugin(store: Store, client: Client) {
+  store.dispatch(
+    selectPlugin({
+      selectedPlugin: TestPlugin.id,
+      selectedApp: client.query.app,
+      deepLinkPayload: null,
+      selectedDevice: store.getState().connections.selectedDevice!,
+    }),
+  );
+}
+
+function getTestPluginState(
+  client: Client,
+): ReturnType<typeof plugin>['state'] {
+  return client.sandyPluginStates.get(TestPlugin.id)!.instanceApi.state;
+}
+
+test('queue - events are processed immediately if plugin is selected', async () => {
+  const {store, client, sendMessage} = await createMockFlipperWithPlugin(
+    TestPlugin,
+  );
+  expect(store.getState().connections.selectedPlugin).toBe('TestPlugin');
+  sendMessage('noop', {});
+  sendMessage('noop', {});
+  sendMessage('inc', {});
+  sendMessage('inc', {delta: 4});
+  sendMessage('noop', {});
+  client.flushMessageBuffer();
+  expect(getTestPluginState(client)).toMatchInlineSnapshot(`
+          Object {
+            "count": 5,
+          }
+        `);
+  expect(store.getState().pluginMessageQueue).toMatchInlineSnapshot(
+    `Object {}`,
+  );
+});
+
+test('queue - events are NOT processed immediately if plugin is NOT selected (but enabled)', async () => {
+  const {
+    store,
+    client,
+    sendMessage,
+    device,
+  } = await createMockFlipperWithPlugin(TestPlugin);
+  selectDeviceLogs(store);
+  expect(store.getState().connections.selectedPlugin).not.toBe('TestPlugin');
+
+  sendMessage('inc', {});
+  sendMessage('inc', {delta: 2});
+  sendMessage('inc', {delta: 3});
+  expect(store.getState().pluginStates).toMatchInlineSnapshot(`Object {}`);
+  expect(getTestPluginState(client).count).toBe(0);
+  // the first message is already visible cause of the leading debounce
+  expect(store.getState().pluginMessageQueue).toMatchInlineSnapshot(`
+        Object {
+          "TestApp#Android#MockAndroidDevice#serial#TestPlugin": Array [
+            Object {
+              "api": "TestPlugin",
+              "method": "inc",
+              "params": Object {},
+            },
+          ],
+        }
+      `);
+  client.flushMessageBuffer();
+  expect(store.getState().pluginMessageQueue).toMatchInlineSnapshot(`
+        Object {
+          "TestApp#Android#MockAndroidDevice#serial#TestPlugin": Array [
+            Object {
+              "api": "TestPlugin",
+              "method": "inc",
+              "params": Object {},
+            },
+            Object {
+              "api": "TestPlugin",
+              "method": "inc",
+              "params": Object {
+                "delta": 2,
+              },
+            },
+            Object {
+              "api": "TestPlugin",
+              "method": "inc",
+              "params": Object {
+                "delta": 3,
+              },
+            },
+          ],
+        }
+      `);
+
+  // process the message
+  const pluginKey = getPluginKey(client.id, device, TestPlugin.id);
+  await processMessageQueue(
+    client.sandyPluginStates.get(TestPlugin.id)!,
+    pluginKey,
+    store,
+  );
+  expect(getTestPluginState(client)).toEqual({
+    count: 6,
+  });
+
+  expect(store.getState().pluginMessageQueue).toEqual({
+    [pluginKey]: [],
+  });
+
+  // unstar. Messages don't arrive anymore
+  starTestPlugin(store, client);
+  // weird state...
+  selectTestPlugin(store, client);
+  sendMessage('inc', {delta: 3});
+  client.flushMessageBuffer();
+  // active, immediately processed
+  expect(client.sandyPluginStates.has(TestPlugin.id)).toBe(false);
+
+  // different plugin, and not starred, message will never arrive
+  selectDeviceLogs(store);
+  sendMessage('inc', {delta: 4});
+  client.flushMessageBuffer();
+  expect(store.getState().pluginMessageQueue).toEqual({
+    [pluginKey]: [],
+  });
+
+  // star again, plugin still not selected, message is queued
+  starTestPlugin(store, client);
+  sendMessage('inc', {delta: 5});
+  client.flushMessageBuffer();
+
+  expect(store.getState().pluginMessageQueue).toEqual({
+    [pluginKey]: [{api: 'TestPlugin', method: 'inc', params: {delta: 5}}],
+  });
+});
+
+test('queue - events are queued for plugins that are favorite when app is not selected', async () => {
+  const {
+    client,
+    device,
+    store,
+    sendMessage,
+    createClient,
+  } = await createMockFlipperWithPlugin(TestPlugin);
+  selectDeviceLogs(store);
+  expect(store.getState().connections.selectedPlugin).not.toBe('TestPlugin');
+
+  const client2 = await createClient(device, 'TestApp2');
+  store.dispatch(selectClient(client2.id));
+
+  // Now we send a message to the second client, it should arrive,
+  // as the plugin was enabled already on the first client as well
+  sendMessage('inc', {delta: 2});
+  expect(getTestPluginState(client)).toEqual({count: 0});
+  expect(store.getState().pluginMessageQueue).toMatchInlineSnapshot(`
+          Object {
+            "TestApp#Android#MockAndroidDevice#serial#TestPlugin": Array [
+              Object {
+                "api": "TestPlugin",
+                "method": "inc",
+                "params": Object {
+                  "delta": 2,
+                },
+              },
+            ],
+          }
+        `);
+});
+
+test('queue - events are queued for plugins that are favorite when app is selected on different device', async () => {
+  const {
+    client,
+    store,
+    sendMessage,
+    createDevice,
+    createClient,
+  } = await createMockFlipperWithPlugin(TestPlugin);
+  selectDeviceLogs(store);
+  expect(store.getState().connections.selectedPlugin).not.toBe('TestPlugin');
+
+  const device2 = createDevice('serial2');
+  const client2 = await createClient(device2, client.query.app); // same app id
+  store.dispatch(selectDevice(device2));
+  store.dispatch(selectClient(client2.id));
+
+  // Now we send a message to the first and second client, it should arrive,
+  // as the plugin was enabled already on the first client as well
+  sendMessage('inc', {delta: 2});
+  sendMessage('inc', {delta: 3}, client2);
+  client.flushMessageBuffer();
+  client2.flushMessageBuffer();
+  expect(getTestPluginState(client)).toEqual({count: 0});
+  expect(store.getState().pluginMessageQueue).toMatchInlineSnapshot(`
+    Object {
+      "TestApp#Android#MockAndroidDevice#serial#TestPlugin": Array [
+        Object {
+          "api": "TestPlugin",
+          "method": "inc",
+          "params": Object {
+            "delta": 2,
+          },
+        },
+      ],
+      "TestApp#Android#MockAndroidDevice#serial2#TestPlugin": Array [
+        Object {
+          "api": "TestPlugin",
+          "method": "inc",
+          "params": Object {
+            "delta": 3,
+          },
+        },
+      ],
+    }
+  `);
+});
+
+test('queue - events processing will be paused', async () => {
+  const {
+    client,
+    device,
+    store,
+    sendMessage,
+  } = await createMockFlipperWithPlugin(TestPlugin);
+  selectDeviceLogs(store);
+
+  sendMessage('inc', {});
+  sendMessage('inc', {delta: 3});
+  sendMessage('inc', {delta: 5});
+  client.flushMessageBuffer();
+
+  // process the message
+  const pluginKey = getPluginKey(client.id, device, TestPlugin.id);
+
+  // controlled idler will signal and and off that idling is needed
+  const idler = new TestIdler();
+
+  const p = processMessageQueue(
+    client.sandyPluginStates.get(TestPlugin.id)!,
+    pluginKey,
+    store,
+    undefined,
+    idler,
+  );
+
+  expect(getTestPluginState(client)).toEqual({
+    count: 4,
+  });
+
+  expect(store.getState().pluginMessageQueue).toEqual({
+    [pluginKey]: [{api: 'TestPlugin', method: 'inc', params: {delta: 5}}],
+  });
+
+  await idler.next();
+  expect(getTestPluginState(client)).toEqual({
+    count: 9,
+  });
+
+  expect(store.getState().pluginMessageQueue).toEqual({
+    [pluginKey]: [],
+  });
+
+  // don't idle anymore
+  idler.run();
+  await p;
+});
+
+test('queue - messages that arrive during processing will be queued', async () => {
+  const {
+    client,
+    device,
+    store,
+    sendMessage,
+  } = await createMockFlipperWithPlugin(TestPlugin);
+  selectDeviceLogs(store);
+
+  sendMessage('inc', {});
+  sendMessage('inc', {delta: 2});
+  sendMessage('inc', {delta: 3});
+  client.flushMessageBuffer();
+
+  // process the message
+  const pluginKey = getPluginKey(client.id, device, TestPlugin.id);
+
+  const idler = new TestIdler();
+
+  const p = processMessageQueue(
+    client.sandyPluginStates.get(TestPlugin.id)!,
+    pluginKey,
+    store,
+    undefined,
+    idler,
+  );
+
+  // first message is consumed
+  expect(store.getState().pluginMessageQueue[pluginKey].length).toBe(1);
+  expect(getTestPluginState(client).count).toBe(3);
+
+  // Select the current plugin as active, still, messages should end up in the queue
+  store.dispatch(
+    selectPlugin({
+      selectedPlugin: TestPlugin.id,
+      selectedApp: client.id,
+      deepLinkPayload: null,
+      selectedDevice: device,
+    }),
+  );
+  expect(store.getState().connections.selectedPlugin).toBe('TestPlugin');
+
+  sendMessage('inc', {delta: 4});
+  client.flushMessageBuffer();
+  // should not be processed yet
+  expect(store.getState().pluginMessageQueue[pluginKey].length).toBe(2);
+  expect(getTestPluginState(client).count).toBe(3);
+
+  await idler.next();
+  expect(store.getState().pluginMessageQueue[pluginKey].length).toBe(0);
+  expect(getTestPluginState(client).count).toBe(10);
+
+  idler.run();
+  await p;
+});
+
+test('queue - processing can be cancelled', async () => {
+  const {
+    client,
+    device,
+    store,
+    sendMessage,
+  } = await createMockFlipperWithPlugin(TestPlugin);
+  selectDeviceLogs(store);
+
+  sendMessage('inc', {});
+  sendMessage('inc', {delta: 2});
+  sendMessage('inc', {delta: 3});
+  sendMessage('inc', {delta: 4});
+  sendMessage('inc', {delta: 5});
+  client.flushMessageBuffer();
+
+  // process the message
+  const pluginKey = getPluginKey(client.id, device, TestPlugin.id);
+
+  const idler = new TestIdler();
+
+  const p = processMessageQueue(
+    client.sandyPluginStates.get(TestPlugin.id)!,
+    pluginKey,
+    store,
+    undefined,
+    idler,
+  );
+
+  // first message is consumed
+  await idler.next();
+  expect(store.getState().pluginMessageQueue[pluginKey].length).toBe(1);
+  expect(getTestPluginState(client).count).toBe(10);
+
+  idler.cancel();
+
+  expect(store.getState().pluginMessageQueue[pluginKey].length).toBe(1);
+  expect(getTestPluginState(client).count).toBe(10);
+  await p;
+});
+
+test('queue - make sure resetting plugin state clears the message queue', async () => {
+  const {
+    client,
+    device,
+    store,
+    sendMessage,
+  } = await createMockFlipperWithPlugin(TestPlugin);
+  selectDeviceLogs(store);
+
+  sendMessage('inc', {});
+  sendMessage('inc', {delta: 2});
+  client.flushMessageBuffer();
+
+  const pluginKey = getPluginKey(client.id, device, TestPlugin.id);
+
+  expect(store.getState().pluginMessageQueue[pluginKey].length).toBe(2);
+
+  store.dispatch({
+    type: 'CLEAR_PLUGIN_STATE',
+    payload: {clientId: client.id, devicePlugins: new Set()},
+  });
+
+  expect(store.getState().pluginMessageQueue[pluginKey]).toBe(undefined);
+});
+
+test('client - incoming messages are buffered and flushed together', async () => {
+  class StubDeviceLogs extends FlipperDevicePlugin<any, any, any> {
+    static id = 'DevicePlugin';
+
+    static supportsDevice() {
+      return true;
+    }
+
+    static persistedStateReducer = jest.fn();
+  }
+
+  const {
+    client,
+    store,
+    device,
+    sendMessage,
+    pluginKey,
+  } = await createMockFlipperWithPlugin(TestPlugin);
+  selectDeviceLogs(store);
+
+  store.dispatch(registerPlugins([StubDeviceLogs]));
+  sendMessage('inc', {});
+  sendMessage('inc', {delta: 2});
+  sendMessage('inc', {delta: 3});
+
+  // send a message to device logs
+  client.onMessage(
+    JSON.stringify({
+      method: 'execute',
+      params: {
+        api: 'DevicePlugin',
+        method: 'log',
+        params: {line: 'suff'},
+      },
+    }),
+  );
+
+  expect(store.getState().pluginStates).toMatchInlineSnapshot(`Object {}`);
+  expect(getTestPluginState(client).count).toBe(0);
+  // the first message is already visible cause of the leading debounce
+  expect(store.getState().pluginMessageQueue).toMatchInlineSnapshot(`
+        Object {
+          "TestApp#Android#MockAndroidDevice#serial#TestPlugin": Array [
+            Object {
+              "api": "TestPlugin",
+              "method": "inc",
+              "params": Object {},
+            },
+          ],
+        }
+      `);
+  expect(client.messageBuffer).toMatchInlineSnapshot(`
+        Object {
+          "TestApp#Android#MockAndroidDevice#serial#DevicePlugin": Object {
+            "messages": Array [
+              Object {
+                "api": "DevicePlugin",
+                "method": "log",
+                "params": Object {
+                  "line": "suff",
+                },
+              },
+            ],
+            "plugin": [Function],
+          },
+          "TestApp#Android#MockAndroidDevice#serial#TestPlugin": Object {
+            "messages": Array [
+              Object {
+                "api": "TestPlugin",
+                "method": "inc",
+                "params": Object {
+                  "delta": 2,
+                },
+              },
+              Object {
+                "api": "TestPlugin",
+                "method": "inc",
+                "params": Object {
+                  "delta": 3,
+                },
+              },
+            ],
+            "plugin": undefined,
+          },
+        }
+      `);
+  expect(client.messageBuffer[pluginKey].plugin).toBeInstanceOf(
+    SandyPluginInstance,
+  );
+
+  await sleep(500);
+  expect(store.getState().pluginMessageQueue).toMatchInlineSnapshot(`
+        Object {
+          "TestApp#Android#MockAndroidDevice#serial#DevicePlugin": Array [
+            Object {
+              "api": "DevicePlugin",
+              "method": "log",
+              "params": Object {
+                "line": "suff",
+              },
+            },
+          ],
+          "TestApp#Android#MockAndroidDevice#serial#TestPlugin": Array [
+            Object {
+              "api": "TestPlugin",
+              "method": "inc",
+              "params": Object {},
+            },
+            Object {
+              "api": "TestPlugin",
+              "method": "inc",
+              "params": Object {
+                "delta": 2,
+              },
+            },
+            Object {
+              "api": "TestPlugin",
+              "method": "inc",
+              "params": Object {
+                "delta": 3,
+              },
+            },
+          ],
+        }
+      `);
+  expect(client.messageBuffer).toMatchInlineSnapshot(`Object {}`);
+  expect(StubDeviceLogs.persistedStateReducer.mock.calls).toMatchInlineSnapshot(
+    `Array []`,
+  );
+
+  // tigger processing the queue
+  const pluginKeyDevice = getPluginKey(client.id, device, StubDeviceLogs.id);
+  await processMessageQueue(StubDeviceLogs, pluginKeyDevice, store);
+
+  expect(StubDeviceLogs.persistedStateReducer.mock.calls)
+    .toMatchInlineSnapshot(`
+        Array [
+          Array [
+            Object {},
+            "log",
+            Object {
+              "line": "suff",
+            },
+          ],
+        ]
+      `);
+
+  expect(store.getState().pluginMessageQueue).toMatchInlineSnapshot(`
+        Object {
+          "TestApp#Android#MockAndroidDevice#serial#DevicePlugin": Array [],
+          "TestApp#Android#MockAndroidDevice#serial#TestPlugin": Array [
+            Object {
+              "api": "TestPlugin",
+              "method": "inc",
+              "params": Object {},
+            },
+            Object {
+              "api": "TestPlugin",
+              "method": "inc",
+              "params": Object {
+                "delta": 2,
+              },
+            },
+            Object {
+              "api": "TestPlugin",
+              "method": "inc",
+              "params": Object {
+                "delta": 3,
+              },
+            },
+          ],
+        }
+      `);
+});

desktop/app/src/utils/messageQueue.tsx

@@ -7,27 +7,27 @@
  * @format
  */
 
-import {
+import {PersistedStateReducer, FlipperDevicePlugin} from '../plugin';
-  PersistedStateReducer,
-  FlipperDevicePlugin,
-  isSandyPlugin,
-} from '../plugin';
 import {State, MiddlewareAPI} from '../reducers/index';
 import {setPluginState} from '../reducers/pluginStates';
-import {flipperRecorderAddEvent} from './pluginStateRecorder';
+import {
+  flipperRecorderAddEvent,
+  isRecordingEvents,
+} from './pluginStateRecorder';
 import {
   clearMessageQueue,
   queueMessages,
   Message,
+  DEFAULT_MAX_QUEUE_SIZE,
 } from '../reducers/pluginMessageQueue';
 import {Idler, BaseIdler} from './Idler';
 import {pluginIsStarred, getSelectedPluginKey} from '../reducers/connections';
 import {deconstructPluginKey} from './clientUtils';
 import {defaultEnabledBackgroundPlugins} from './pluginUtils';
-import {SandyPluginDefinition} from 'flipper-plugin';
+import {SandyPluginInstance} from 'flipper-plugin';
 import {addBackgroundStat} from './pluginStats';
 
-function processMessage(
+function processMessageClassic(
   state: State,
   pluginKey: string,
   plugin: {
@@ -52,19 +52,47 @@ function processMessage(
   }
 }
 
+function processMessagesSandy(
+  pluginKey: string,
+  plugin: SandyPluginInstance,
+  messages: Message[],
+) {
+  const reducerStartTime = Date.now();
+  if (isRecordingEvents(pluginKey)) {
+    messages.forEach((message) => {
+      flipperRecorderAddEvent(pluginKey, message.method, message.params);
+    });
+  }
+  try {
+    plugin.receiveMessages(messages);
+    addBackgroundStat(plugin.definition.id, Date.now() - reducerStartTime);
+  } catch (e) {
+    console.error(
+      `Failed to process event for plugin ${plugin.definition.id}`,
+      e,
+    );
+  }
+}
+
 export function processMessagesImmediately(
   store: MiddlewareAPI,
   pluginKey: string,
-  plugin: {
+  plugin:
+    | {
         defaultPersistedState: any;
         id: string;
         persistedStateReducer: PersistedStateReducer | null;
-  },
+      }
+    | SandyPluginInstance,
   messages: Message[],
 ) {
+  if (plugin instanceof SandyPluginInstance) {
+    processMessagesSandy(pluginKey, plugin, messages);
+  } else {
     const persistedState = getCurrentPluginState(store, plugin, pluginKey);
     const newPluginState = messages.reduce(
-    (state, message) => processMessage(state, pluginKey, plugin, message),
+      (state, message) =>
+        processMessageClassic(state, pluginKey, plugin, message),
       persistedState,
     );
     if (persistedState !== newPluginState) {
@@ -75,6 +103,7 @@ export function processMessagesImmediately(
         }),
       );
     }
+  }
 }
 
 export function processMessagesLater(
@@ -87,54 +116,61 @@ export function processMessagesLater(
         persistedStateReducer: PersistedStateReducer | null;
         maxQueueSize?: number;
       }
-    | SandyPluginDefinition,
+    | SandyPluginInstance,
   messages: Message[],
 ) {
-  // @ts-ignore
+  const pluginId =
-  if (isSandyPlugin(plugin)) {
+    plugin instanceof SandyPluginInstance ? plugin.definition.id : plugin.id;
-    // TODO:
-    throw new Error(
-      'Receiving messages is not yet supported for Sandy plugins',
-    );
-  }
   const isSelected =
     pluginKey === getSelectedPluginKey(store.getState().connections);
   switch (true) {
-    case plugin.id === 'Navigation': // Navigation events are always processed, to make sure the navbar stays up to date
+    case pluginId === 'Navigation': // Navigation events are always processed, to make sure the navbar stays up to date
     case isSelected && getPendingMessages(store, pluginKey).length === 0:
       processMessagesImmediately(store, pluginKey, plugin, messages);
       break;
+    // TODO: support SandyDevicePlugin T68738317
     case isSelected:
+    case plugin instanceof SandyPluginInstance:
     case plugin instanceof FlipperDevicePlugin:
     case (plugin as any).prototype instanceof FlipperDevicePlugin:
     case pluginIsStarred(
       store.getState().connections.userStarredPlugins,
       deconstructPluginKey(pluginKey).client,
-      plugin.id,
+      pluginId,
     ):
-      store.dispatch(queueMessages(pluginKey, messages, plugin.maxQueueSize));
+      store.dispatch(
+        queueMessages(
+          pluginKey,
+          messages,
+          plugin instanceof SandyPluginInstance
+            ? DEFAULT_MAX_QUEUE_SIZE
+            : plugin.maxQueueSize,
+        ),
+      );
       break;
     default:
       // In all other cases, messages will be dropped...
-      if (!defaultEnabledBackgroundPlugins.includes(plugin.id))
+      if (!defaultEnabledBackgroundPlugins.includes(pluginId))
         console.warn(
-          `Received message for disabled plugin ${plugin.id}, dropping..`,
+          `Received message for disabled plugin ${pluginId}, dropping..`,
         );
   }
 }
 
 export async function processMessageQueue(
-  plugin: {
+  plugin:
+    | {
         defaultPersistedState: any;
         id: string;
         persistedStateReducer: PersistedStateReducer | null;
-  },
+      }
+    | SandyPluginInstance,
   pluginKey: string,
   store: MiddlewareAPI,
   progressCallback?: (progress: {current: number; total: number}) => void,
   idler: BaseIdler = new Idler(),
 ): Promise<boolean> {
-  if (!plugin.persistedStateReducer) {
+  if (!SandyPluginInstance.is(plugin) && !plugin.persistedStateReducer) {
     return true;
   }
   const total = getPendingMessages(store, pluginKey).length;
@@ -145,16 +181,24 @@ export async function processMessageQueue(
       break;
     }
     // there are messages to process! lets do so until we have to idle
-    const persistedState = getCurrentPluginState(store, plugin, pluginKey);
+    // persistedState is irrelevant for SandyPlugins, as they store state locally
+    const persistedState = SandyPluginInstance.is(plugin)
+      ? undefined
+      : getCurrentPluginState(store, plugin, pluginKey);
     let offset = 0;
     let newPluginState = persistedState;
     do {
-      newPluginState = processMessage(
+      if (SandyPluginInstance.is(plugin)) {
+        // Optimization: we could send a batch of messages here
+        processMessagesSandy(pluginKey, plugin, [messages[offset]]);
+      } else {
+        newPluginState = processMessageClassic(
           newPluginState,
           pluginKey,
           plugin,
           messages[offset],
         );
+      }
       offset++;
       progress++;
 
@@ -168,7 +212,7 @@ export async function processMessageQueue(
     // resistent to kicking off this process twice; grabbing, processing messages, saving state is done synchronosly
     // until the idler has to break
     store.dispatch(clearMessageQueue(pluginKey, offset));
-    if (newPluginState !== persistedState) {
+    if (!SandyPluginInstance.is(plugin) && newPluginState !== persistedState) {
       store.dispatch(
         setPluginState({
           pluginKey,

desktop/app/src/utils/pluginStateRecorder.tsx

@@ -32,6 +32,10 @@ function initialRecordingState(): typeof pluginRecordingState {
   };
 }
 
+export function isRecordingEvents(pluginKey: string) {
+  return pluginRecordingState.recording === pluginKey;
+}
+
 export function flipperRecorderAddEvent(
   pluginKey: string,
   method: string,

desktop/flipper-plugin/src/__tests__/TestPlugin.tsx

@@ -24,6 +24,9 @@ export function plugin(client: FlipperClient<Events, Methods>) {
   const connectStub = jest.fn();
   const disconnectStub = jest.fn();
   const destroyStub = jest.fn();
+  const state = {
+    count: 0,
+  };
 
   // TODO: add tests for sending and receiving data T68683442
   // including typescript assertions
@@ -31,12 +34,23 @@ export function plugin(client: FlipperClient<Events, Methods>) {
   client.onConnect(connectStub);
   client.onDisconnect(disconnectStub);
   client.onDestroy(destroyStub);
+  client.onMessage('inc', ({delta}) => {
+    state.count += delta;
+  });
 
   function _unused_JustTypeChecks() {
     // @ts-expect-error Argument of type '"bla"' is not assignable
     client.send('bla', {});
     // @ts-expect-error Argument of type '{ stuff: string; }' is not assignable to parameter of type
     client.send('currentState', {stuff: 'nope'});
+    // @ts-expect-error
+    client.onMessage('stuff', (_params) => {
+      // noop
+    });
+    client.onMessage('inc', (params) => {
+      // @ts-expect-error
+      params.bla;
+    });
   }
 
   async function getCurrentState() {
@@ -48,6 +62,7 @@ export function plugin(client: FlipperClient<Events, Methods>) {
     destroyStub,
     disconnectStub,
     getCurrentState,
+    state,
   };
 }
 

desktop/flipper-plugin/src/__tests__/test-utils.node.tsx

@@ -93,3 +93,11 @@ test('a plugin cannot send messages after being disconnected', async () => {
   }
   expect(threw).toBeTruthy();
 });
+
+test('a plugin can receive messages', async () => {
+  const {instance, sendEvent} = TestUtils.startPlugin(testPlugin);
+  expect(instance.state.count).toBe(0);
+
+  sendEvent('inc', {delta: 2});
+  expect(instance.state.count).toBe(2);
+});

desktop/flipper-plugin/src/plugin/Plugin.tsx

@@ -13,6 +13,11 @@ import {EventEmitter} from 'events';
 type EventsContract = Record<string, any>;
 type MethodsContract = Record<string, (params: any) => Promise<any>>;
 
+type Message = {
+  method: string;
+  params?: any;
+};
+
 /**
  * API available to a plugin factory
  */
@@ -47,6 +52,17 @@ export interface FlipperClient<
     method: Method,
     params: Parameters<Methods[Method]>[0],
   ): ReturnType<Methods[Method]>;
+
+  /**
+   * Subscribe to a specific event arriving from the device.
+   *
+   * Messages can only arrive if the plugin is enabled and connected.
+   * For background plugins messages will be batched and arrive the next time the plugin is connected.
+   */
+  onMessage<Event extends keyof Events>(
+    event: Event,
+    callback: (params: Events[Event]) => void,
+  ): void;
 }
 
 /**
@@ -73,6 +89,10 @@ export type FlipperPluginFactory<
 export type FlipperPluginComponent = React.FC<{}>;
 
 export class SandyPluginInstance {
+  static is(thing: any): thing is SandyPluginInstance {
+    return thing instanceof SandyPluginInstance;
+  }
+
   /** base client provided by Flipper */
   realClient: RealFlipperClient;
   /** client that is bound to this instance */
@@ -111,6 +131,9 @@ export class SandyPluginInstance {
           params as any,
         );
       },
+      onMessage: (event, callback) => {
+        this.events.on('event-' + event, callback);
+      },
     };
     this.instanceApi = definition.module.plugin(this.client);
   }
@@ -156,6 +179,12 @@ export class SandyPluginInstance {
     this.destroyed = true;
   }
 
+  receiveMessages(messages: Message[]) {
+    messages.forEach((message) => {
+      this.events.emit('event-' + message.method, message.params);
+    });
+  }
+
   toJSON() {
     this.assertNotDestroyed();
     // TODO: T68683449

desktop/flipper-plugin/src/test-utils/test-utils.tsx

@@ -36,12 +36,19 @@ interface StartPluginOptions {
 type ExtractClientType<Module extends FlipperPluginModule<any>> = Parameters<
   Module['plugin']
 >[0];
+
 type ExtractMethodsType<
   Module extends FlipperPluginModule<any>
 > = ExtractClientType<Module> extends FlipperClient<any, infer Methods>
   ? Methods
   : never;
 
+type ExtractEventsType<
+  Module extends FlipperPluginModule<any>
+> = ExtractClientType<Module> extends FlipperClient<infer Events, any>
+  ? Events
+  : never;
+
 interface StartPluginResult<Module extends FlipperPluginModule<any>> {
   /**
    * the instantiated plugin for this test
@@ -73,6 +80,24 @@ interface StartPluginResult<Module extends FlipperPluginModule<any>> {
       params: Parameters<ExtractMethodsType<Module>[Method]>[0],
     ) => ReturnType<ExtractMethodsType<Module>[Method]>
   >;
+  /**
+   * Send event to the plugin
+   */
+  sendEvent<Event extends keyof ExtractEventsType<Module>>(
+    event: Event,
+    params: ExtractEventsType<Module>[Event],
+  ): void;
+  /**
+   * Send events to the plugin
+   * The structure used here reflects events that can be recorded
+   * with the pluginRecorder
+   */
+  sendEvents(
+    events: {
+      method: keyof ExtractEventsType<Module>;
+      params: any; // afaik we can't type this :-(
+    }[],
+  ): void;
 }
 
 export function startPlugin<Module extends FlipperPluginModule<any>>(
@@ -107,16 +132,28 @@ export function startPlugin<Module extends FlipperPluginModule<any>>(
   // we start connected
   pluginInstance.connect();
 
-  return {
+  const res: StartPluginResult<Module> = {
     module,
     instance: pluginInstance.instanceApi,
     connect: () => pluginInstance.connect(),
     disconnect: () => pluginInstance.disconnect(),
     destroy: () => pluginInstance.destroy(),
     onSend: sendStub,
-    // @ts-ignore
+    sendEvent: (event, params) => {
-    _backingInstance: pluginInstance,
+      res.sendEvents([
+        {
+          method: event,
+          params,
+        },
+      ]);
+    },
+    sendEvents: (messages) => {
+      pluginInstance.receiveMessages(messages as any);
+    },
   };
+  // @ts-ignore
+  res._backingInstance = pluginInstance;
+  return res;
 }
 
 export function renderPlugin<Module extends FlipperPluginModule<any>>(