Add support for subscriptions

.changeset/beige-islands-type.md

@@ -0,0 +1,5 @@
+---
+'electron-trpc': minor
+---
+
+Added support for subscriptions.

README.md

@@ -19,7 +19,7 @@
 - Expose APIs from Electron's main process to one or more render processes.
 - Build fully type-safe IPC.
 - Secure alternative to opening servers on localhost.
-- _Subscription support coming soon_.
+- Full support for queries, mutations, and subscriptions.
 
 ## Installation
 
@@ -44,14 +44,14 @@ npm install --save electron-trpc
    import { router } from './api';
 
    app.on('ready', () => {
-     createIPCHandler({ router });
-
      const win = new BrowserWindow({
        webPreferences: {
          // Replace this path with the path to your preload file (see next step)
          preload: 'path/to/preload.js',
        },
      });
+
+     createIPCHandler({ router, windows: [win] });
    });
    ```
 

examples/basic/electron/api.ts

@@ -1,16 +1,34 @@
 import z from 'zod';
 import { initTRPC } from '@trpc/server';
+import { observable } from '@trpc/server/observable';
+import { EventEmitter } from 'events';
+
+const ee = new EventEmitter();
 
 const t = initTRPC.create({ isServer: true });
 
 export const router = t.router({
   greeting: t.procedure.input(z.object({ name: z.string() })).query((req) => {
     const { input } = req;
 
+    ee.emit('greeting', `Greeted ${input.name}`);
     return {
       text: `Hello ${input.name}` as const,
     };
   }),
+  subscription: t.procedure.subscription(() => {
+    return observable((emit) => {
+      function onGreet(text: string) {
+        emit.next({ text });
+      }
+
+      ee.on('greeting', onGreet);
+
+      return () => {
+        ee.off('greeting', onGreet);
+      };
+    });
+  }),
 });
 
 export type AppRouter = typeof router;

examples/basic/electron/index.ts

@@ -10,14 +10,14 @@ const preload = path.join(__dirname, './preload.js');
 const url = process.env['VITE_DEV_SERVER_URL'];
 
 app.on('ready', () => {
-  createIPCHandler({ router });
-
   const win = new BrowserWindow({
     webPreferences: {
       preload,
     },
   });
 
+  createIPCHandler({ router, windows: [win] });
+
   if (url) {
     win.loadURL(url);
   } else {

examples/basic/src/index.tsx

@@ -26,6 +26,11 @@ function App() {
 
 function HelloElectron() {
   const { data } = trpcReact.greeting.useQuery({ name: 'Electron' });
+  trpcReact.subscription.useSubscription(undefined, {
+    onData: (data) => {
+      console.log(data);
+    },
+  });
 
   if (!data) {
     return null;

examples/basic/tsconfig.json

@@ -6,7 +6,7 @@
     "jsx": "react",
     "lib": ["dom", "esnext"],
     "module": "esnext",
-    "moduleResolution": "node",
+    "moduleResolution": "node16",
     "noEmit": true,
     "noFallthroughCasesInSwitch": true,
     "noUnusedLocals": true,

packages/electron-trpc/src/main/__tests__/handleIPCOperation.test.ts

@@ -0,0 +1,83 @@
+import { describe, expect, test, vi } from 'vitest';
+import { z } from 'zod';
+import * as trpc from '@trpc/server';
+import { observable } from '@trpc/server/observable';
+import { EventEmitter } from 'events';
+import { handleIPCOperation } from '../handleIPCOperation';
+
+const ee = new EventEmitter();
+
+const t = trpc.initTRPC.create();
+const testRouter = t.router({
+  testQuery: t.procedure
+    .input(
+      z.object({
+        id: z.string(),
+      })
+    )
+    .query(({ input }) => {
+      return { id: input.id, isTest: true };
+    }),
+  testSubscription: t.procedure.subscription(() => {
+    return observable((emit) => {
+      function testResponse() {
+        emit.next('test response');
+      }
+
+      ee.on('test', testResponse);
+      return () => ee.off('test', testResponse);
+    });
+  }),
+});
+
+describe('api', () => {
+  test('can manually call into API', async () => {
+    const respond = vi.fn();
+    await handleIPCOperation({
+      createContext: async () => ({}),
+      operation: { context: {}, id: 1, input: { id: 'test-id' }, path: 'testQuery', type: 'query' },
+      router: testRouter,
+      respond,
+    });
+
+    expect(respond).toHaveBeenCalledOnce();
+    expect(respond.mock.lastCall[0]).toMatchObject({
+      id: 1,
+      result: {
+        data: {
+          id: 'test-id',
+          isTest: true,
+        },
+      },
+    });
+  });
+
+  test('does not handle subscriptions', async () => {
+    const respond = vi.fn();
+
+    await handleIPCOperation({
+      createContext: async () => ({}),
+      operation: {
+        context: {},
+        id: 1,
+        input: undefined,
+        path: 'testSubscription',
+        type: 'subscription',
+      },
+      router: testRouter,
+      respond,
+    });
+
+    expect(respond).not.toHaveBeenCalled();
+
+    ee.emit('test');
+
+    expect(respond).toHaveBeenCalledOnce();
+    expect(respond.mock.lastCall[0]).toMatchObject({
+      id: 1,
+      result: {
+        data: 'test response',
+      },
+    });
+  });
+});

packages/electron-trpc/src/main/createIPCHandler.ts

@@ -1,18 +1,58 @@
 import type { Operation } from '@trpc/client';
 import type { AnyRouter, inferRouterContext } from '@trpc/server';
+import type { TRPCResponseMessage } from '@trpc/server/rpc';
 import { ipcMain } from 'electron';
-import type { IpcMainInvokeEvent } from 'electron';
-import { resolveIPCResponse } from './resolveIPCResponse';
+import type { BrowserWindow, IpcMainInvokeEvent } from 'electron';
+import { handleIPCOperation } from './handleIPCOperation';
 import { ELECTRON_TRPC_CHANNEL } from '../constants';
 
-export function createIPCHandler<TRouter extends AnyRouter>({
+class IPCHandler<TRouter extends AnyRouter> {
+  #windows: BrowserWindow[];
+
+  constructor({
+    createContext,
+    router,
+    windows = [],
+  }: {
+    createContext?: () => Promise<inferRouterContext<TRouter>>;
+    router: TRouter;
+    windows?: BrowserWindow[];
+  }) {
+    this.#windows = windows;
+
+    ipcMain.on(ELECTRON_TRPC_CHANNEL, (_event: IpcMainInvokeEvent, args: Operation) => {
+      handleIPCOperation({
+        router,
+        createContext,
+        operation: args,
+        respond: (response) => this.#sendToAllWindows(response),
+      });
+    });
+  }
+
+  #sendToAllWindows(response: TRPCResponseMessage) {
+    this.#windows.forEach((win) => {
+      win.webContents.send(ELECTRON_TRPC_CHANNEL, response);
+    });
+  }
+
+  attachWindow(win: BrowserWindow) {
+    this.#windows.push(win);
+  }
+
+  detachWindow(win: BrowserWindow) {
+    this.#windows = this.#windows.filter((w) => w !== win);
+  }
+}
+
+export const createIPCHandler = <TRouter extends AnyRouter>({
   createContext,
   router,
+  windows = [],
 }: {
   createContext?: () => Promise<inferRouterContext<TRouter>>;
   router: TRouter;
-}) {
-  ipcMain.handle(ELECTRON_TRPC_CHANNEL, (_event: IpcMainInvokeEvent, args: Operation) => {
-    return resolveIPCResponse({ router, createContext, operation: args });
-  });
-}
+  windows?: Electron.BrowserWindow[];
+}) => {
+  return new IPCHandler({ createContext, router, windows });
+};

packages/electron-trpc/src/main/exposeElectronTRPC.ts

@@ -1,9 +1,11 @@
 import type { Operation } from '@trpc/client';
+import type { TRPCResponseMessage } from '@trpc/server/rpc';
 import { ipcRenderer, contextBridge } from 'electron';
 import { ELECTRON_TRPC_CHANNEL } from '../constants';
 
 export const exposeElectronTRPC = () => {
   contextBridge.exposeInMainWorld('electronTRPC', {
-    rpc: (args: Operation) => ipcRenderer.invoke(ELECTRON_TRPC_CHANNEL, args),
+    sendMessage: (args: Operation) => ipcRenderer.send(ELECTRON_TRPC_CHANNEL, args),
+    onMessage: (callback: (args: TRPCResponseMessage) => void) => ipcRenderer.on(ELECTRON_TRPC_CHANNEL, (_event, args) => callback(args)),
   });
 };