Send messages from Electron Renderer to Main processes, with an RxJS observable to communicate progress and completion

_example.ts

// In Renderer
export function makeHttpRequest(url: string; method: string) {
  return createIpcSender<{ url: string; method: string }, void>(
    'http-request',
    { url, method },
  );
}
    
// ... in browser window code ...
makeHttpRequest('https://httpbin.org', 'GET').subscribe({
  next: value => console.log('Response from server', value),
  error: err => console.error(err),
  complete: () => console.log('End'),
});
    
    
// In Main
createIpcReceiver(
  'http-request',
  ({ url, method }: { url: string; method: string }) =>
    http.request$(url, method),
);
// http.request$ => fictional method that makes an HTTP request and returns an Observable.

ipc.ts

export function buildChannelNames(channel: string) {
  return {
    source: channel,
    next: `${channel}/next`,
    error: `${channel}/error`,
    complete: `${channel}/complete`,
  };
}

receiver.ts

import { Event, ipcMain } from 'electron';
import { Observable } from 'rxjs';
import { buildChannelNames } from './ipc';

export function createIpcReceiver<A, R>(channel: string, invoke: (args: A) => Observable<R>) {
  ipcMain.on(channel, async (event: Event, args: A) => {
    console.log(channel, args);
    const channels = buildChannelNames(channel);

    invoke(args).subscribe({
      next(value) {
        console.log(channels.next, value);
        event.sender.send(channels.next, {
          args,
          value,
        });
      },
      error(error) {
        let serializedError = error;
        if (error instanceof Error) {
          serializedError = {
            name: error.name,
            message: error.message,
            stack: error.stack,
          };
        }

        console.log(channels.error, serializedError);
        event.sender.send(channels.error, {
          args,
          error: serializedError,
        });
      },
      complete() {
        console.log(channels.complete);
        event.sender.send(channels.complete, {
          args,
        });
      },
    });
  });
}

sender.ts

import { ipcRenderer } from 'electron';
import { defer, Observable } from 'rxjs';
import { buildChannelNames } from './ipc';

export function createIpcSender<A, R>(channel: string, payload: A) {
  return defer(() => {
    ipcRenderer.send(channel, payload);

    const message$: Observable<R> = new Observable((observer: any) => {
      const channels = buildChannelNames(channel);

      const onNext = ({  }: Event, args: { args: any; value: any }) => {
        observer.next(args.value);
      };

      const onError = ({  }: Event, args: { args: any; error: any }) => {
        removeListeners();
        observer.error(args.error);
      };

      const onComplete = ({  }: Event) => {
        removeListeners();
        observer.complete();
      };

      const removeListeners = () => {
        ipcRenderer.removeListener(channels.next, onNext);
        ipcRenderer.removeListener(channels.error, onError);
        ipcRenderer.removeListener(channels.complete, onComplete);
      };

      ipcRenderer.on(channels.next, onNext);
      ipcRenderer.once(channels.error, onError);
      ipcRenderer.once(channels.complete, onComplete);

      return removeListeners;
    });

    return message$;
  });
}