add rpc system to improve ipc

events/EventBus.ts

@@ -1,50 +0,0 @@
-import { IpcMain, ipcMain, ipcRenderer } from 'electron'
-
-import { Event } from './Events'
-import { IpcRendererEventBus } from './IpcRendererEventBus'
-
-export interface EventBusInterface {
-  subscribe<MessageType>(event: Event<MessageType>, callback: (msg: MessageType) => void): void
-  unsubscribeAll<MessageType>(event: Event<MessageType>): void
-  emit<MessageType>(event: Event<MessageType>, msg: MessageType): void
-  unsubscribe<MessageType>(event: Event<MessageType>, callback: any): void
-}
-
-export interface CallbackStore {
-  wrappedCallback: any
-  callback: any
-}
-
-class IpcMainEventBus implements EventBusInterface {
-  private ipc: IpcMain
-  private client: any
-  constructor(ipc: IpcMain) {
-    this.ipc = ipc
-  }
-
-  public subscribe<MessageType>(subscribeEvent: Event<MessageType>, callback: (msg: MessageType) => void) {
-    console.log('subscribing', subscribeEvent.topic)
-    this.ipc.on(subscribeEvent.topic, (event: any, arg: any) => {
-      this.client = event.sender
-      callback(arg)
-    })
-  }
-
-  public unsubscribeAll<MessageType>(event: Event<MessageType>) {
-    console.log('unsubscribeAll', event.topic)
-    this.ipc.removeAllListeners(event.topic)
-  }
-
-  public unsubscribe<MessageType>(event: Event<MessageType>, callback: any) {
-    throw new Error('Not implemented') // Todo: implement
-  }
-
-  public emit<MessageType>(event: Event<MessageType>, msg: MessageType) {
-    if (!this.client.isDestroyed()) {
-      this.client.send(event.topic, msg)
-    }
-  }
-}
-
-export const rendererEvents = new IpcRendererEventBus(ipcRenderer)
-export const backendEvents = new IpcMainEventBus(ipcMain)

events/EventSystem/CallbackStore.ts

@@ -0,0 +1,4 @@
+export interface CallbackStore {
+    wrappedCallback: any;
+    callback: any;
+}

events/EventSystem/EventBus.ts

@@ -0,0 +1,12 @@
+import { ipcMain, ipcRenderer } from 'electron'
+
+import { IpcMainEventBus } from './IpcMainEventBus'
+import { IpcRendererEventBus } from './IpcRendererEventBus'
+import { Rpc } from './Rpc'
+
+export const rendererEvents = new IpcRendererEventBus(ipcRenderer)
+export const backendEvents = new IpcMainEventBus(ipcMain)
+
+// Preferred way to communicate typesafe
+export const rendererRpc = new Rpc(rendererEvents)
+export const backendRpc = new Rpc(backendEvents)

events/EventSystem/EventBusInterface.ts

@@ -0,0 +1,8 @@
+import { Event } from '../Events';
+
+export interface EventBusInterface {
+    subscribe<MessageType>(event: Event<MessageType>, callback: (msg: MessageType) => void): void;
+    unsubscribeAll<MessageType>(event: Event<MessageType>): void;
+    emit<MessageType>(event: Event<MessageType>, msg: MessageType): void;
+    unsubscribe<MessageType>(event: Event<MessageType>, callback: any): void;
+}

events/EventSystem/IpcMainEventBus.ts

@@ -0,0 +1,34 @@
+import { IpcMain } from 'electron';
+import { Event } from '../Events';
+import { EventBusInterface } from "./EventBusInterface";
+
+export class IpcMainEventBus implements EventBusInterface {
+    private ipc: IpcMain;
+    private client: any;
+    constructor(ipc: IpcMain) {
+        this.ipc = ipc;
+    }
+
+    public subscribe<MessageType>(subscribeEvent: Event<MessageType>, callback: (msg: MessageType) => void) {
+        console.log('subscribing', subscribeEvent.topic);
+        this.ipc.on(subscribeEvent.topic, (event: any, arg: any) => {
+            this.client = event.sender;
+            callback(arg);
+        });
+    }
+
+    public unsubscribeAll<MessageType>(event: Event<MessageType>) {
+        console.log('unsubscribeAll', event.topic);
+        this.ipc.removeAllListeners(event.topic);
+    }
+
+    public unsubscribe<MessageType>(event: Event<MessageType>, callback: any) {
+        throw new Error('Not implemented'); // Todo: implement
+    }
+
+    public emit<MessageType>(event: Event<MessageType>, msg: MessageType) {
+        if (!this.client.isDestroyed()) {
+            this.client.send(event.topic, msg);
+        }
+    }
+}

events/EventSystem/IpcRendererEventBus.ts

@@ -1,26 +1,32 @@
-import { CallbackStore, EventBusInterface } from './EventBus'
-import { Event } from './Events'
+import { CallbackStore } from "./CallbackStore"
+import { EventBusInterface } from "./EventBusInterface"
+import { Event } from '../Events'
 import { IpcRenderer } from 'electron'
 
 export class IpcRendererEventBus implements EventBusInterface {
   private ipc: IpcRenderer
   private callbacks: Array<CallbackStore> = []
+
   constructor(ipc: IpcRenderer) {
     this.ipc = ipc
   }
+
   public subscribe<MessageType>(event: Event<MessageType>, callback: (msg: MessageType) => void) {
     const wrappedCallback = (_: any, arg: any) => {
       callback(arg)
     }
+    console.log("subscribing", event.topic)
     this.ipc.on(event.topic, wrappedCallback)
     this.callbacks.push({
       callback,
       wrappedCallback,
     })
   }
+
   public unsubscribeAll<MessageType>(event: Event<MessageType>) {
     this.ipc.removeAllListeners(event.topic)
   }
+
   public unsubscribe<MessageType>(event: Event<MessageType>, callback: any) {
     const item = this.callbacks.find(store => store.callback === callback)
     if (!item) {
@@ -29,6 +35,7 @@ export class IpcRendererEventBus implements EventBusInterface {
     this.ipc.removeListener(event.topic, item.wrappedCallback)
     this.callbacks = this.callbacks.filter(a => a !== item)
   }
+
   public emit<MessageType>(event: Event<MessageType>, msg: MessageType) {
     this.ipc.send(event.topic, msg)
   }

events/EventSystem/Rpc.ts

@@ -0,0 +1,54 @@
+import { Event } from '../Events';
+import { EventBusInterface } from './EventBusInterface'
+import { v4 } from 'uuid';
+
+export type RpcEvent<RequstType, ResponseType> = {
+    topic: string;
+};
+
+export class Rpc {
+    constructor(private participant: EventBusInterface) { }
+
+    async call<RpcRequest, RpcResponse>(event: RpcEvent<RpcRequest, RpcResponse>, request: RpcRequest, timeout: number = 0): Promise<RpcResponse> {
+        return new Promise((resolve, reject) => {
+            let id = v4();
+
+            let responseEvent: Event<any> = { topic: `${event.topic}/response` };
+            let requestEvent: Event<any> = { topic: `${event.topic}/request` };
+            let callback = (result: { id: string; payload: RpcResponse; error: unknown }) => {
+                this.participant.unsubscribe(responseEvent as any, callback);
+                if (result.id === id) {
+                    if (result.error) {
+                        reject(result.error)
+                    } else {
+                        resolve(result.payload);
+                    }
+                }
+                console.log("received", result)
+            };
+            this.participant.subscribe(responseEvent, callback);
+            this.participant.emit(requestEvent, { id, payload: request });
+
+            if (timeout > 0) {
+                setTimeout(() => {
+                    reject(new Error(`Did not respond to ${event.topic} within ${timeout}ms`))
+                    this.participant.unsubscribe(responseEvent as any, callback);
+                }, 10000)
+            }
+        });
+    }
+
+    async on<RpcRequest, RpcResponse>(event: RpcEvent<RpcRequest, RpcResponse>, handler: (request: RpcRequest) => Promise<RpcResponse>) {
+        this.participant.subscribe<RpcRequest>({ topic: `${event.topic}/request` } as RpcEvent<any, any>, async (request) => {
+            let payload;
+            let error;
+            try {
+                payload = await handler((request as any).payload);
+            } catch (e) {
+                error = e
+            }
+            console.log("Responding with", payload, error)
+            this.participant.emit({ topic: `${event.topic}/response` }, { id: (request as any).id, payload, error });
+        });
+    }
+}

events/Events.ts

@@ -1,7 +1,7 @@
 import { Base64Message } from '../backend/src/Model/Base64Message'
 import { DataSourceState, MqttOptions } from '../backend/src/DataSource'
 import { UpdateInfo } from 'builder-util-runtime'
-import { OpenDialogOptions, OpenDialogReturnValue } from 'electron'
+import { RpcEvent } from './EventSystem/Rpc';
 
 export type Event<MessageType> = {
   topic: string
@@ -51,25 +51,6 @@ export function makeConnectionMessageEvent(connectionId: string): Event<MqttMess
   }
 }
 
-export interface OpenDialogRequest {
-  identifier: string
-  options: OpenDialogOptions
+export const getAppVersion: RpcEvent<void, string> = {
+  topic: `getAppVersion`
 }
-
-export function requestOpenDialog(): Event<OpenDialogRequest> {
-  return {
-    topic: `requestOpenDialog`,
-  }
-}
-
-export interface OpenDialogResponse {
-  identifier: string
-  result: OpenDialogReturnValue
-}
-
-export function openDialogResponse(): Event<OpenDialogResponse> {
-  return {
-    topic: `openDialogResponse`,
-  }
-}
-

events/OpenDialogRequest.ts

@@ -0,0 +1,8 @@
+import { OpenDialogOptions, OpenDialogReturnValue } from 'electron';
+import { RpcEvent } from './EventSystem/Rpc';
+
+export function makeOpenDialogRpc(): RpcEvent<OpenDialogOptions, OpenDialogReturnValue> {
+    return {
+        topic: `openDialog`
+    };
+}

events/StorageEvents.ts

@@ -1,39 +1,22 @@
-import { Event } from './'
+import { RpcEvent } from './EventSystem/Rpc'
 
-interface StorageEvent {
-  transactionId: string
+export interface StoreCommand {
+  store: string
+  data: any
 }
 
-export interface StoreCommand extends StorageEvent {
-  store?: string
-  data?: any
-  error?: any
-}
-
-export interface LoadCommand extends StorageEvent {
+export interface LoadCommand {
   store: string
 }
 
-export const storageStoreEvent: Event<StoreCommand> = {
+export const storageStoreEvent: RpcEvent<StoreCommand, void> = {
   topic: 'storage/store',
 }
 
-export const storageLoadEvent: Event<LoadCommand> = {
+export const storageLoadEvent: RpcEvent<LoadCommand, StoreCommand> = {
   topic: 'storage/load',
 }
 
-export function makeStorageAcknowledgementEvent(transactionId: string): Event<StoreCommand> {
-  return {
-    topic: `storage/ack/${transactionId}`,
-  }
-}
-
-export function makeStorageResponseEvent(transactionId: string): Event<StoreCommand> {
-  return {
-    topic: `storage/response/${transactionId}`,
-  }
-}
-
-export const storageClearEvent: Event<StorageEvent> = {
+export const storageClearEvent: RpcEvent<void, void> = {
   topic: 'storage/clear',
 }

events/index.ts

@@ -1,3 +1,4 @@
 export * from './Events'
-export * from './EventDispatcher'
-export * from './EventBus'
+export * from './EventSystem/EventDispatcher'
+export * from './EventSystem/EventBus'
+export * from './EventSystem/EventBusInterface'