Introduce abstraction for publishing and subscribing between processes

Co-authored-by: Mikko Aspiala <mikko.aspiala@gmail.com>

Signed-off-by: Janne Savolainen <janne.savolainen@live.fi>

src/common/communication-between-processes/publish-to-channel-injection-token.ts

@@ -0,0 +1,16 @@
+/**
+ * Copyright (c) OpenLens Authors. All rights reserved.
+ * Licensed under MIT License. See LICENSE in root directory for more information.
+ */
+import { getInjectionToken } from "@ogre-tools/injectable";
+import type { Channel } from "../ipc-channel/channel";
+
+export type PublishToChannel = <TChannel extends Channel<TMessage>, TMessage>(
+  channel: TChannel,
+  message: TChannel["_template"]
+) => void;
+
+export const publishToChannelInjectionToken =
+  getInjectionToken<PublishToChannel>({
+    id: "publish-to-channel",
+  });

src/common/communication-between-processes/subscribe-to-channel-fake.test.ts

@@ -0,0 +1,97 @@
+/**
+ * Copyright (c) OpenLens Authors. All rights reserved.
+ * Licensed under MIT License. See LICENSE in root directory for more information.
+ */
+import type { DiContainer } from "@ogre-tools/injectable";
+import { getDisForUnitTesting } from "../../test-utils/get-dis-for-unit-testing";
+import {
+  SubscribeToChannel,
+  subscribeToChannelInjectionToken,
+} from "./subscribe-to-channel-injection-token";
+import { PublishToChannel, publishToChannelInjectionToken } from "./publish-to-channel-injection-token";
+import { createChannel } from "../ipc-channel/create-channel/create-channel";
+import type { Channel } from "../ipc-channel/channel";
+
+describe("subscribe-to-channel-fake, given IPC bridge", () => {
+  let mainDi: DiContainer;
+  let rendererDi: DiContainer;
+
+  beforeEach(() => {
+    const dis = getDisForUnitTesting();
+
+    mainDi = dis.mainDi;
+    rendererDi = dis.rendererDi;
+  });
+
+  [
+    { scenario: "given in renderer", getDiForPublish: () => mainDi, getDiForSubscribe: () => rendererDi },
+    { scenario: "given in main", getDiForPublish: () => rendererDi, getDiForSubscribe: () => mainDi },
+  ].forEach(({ scenario, getDiForPublish, getDiForSubscribe }) => {
+    describe(scenario, () => {
+      let publishToChannel: PublishToChannel;
+      let subscribeToChannel: SubscribeToChannel;
+      let channel: Channel<string>;
+      let diForPublish: DiContainer;
+      let diForSubscribe: DiContainer;
+
+      beforeEach(() => {
+        diForPublish = getDiForPublish();
+        diForSubscribe = getDiForSubscribe();
+
+        publishToChannel = diForPublish.inject(publishToChannelInjectionToken);
+        subscribeToChannel = diForSubscribe.inject(subscribeToChannelInjectionToken);
+
+        channel = createChannel("some-channel");
+      });
+
+      it("given no subscribers, when publishing message to channel, throws", () => {
+        expect(() => {
+          publishToChannel(channel, "some-message");
+        }).toThrow('Tried to publish message "some-message" to channel "some-channel" when there is no subscribers.');
+      });
+
+      describe("when subscribing", () => {
+        let subscriberMock: jest.Mock<(message: string) => void>;
+
+        beforeEach(() => {
+          subscriberMock = jest.fn();
+
+          subscribeToChannel(channel, subscriberMock);
+        });
+
+        it("does not call subscribers yet", () => {
+          expect(subscriberMock).not.toHaveBeenCalled();
+        });
+
+        describe("when publishing to channel", () => {
+          beforeEach(() => {
+            publishToChannel(channel, "some-message");
+          });
+
+          it("notifies the subscribers", () => {
+            expect(subscriberMock).toHaveBeenCalledWith("some-message");
+          });
+
+          it("when published again, notifies again", () => {
+            subscriberMock.mockClear();
+
+            publishToChannel(channel, "some-other-message");
+
+            expect(subscriberMock).toHaveBeenCalledWith("some-other-message");
+          });
+        });
+      });
+
+      it("given multiple subscribers, when publishing, notifies all subscribers", () => {
+        const subscriberMock = jest.fn();
+
+        subscribeToChannel(channel, subscriberMock);
+        subscribeToChannel(channel, subscriberMock);
+
+        publishToChannel(channel, "some-message");
+
+        expect(subscriberMock).toHaveBeenCalledTimes(2);
+      });
+    });
+  });
+});

src/common/communication-between-processes/subscribe-to-channel-injection-token.ts

@@ -0,0 +1,15 @@
+/**
+ * Copyright (c) OpenLens Authors. All rights reserved.
+ * Licensed under MIT License. See LICENSE in root directory for more information.
+ */
+import { getInjectionToken } from "@ogre-tools/injectable";
+import type { Channel } from "../ipc-channel/channel";
+
+export type SubscribeToChannel = <TChannel extends Channel<TMessage>, TMessage>(
+  channel: TChannel,
+  callback: (message: TChannel["_template"]) => void
+) => void;
+
+export const subscribeToChannelInjectionToken = getInjectionToken<SubscribeToChannel>({
+  id: "subscribe-to-channel",
+});

src/common/communication-between-processes/subscribe-to-channel.test.ts

@@ -0,0 +1,154 @@
+/**
+ * Copyright (c) OpenLens Authors. All rights reserved.
+ * Licensed under MIT License. See LICENSE in root directory for more information.
+ */
+import { getDiForUnitTesting as getMainDi } from "../../main/getDiForUnitTesting";
+import { getDiForUnitTesting as getRendererDi } from "../../renderer/getDiForUnitTesting";
+import ipcMainInjectable from "../../main/app-paths/register-channel/ipc-main/ipc-main.injectable";
+import ipcRendererInjectable from "../../renderer/app-paths/get-value-from-registered-channel/ipc-renderer/ipc-renderer.injectable";
+import type { DiContainer } from "@ogre-tools/injectable";
+import { subscribeToChannelInjectionToken } from "./subscribe-to-channel-injection-token";
+import { createChannel } from "../ipc-channel/create-channel/create-channel";
+import type { IpcMain, IpcRenderer, WebContents } from "electron";
+import type { IpcMainEvent, IpcRendererEvent } from "../../extensions/common-api/types";
+import { publishToChannelInjectionToken } from "./publish-to-channel-injection-token";
+import webContentsInjectable from "../../main/communication-between-processes/web-contents/web-contents.injectable";
+
+describe("subscribe-to-channel", () => {
+  let mainDi: DiContainer;
+  let rendererDi: DiContainer;
+  let ipcMainStub: IpcMain;
+
+  beforeEach(() => {
+    mainDi = getMainDi({ doGeneralOverrides: true });
+    rendererDi = getRendererDi({ doGeneralOverrides: true });
+
+    const mainSubscriberMap = new Map<string|symbol, ((nativeEvent: IpcMainEvent, ...args: any[]) => void)[]>();
+    const rendererSubscriberMap = new Map<string|symbol, ((nativeEvent: IpcRendererEvent, ...args: any[]) => void)[]>();
+
+    ipcMainStub = {
+      on: (channelName, subscriber) => {
+        const subscribers = mainSubscriberMap.get(channelName) || [];
+
+        mainSubscriberMap.set(channelName, [...subscribers, subscriber]);
+      },
+
+      listeners: (channelName) => {
+        const listeners: Function[] = mainSubscriberMap.get(channelName);
+
+        return listeners;
+      },
+    } as IpcMain;
+
+    const ipcRendererStub = {
+      send: (channelName, message) => {
+        const subscribers = mainSubscriberMap.get(channelName);
+
+        const ipcMainEventStub = {} as IpcMainEvent;
+
+        subscribers.forEach(subscriber => {
+          subscriber(
+            ipcMainEventStub,
+            message,
+          );
+        });
+      },
+
+      on: (channelName, subscriber) => {
+        const subscribers = rendererSubscriberMap.get(channelName) || [];
+
+        rendererSubscriberMap.set(channelName, [...subscribers, subscriber]);
+      },
+    } as IpcRenderer;
+
+    const webContentsStub = {
+      getAllWebContents: () => [
+        { isDestroyed: () => true },
+
+        {
+          isDestroyed: () => false,
+
+          send: (channelName: string, message: any) => {
+            const subscribers = rendererSubscriberMap.get(channelName);
+
+            subscribers.forEach(subscriber => subscriber(null, message));
+          },
+        },
+      ],
+    } as unknown as typeof WebContents;
+
+    mainDi.override(webContentsInjectable, () => webContentsStub);
+    mainDi.override(ipcMainInjectable, () => ipcMainStub);
+    rendererDi.override(ipcRendererInjectable, () => ipcRendererStub);
+  });
+
+  it("when publishing message from renderer, notifies subscribers in all environments", () => {
+    const subscribeInRenderer = rendererDi.inject(subscribeToChannelInjectionToken);
+    const subscribeInMain = mainDi.inject(subscribeToChannelInjectionToken);
+
+    const someChannel = createChannel("some-channel");
+    const callbackMock = jest.fn();
+
+    subscribeInRenderer(someChannel, (message) => callbackMock("in-renderer", message));
+    subscribeInMain(someChannel, (message) => callbackMock("in-main", message));
+
+    const publishFromRenderer = rendererDi.inject(publishToChannelInjectionToken);
+
+    publishFromRenderer(someChannel, "some-message");
+
+    expect(callbackMock.mock.calls).toEqual([
+      ["in-main", "some-message"],
+      ["in-renderer", "some-message"],
+    ]);
+  });
+
+  it("when publishing message from main, notifies subscribers in all environments", () => {
+    const subscribeInRenderer = rendererDi.inject(subscribeToChannelInjectionToken);
+    const subscribeInMain = mainDi.inject(subscribeToChannelInjectionToken);
+
+    const someChannel = createChannel("some-channel");
+    const callbackMock = jest.fn();
+
+    subscribeInRenderer(someChannel, (message) => callbackMock("in-renderer", message));
+    subscribeInMain(someChannel, (message) => callbackMock("in-main", message));
+
+    const publishFromMain = mainDi.inject(publishToChannelInjectionToken);
+
+    publishFromMain(someChannel, "some-message");
+
+    expect(callbackMock.mock.calls).toEqual([
+      ["in-main", "some-message"],
+      ["in-renderer", "some-message"],
+    ]);
+  });
+
+  it("given subscribing from main, when publishing message from renderer, notifies the subscriber in main", () => {
+    const subscribeToChannel = mainDi.inject(subscribeToChannelInjectionToken);
+    const publishToChannel = rendererDi.inject(publishToChannelInjectionToken);
+
+    const channel = createChannel("some-channel");
+
+    const callbackMock = jest.fn();
+
+    subscribeToChannel(channel, callbackMock);
+
+    publishToChannel(channel, "some-message");
+
+    expect(callbackMock).toHaveBeenCalledWith("some-message");
+  });
+
+  it("given subscribing from main, when publishing message from main, notifies the subscriber in main", () => {
+    const subscribeToChannel = mainDi.inject(subscribeToChannelInjectionToken);
+    const publishToChannel = mainDi.inject(publishToChannelInjectionToken);
+
+    const channel = createChannel("some-channel");
+
+    const callbackMock = jest.fn();
+
+    subscribeToChannel(channel, callbackMock);
+
+    publishToChannel(channel, "some-message");
+
+    expect(callbackMock).toHaveBeenCalledWith("some-message");
+  });
+});

src/main/communication-between-processes/publish-to-channel.injectable.ts

@@ -0,0 +1,61 @@
+/**
+ * Copyright (c) OpenLens Authors. All rights reserved.
+ * Licensed under MIT License. See LICENSE in root directory for more information.
+ */
+import { pipeline } from "@ogre-tools/fp";
+import { getInjectable } from "@ogre-tools/injectable";
+import type { IpcMain, IpcRendererEvent, WebContents } from "electron";
+import { filter, forEach } from "lodash/fp";
+import { publishToChannelInjectionToken } from "../../common/communication-between-processes/publish-to-channel-injection-token";
+import ipcMainInjectable from "../app-paths/register-channel/ipc-main/ipc-main.injectable";
+import webContentsInjectable from "./web-contents/web-contents.injectable";
+import type { Channel } from "../../common/ipc-channel/channel";
+
+const publishToChannelInjectable = getInjectable({
+  id: "publish-to-channel",
+
+  instantiate: (di) => {
+    const publishInMain = publishInMainFor(di.inject(ipcMainInjectable));
+    const publishInRenderer = publishInRendererFor(
+      di.inject(webContentsInjectable),
+    );
+
+    return (channel, message) => {
+      publishInMain(channel, message);
+
+      publishInRenderer(channel, message);
+    };
+  },
+
+  injectionToken: publishToChannelInjectionToken,
+});
+
+export default publishToChannelInjectable;
+
+const publishInMainFor =
+  (ipcMain: IpcMain) =>
+  <TChannel extends Channel<unknown>>(
+      channel: TChannel,
+      message: TChannel["_template"],
+    ) => {
+    const listeners = ipcMain.listeners(channel.name) as ((
+      nativeEvent: IpcRendererEvent,
+      ...args: any[]
+    ) => void)[];
+
+    listeners.forEach((listener) => listener(null, message));
+  };
+
+const publishInRendererFor =
+  (webContents: typeof WebContents) =>
+  <TChannel extends Channel<unknown>>(
+      channel: TChannel,
+      message: TChannel["_template"],
+    ) => {
+    pipeline(
+      webContents.getAllWebContents(),
+      filter((view) => view.isDestroyed() === false),
+
+      (views) => forEach((view) => view.send(channel.name, message), views),
+    );
+  };

src/main/communication-between-processes/subscribe-to-channel.injectable.ts

@@ -0,0 +1,28 @@
+/**
+ * Copyright (c) OpenLens Authors. All rights reserved.
+ * Licensed under MIT License. See LICENSE in root directory for more information.
+ */
+import { getInjectable } from "@ogre-tools/injectable";
+import { subscribeToChannelInjectionToken } from "../../common/communication-between-processes/subscribe-to-channel-injection-token";
+import ipcMainInjectable from "../app-paths/register-channel/ipc-main/ipc-main.injectable";
+import type { Channel } from "../../common/ipc-channel/channel";
+
+const subscribeToChannelInjectable = getInjectable({
+  id: "subscribe-to-channel",
+
+  instantiate: (di) => {
+    const ipcMain = di.inject(ipcMainInjectable);
+
+    return <TChannel extends Channel<TMessage>, TMessage>(channel: TChannel, callback: (message: TChannel["_template"]) => void) => {
+      ipcMain.on(channel.name, (nonUsedNativeEvent, message: unknown) => {
+        const channelMessage = message as TChannel["_template"];
+
+        return callback(channelMessage);
+      });
+    };
+  },
+
+  injectionToken: subscribeToChannelInjectionToken,
+});
+
+export default subscribeToChannelInjectable;

src/main/communication-between-processes/web-contents/web-contents.injectable.ts

@@ -0,0 +1,14 @@
+/**
+ * Copyright (c) OpenLens Authors. All rights reserved.
+ * Licensed under MIT License. See LICENSE in root directory for more information.
+ */
+import { getInjectable } from "@ogre-tools/injectable";
+import { webContents } from "electron";
+
+const webContentsInjectable = getInjectable({
+  id: "web-contents",
+  instantiate: () => webContents,
+  causesSideEffects: true,
+});
+
+export default webContentsInjectable;

src/renderer/communication-between-processes/publish-to-channel.injectable.ts

@@ -0,0 +1,23 @@
+/**
+ * Copyright (c) OpenLens Authors. All rights reserved.
+ * Licensed under MIT License. See LICENSE in root directory for more information.
+ */
+import { getInjectable } from "@ogre-tools/injectable";
+import { publishToChannelInjectionToken } from "../../common/communication-between-processes/publish-to-channel-injection-token";
+import ipcRendererInjectable from "../app-paths/get-value-from-registered-channel/ipc-renderer/ipc-renderer.injectable";
+
+const publishToChannelInjectable = getInjectable({
+  id: "publish-to-channel",
+
+  instantiate: (di) => {
+    const ipcRenderer = di.inject(ipcRendererInjectable);
+
+    return (channel, message) => {
+      ipcRenderer.send(channel.name, message);
+    };
+  },
+
+  injectionToken: publishToChannelInjectionToken,
+});
+
+export default publishToChannelInjectable;

src/renderer/communication-between-processes/subscribe-to-channel.injectable.ts

@@ -0,0 +1,28 @@
+/**
+ * Copyright (c) OpenLens Authors. All rights reserved.
+ * Licensed under MIT License. See LICENSE in root directory for more information.
+ */
+import { getInjectable } from "@ogre-tools/injectable";
+import { subscribeToChannelInjectionToken } from "../../common/communication-between-processes/subscribe-to-channel-injection-token";
+import type { Channel } from "../../common/ipc-channel/channel";
+import ipcRendererInjectable from "../app-paths/get-value-from-registered-channel/ipc-renderer/ipc-renderer.injectable";
+
+const subscribeToChannelInjectable = getInjectable({
+  id: "subscribe-to-channel",
+
+  instantiate: (di) => {
+    const ipcRenderer = di.inject(ipcRendererInjectable);
+
+    return <TChannel extends Channel<TMessage>, TMessage>(channel: TChannel, callback: (message: TChannel["_template"]) => void) => {
+      ipcRenderer.on(channel.name, (nonUsedNativeEvent, message: unknown) => {
+        const channelMessage = message as TChannel["_template"];
+
+        return callback(channelMessage);
+      });
+    };
+  },
+
+  injectionToken: subscribeToChannelInjectionToken,
+});
+
+export default subscribeToChannelInjectable;

src/test-utils/override-ipc-bridge.ts

@@ -11,6 +11,9 @@ import registerIpcChannelListenerInjectable from "../renderer/app-paths/get-valu
 import windowManagerInjectable from "../main/window-manager.injectable";
 import type { SendToViewArgs, WindowManager } from "../main/window-manager";
 import { appNavigationIpcChannel } from "../common/front-end-routing/navigation-ipc-channel";
+import subscribeToChannelInjectable from "../renderer/communication-between-processes/subscribe-to-channel.injectable";
+import publishToChannelInjectable from "../main/communication-between-processes/publish-to-channel.injectable";
+import { isEmpty } from "lodash/fp";
 
 export const overrideIpcBridge = ({
   rendererDi,
@@ -80,6 +83,8 @@ export const overrideIpcBridge = ({
       },
   );
 
+  overridePublishAndSubscribe(mainDi, rendererDi);
+
   mainDi.override(
     windowManagerInjectable,
     () =>
@@ -97,3 +102,39 @@ export const overrideIpcBridge = ({
       } as unknown as WindowManager),
   );
 };
+
+const overridePublishAndSubscribe = (mainDi: DiContainer, rendererDi: DiContainer) => {
+  [
+    [mainDi, rendererDi],
+    [rendererDi, mainDi],
+  ].forEach(([diForSubscribe, diForPublish]) => {
+    const fakeSubscriptionsMap = new Map<
+      Channel<unknown>,
+      ((message: any) => void)[]
+    >();
+
+    diForSubscribe.override(
+      subscribeToChannelInjectable,
+      () => (channel, subscriber) => {
+        const subscribers = fakeSubscriptionsMap.get(channel) || [];
+
+        fakeSubscriptionsMap.set(channel, [...subscribers, subscriber]);
+      },
+    );
+
+    diForPublish.override(
+      publishToChannelInjectable,
+      () => (channel, message) => {
+        const subscribers = fakeSubscriptionsMap.get(channel);
+
+        if (isEmpty(subscribers)) {
+          throw new Error(
+            `Tried to publish message "${message}" to channel "${channel.name}" when there is no subscribers.`,
+          );
+        }
+
+        subscribers.forEach((subscriber) => subscriber(message));
+      },
+    );
+  });
+};