#108 - implement IPC stream

src/main/event/main-event-service.ts

@@ -8,6 +8,7 @@ import { PersistenceService } from '../persistence/service/persistence-service';
 import { RufusObject } from 'shim/objects';
 import * as console from 'node:console';
 import { EnvironmentService } from 'main/environment/service/environment-service';
+import './stream-events';
 
 const persistenceService = PersistenceService.instance;
 const environmentService = EnvironmentService.instance;
@@ -60,7 +61,7 @@ function toError(error: unknown) {
 export class MainEventService implements IEventService {
   public static readonly instance = new MainEventService();
 
-  constructor() {
+  private constructor() {
     for (const propertyName of Reflect.ownKeys(MainEventService.prototype)) {
       registerEvent(this, propertyName as keyof MainEventService);
     }

src/main/event/stream-events.ts

@@ -0,0 +1,32 @@
+import { ipcMain } from 'electron';
+import { createReadStream, ReadStream } from 'node:fs';
+
+let nextId = 0;
+
+const streams = new Map<number, ReadStream>();
+
+ipcMain.handle('stream-open', (event, filePath: string) => {
+  const { sender } = event;
+  const stream = createReadStream(filePath, 'utf8');
+  const id = nextId++;
+  streams.set(id, stream);
+
+  stream.on('data', (chunk: string) => {
+    sender.send('stream-data', id, chunk);
+  });
+
+  stream.on('end', () => {
+    sender.send('stream-end', id);
+  });
+
+  stream.on('error', (error) => {
+    sender.send('stream-error', id, error);
+  });
+
+  return id;
+});
+
+ipcMain.on('stream-close', (event, id: number) => {
+  streams.get(id)?.close();
+  streams.delete(id);
+});

src/main/preload.ts

@@ -4,11 +4,14 @@ import { contextBridge, ipcRenderer } from 'electron';
 
 const electronHandler = {
   ipcRenderer: {
-    send: ipcRenderer.send,
-    on: ipcRenderer.on,
-    once: ipcRenderer.once,
-    invoke: ipcRenderer.invoke,
-    removeListener: ipcRenderer.removeListener,
+    send: ipcRenderer.send.bind(ipcRenderer) as typeof ipcRenderer.send,
+    sendSync: ipcRenderer.sendSync.bind(ipcRenderer) as typeof ipcRenderer.sendSync,
+    on: ipcRenderer.on.bind(ipcRenderer) as typeof ipcRenderer.on,
+    once: ipcRenderer.once.bind(ipcRenderer) as typeof ipcRenderer.once,
+    invoke: ipcRenderer.invoke.bind(ipcRenderer) as typeof ipcRenderer.invoke,
+    removeListener: ipcRenderer.removeListener.bind(
+      ipcRenderer
+    ) as typeof ipcRenderer.removeListener,
   },
 };
 

src/renderer/lib/event-emitter.ts

@@ -0,0 +1,25 @@
+export type EventListener = (...args: unknown[]) => void;
+
+export abstract class EventEmitter {
+  private readonly listeners = new Map<string, Set<EventListener>>();
+
+  on(event: string, listener: EventListener) {
+    if (!this.listeners.has(event)) {
+      this.listeners.set(event, new Set());
+    }
+    this.listeners.get(event).add(listener);
+    return this;
+  }
+
+  off(event: string, listener: EventListener) {
+    this.listeners.get(event)?.delete(listener);
+    return this;
+  }
+
+  protected emit(event: string, ...args: unknown[]) {
+    for (const listener of this.listeners.get(event) ?? []) {
+      listener(...args);
+    }
+    return this;
+  }
+}

src/renderer/lib/ipc-stream.ts

@@ -0,0 +1,45 @@
+import { EventEmitter } from '@/lib/event-emitter';
+
+const { ipcRenderer } = window.electron;
+
+const streams = new Map<number, IpcPushStream>();
+
+export interface IpcPushStream {
+  on(event: 'data', listener: (chunk: string) => void): this;
+
+  on(event: 'end', listener: () => void): this;
+
+  on(event: 'error', listener: (error: Error) => void): this;
+}
+
+export class IpcPushStream extends EventEmitter {
+  static {
+    ipcRenderer.on('stream-data', (event, id: number, chunk: string) => {
+      streams.get(id)?.emit('data', chunk);
+    });
+
+    ipcRenderer.on('stream-end', (event, id: number) => {
+      streams.get(id)?.emit('end');
+      streams.delete(id);
+    });
+
+    ipcRenderer.on('stream-error', (event, id: number, error: Error) => {
+      streams.get(id)?.emit('error', error);
+      streams.delete(id);
+    });
+  }
+
+  private constructor(private readonly id: number) {
+    super();
+    streams.set(id, this);
+  }
+
+  public static async open(filePath: string) {
+    return new IpcPushStream(await window.electron.ipcRenderer.invoke('stream-open', filePath));
+  }
+
+  public close() {
+    streams.delete(this.id);
+    ipcRenderer.send('stream-close', this.id);
+  }
+}