Merge pull request #116 from EXXETA/feature/108-request-body-stream

feat(108): Stream Request and Response Bodies

.github/workflows/ci.yml

@@ -18,10 +18,10 @@ jobs:
       - name: Checkout code
         uses: actions/checkout@v4
 
-      - name: Use Node.js 20
+      - name: Use Node.js 22
         uses: actions/setup-node@v4
         with:
-          node-version: 20.x
+          node-version: 22.x
           cache: 'npm'
 
       - name: Install dependencies

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

@@ -1,4 +1,3 @@
-import { MainEventService } from './main-event-service';
 import path from 'node:path';
 import { tmpdir } from 'node:os';
 import { fs } from 'memfs';
@@ -12,7 +11,7 @@ jest.mock('electron', () => ({
   },
 }));
 
-const eventService = MainEventService.instance;
+jest.mock('./stream-events', () => ({}));
 
 const TEST_STRING = 'Hello, World!';
 const TEST_FILE_PATH = path.join(tmpdir(), 'test.txt');
@@ -23,30 +22,7 @@ describe('MainEventService', () => {
   });
 
   it('should register event functions on the backend', async () => {
+    await import('./main-event-service');
     expect((await import('electron')).ipcMain.handle).toHaveBeenCalled();
   });
-
-  it('should read the file correctly providing no parameters', async () => {
-    // Act
-    const buffer = await eventService.readFile(TEST_FILE_PATH);
-
-    // Assert
-    expect(Buffer.from(buffer).toString()).toBe(TEST_STRING);
-  });
-
-  it('should read the file correctly with offset', async () => {
-    // Act
-    const buffer = await eventService.readFile(TEST_FILE_PATH, 1);
-
-    // Assert
-    expect(Buffer.from(buffer).toString()).toBe(TEST_STRING.substring(1));
-  });
-
-  it('should read the file correctly with offset and length', async () => {
-    // Act
-    const buffer = await eventService.readFile(TEST_FILE_PATH, 1, 2);
-
-    // Assert
-    expect(Buffer.from(buffer).toString()).toBe(TEST_STRING.substring(1, 3));
-  });
 });

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

@@ -1,13 +1,11 @@
 import { IEventService } from 'shim/event-service';
 import { HttpService } from 'main/network/service/http-service';
 import { app, ipcMain } from 'electron';
-import { FileHandle, open, stat } from 'node:fs/promises';
-import { RequestBodyType, RufusRequest } from 'shim/objects/request';
-import { Buffer } from 'node:buffer';
+import { RufusRequest } from 'shim/objects/request';
 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 +58,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);
     }
@@ -75,35 +73,6 @@ export class MainEventService implements IEventService {
     return await HttpService.instance.fetchAsync(request);
   }
 
-  async readFile(filePath: string, offset = 0, length?: number) {
-    console.debug(
-      'Reading file at',
-      filePath,
-      'with offset',
-      offset,
-      'and length limited to',
-      length ?? 'unlimited',
-      'bytes'
-    );
-
-    let file: FileHandle | null = null;
-    try {
-      // get file size if length is not provided
-      if (length === undefined) {
-        const stats = await stat(filePath);
-        length = Math.max(stats.size - offset, 0);
-      }
-
-      const buffer = Buffer.alloc(length);
-      file = await open(filePath);
-      const read = await file.read(buffer, 0, length, offset);
-      console.debug('Read', read.bytesRead, 'bytes from file');
-      return buffer.subarray(0, read.bytesRead).buffer;
-    } finally {
-      if (file !== null) await file.close();
-    }
-  }
-
   async saveRequest(request: RufusRequest, textBody?: string) {
     await persistenceService.saveRequest(request, textBody);
   }
@@ -123,19 +92,4 @@ export class MainEventService implements IEventService {
   async deleteObject(object: RufusObject) {
     await persistenceService.delete(object);
   }
-
-  async loadTextRequestBody(request: RufusRequest) {
-    let text = '';
-
-    // TODO: Do not load the entire body into memory. Use ITextSnapshot instead
-    if (request.body?.type === RequestBodyType.TEXT) {
-      const stream = await persistenceService.loadTextBodyOfRequest(request);
-      if (stream == null) return '';
-      for await (const chunk of stream) {
-        text += chunk;
-      }
-    }
-
-    return text;
-  }
 }

src/main/event/stream-events.ts

@@ -0,0 +1,34 @@
+import { ipcMain } from 'electron';
+import { createReadStream, ReadStream } from 'node:fs';
+import { RufusRequest } from 'shim/objects/request';
+import { PersistenceService } from 'main/persistence/service/persistence-service';
+
+let nextId = 0;
+
+const streams = new Map<number, ReadStream>();
+
+const persistenceService = PersistenceService.instance;
+
+ipcMain.handle('stream-open', async (event, input: string | RufusRequest) => {
+  const { sender } = event;
+  const id = nextId++;
+
+  let stream: ReadStream;
+  if (typeof input === 'string') {
+    stream = createReadStream(input, 'utf8');
+  } else if ((stream = await persistenceService.loadTextBodyOfRequest(input, 'utf8')) == null) {
+    setImmediate(() => sender.send('stream-end', id));
+    return id;
+  }
+
+  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/main.ts

@@ -1,7 +1,7 @@
 import { app, BrowserWindow, shell } from 'electron';
-import 'main/event/main-event-service';
 import { EnvironmentService } from 'main/environment/service/environment-service';
 import installExtension, { REDUX_DEVTOOLS } from 'electron-devtools-installer';
+import 'main/event/main-event-service';
 
 // This allows TypeScript to pick up the magic constants that's auto-generated by Forge's Webpack
 // plugin that tells the Electron app where to look for the Webpack-bundled app code (depending on
@@ -35,7 +35,7 @@ const createWindow = async () => {
     return { action: 'deny' };
   });
 
-  // and load the index.html of the app.
+  // Load the index.html of the app.
   await mainWindow.loadURL(MAIN_WINDOW_WEBPACK_ENTRY);
 };
 

src/main/network/service/http-service.ts

@@ -7,7 +7,7 @@ import { Readable } from 'stream';
 import { EnvironmentService } from 'main/environment/service/environment-service';
 import { RequestBodyType, RufusRequest } from 'shim/objects/request';
 import { RufusResponse } from 'shim/objects/response';
-import { PersistenceService } from '../../persistence/service/persistence-service';
+import { PersistenceService } from 'main/persistence/service/persistence-service';
 import { RufusHeader } from 'shim/objects/headers';
 import { calculateResponseSize } from 'main/util/size-calculation';
 
@@ -33,7 +33,7 @@ export class HttpService {
    * @returns response object
    */
   public async fetchAsync(request: RufusRequest) {
-    console.info('Sending request: ', request);
+    console.info('Sending request:', request);
 
     const now = getSteadyTimestamp();
     const body = await this.readBody(request);
@@ -49,12 +49,12 @@ export class HttpService {
     });
 
     const duration = getDurationFromNow(now);
-    console.info(`Received response in ${duration} milliseconds:`, responseData);
+    console.info(`Received response in ${duration} milliseconds`);
 
     // write the response body to a temporary file
     const bodyFile = fileSystemService.temporaryFile();
     if (responseData.body != null) {
-      console.debug('Writing response body to temporary file: ', bodyFile.name);
+      console.debug('Writing response body to temporary file:', bodyFile.name);
       await pipeline(responseData.body, fs.createWriteStream('', { fd: bodyFile.fd }));
       console.debug('Successfully written response body');
     }
@@ -73,7 +73,7 @@ export class HttpService {
       bodyFilePath: responseData.body != null ? bodyFile.name : null,
     };
 
-    console.debug('Returning response: ', response);
+    console.debug('Returning response:', response);
     return response;
   }
 

src/main/persistence/service/persistence-service.test.ts

@@ -342,6 +342,24 @@ describe('PersistenceService', () => {
     expect(await streamToString(result)).toBe(textBody);
   });
 
+  it('loadTextBodyOfRequest() should load the text body of a request with utf8', async () => {
+    // Arrange
+    const textBody = 'text body';
+    const request = getExampleRequest(collection.id);
+    collection.children.push(request);
+
+    await persistenceService.saveCollectionRecursive(collection);
+    await persistenceService.saveRequest(request, textBody);
+
+    // Act
+    const result = (
+      await Array.fromAsync(await persistenceService.loadTextBodyOfRequest(request, 'utf8'))
+    ).join('');
+
+    // Assert
+    expect(result).toBe(textBody);
+  });
+
   it('loadTextBodyOfRequest() should load the text body of a draft request', async () => {
     // Arrange
     const textBody = 'text body';

src/main/persistence/service/persistence-service.ts

@@ -14,7 +14,6 @@ import {
 import { exists, USER_DATA_DIR } from 'main/util/fs-util';
 import { isCollection, isFolder, isRequest, RufusObject } from 'shim/objects';
 import { generateDefaultCollection } from './default-collection';
-import { Readable } from 'stream';
 import { randomUUID } from 'node:crypto';
 
 /**
@@ -255,20 +254,19 @@ export class PersistenceService {
   /**
    * Loads the text body of a request from the file system.
    * @param request the request to load the text body for
+   * @param encoding the encoding of the text body. Default is binary.
    * @returns the text body of the request if it exists
    */
-  public async loadTextBodyOfRequest(request: RufusRequest) {
-    console.log('Loading text body of request', request.id);
+  public async loadTextBodyOfRequest(request: RufusRequest, encoding?: BufferEncoding) {
+    console.info('Loading text body of request', request.id);
     if (request.body.type === RequestBodyType.TEXT) {
-      if (request.body.text != null) {
-        return Readable.from([request.body.text]);
-      }
-
       const fileName = request.draft ? DRAFT_TEXT_BODY_FILE_NAME : TEXT_BODY_FILE_NAME;
       const filePath = path.join(this.getDirPath(request), fileName);
       if (await exists(filePath)) {
-        return createReadStream(filePath);
+        console.debug(`Opening text body file at ${filePath}`);
+        return createReadStream(filePath, encoding);
       }
+      console.warn('Text body file does not exist for request', request.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/components/mainWindow/bodyTabs/OutputTabs.tsx

@@ -7,10 +7,7 @@ import { useEffect, useRef } from 'react';
 import { ResponseStatus } from '@/components/mainWindow/responseStatus/ResponseStatus';
 import { selectResponse, selectResponseEditor, setResponseEditor } from '@/state/responsesSlice';
 import { useDispatch, useSelector } from 'react-redux';
-import { RendererEventService } from '@/services/event/renderer-event-service';
-
-const eventService = RendererEventService.instance;
-const textDecoder = new TextDecoder();
+import { IpcPushStream } from '@/lib/ipc-stream';
 
 const monacoOptions = {
   ...DEFAULT_MONACO_OPTIONS,
@@ -39,13 +36,6 @@ function getContentType(headers?: HttpHeaders) {
   }
 }
 
-async function loadRequestBody(filePath: string) {
-  console.debug('Reading response body from', filePath);
-  const buffer = await eventService.readFile(filePath);
-  console.debug('Received response body of', buffer.byteLength, 'bytes');
-  return textDecoder.decode(buffer); // TODO: decode with encoding from response headers
-}
-
 interface OutputTabsProps {
   className: string;
 }
@@ -63,7 +53,10 @@ export function OutputTabs(props: OutputTabsProps) {
     } else if (response?.bodyFilePath == null) {
       editor.setValue('');
     } else {
-      loadRequestBody(response.bodyFilePath).then((body) => editor.setValue(body));
+      editor.setValue('');
+      IpcPushStream.open(response.bodyFilePath)
+        .then((stream) => IpcPushStream.collect(stream))
+        .then((content) => editor.setValue(content));
     }
   }, [response?.bodyFilePath, editor]);
 

src/renderer/components/sidebar/SidebarRequestList.tsx

@@ -3,9 +3,10 @@ import { AppDispatch, RootState } from '@/state/store';
 import { deleteRequest, setSelectedRequest } from '@/state/requestsSlice';
 import { httpMethodColor } from '@/services/StyleHelper';
 import { RequestBodyType, RufusRequest } from 'shim/objects/request';
-import { FaTimes } from 'react-icons/fa'; // Importing cross icon
+import { FaTimes } from 'react-icons/fa';
 import { RendererEventService } from '@/services/event/renderer-event-service';
 import { MouseEvent, useCallback, useEffect } from 'react';
+import { IpcPushStream } from '@/lib/ipc-stream';
 
 interface SidebarRequestListProps {
   requests: RufusRequest[];
@@ -20,17 +21,19 @@ export const SidebarRequestList = ({ requests = [] }: SidebarRequestListProps) =
   const request: RufusRequest | undefined = requests[selectedRequestIndex];
 
   useEffect(() => {
-    const model = requestEditor?.getModel();
-    if (model == null) {
+    if (requestEditor == null) {
       return;
-    }
-
-    if (request?.body?.type === RequestBodyType.TEXT) {
-      eventService.loadTextRequestBody(request).then((content) => model.setValue(content));
+    } else if (request?.body?.type === RequestBodyType.TEXT) {
+      IpcPushStream.open(request)
+        .then((stream) => IpcPushStream.collect(stream))
+        .then((content) => {
+          console.log(content);
+          requestEditor.setValue(content);
+        });
     } else {
-      model.setValue('');
+      requestEditor.setValue('');
     }
-  }, [request?.id, requestEditor?.getModel()]);
+  }, [request?.id, requestEditor]);
 
   const handleRowClick = useCallback(
     async (index: number) => {