refactor: update node ws channel code (#3241)

Author: bytemain

jest.config.js

@@ -3,11 +3,14 @@ const { pathsToModuleNameMapper } = require('ts-jest');
 const tsconfig = require('./configs/ts/tsconfig.resolve.json');
 
 const tsModuleNameMapper = pathsToModuleNameMapper(tsconfig.compilerOptions.paths, { prefix: '<rootDir>/configs/' });
+
+/**
+ * @type {import('@jest/types').Config.InitialOptions}
+ */
 const baseConfig = {
   preset: 'ts-jest',
   testRunner: 'jest-jasmine2',
   resolver: '<rootDir>/tools/dev-tool/src/jest-resolver.js',
-  coverageProvider: process.env.JEST_COVERAGE_PROVIDER || 'babel',
   // https://dev.to/vantanev/make-your-jest-tests-up-to-20-faster-by-changing-a-single-setting-i36
   maxWorkers: 2,
   collectCoverageFrom: [
@@ -58,6 +61,11 @@ const baseConfig = {
     },
   },
 };
+
+if (process.env.JEST_COVERAGE_PROVIDER) {
+  baseConfig.coverageProvider = process.env.JEST_COVERAGE_PROVIDER;
+}
+
 /**
  * @type {import('@jest/types').Config.InitialOptions}
  */

jest.setup.base.js

@@ -1,3 +1,5 @@
+const { TextEncoder, TextDecoder } = require('util');
+
 // Do not log message on GitHub Actions.
 // Because these logs will affect the detection of real problems.
 const _console = global.console;
@@ -13,6 +15,9 @@ global.console = process.env.CI
     }
   : _console;
 
+global.TextEncoder = TextEncoder;
+global.TextDecoder = TextDecoder;
+
 process.on('unhandledRejection', (error) => {
   _console.error('unhandledRejection', error);
   if (process.env.EXIT_ON_UNHANDLED_REJECTION) {

jest.setup.node.js

@@ -54,8 +54,6 @@ global.document.queryCommandSupported = () => {};
 global.document.execCommand = () => {};
 global.HTMLElement = jsdom.window.HTMLElement;
 global.self = global;
-global.TextEncoder = TextEncoder;
-global.TextDecoder = TextDecoder;
 
 global.ElectronIpcRenderer = {
   send: () => {},

package.json

@@ -95,7 +95,7 @@
     "@types/react-is": "^16.7.1",
     "@types/socket.io-client": "^1.4.32",
     "@types/temp": "^0.9.1",
-    "@types/ws": "^6.0.1",
+    "@types/ws": "^8.5.10",
     "@typescript-eslint/eslint-plugin": "^5.12.1",
     "@typescript-eslint/parser": "^5.12.1",
     "async-retry": "^1.3.1",

packages/connection/__test__/browser/index.test.ts

@@ -1,34 +1,39 @@
-import { WebSocket, Server } from 'mock-socket';
+import { ReconnectingWebSocketConnection } from '@opensumi/ide-connection/lib/common/connection/drivers/reconnecting-websocket';
+import { WebSocket, Server } from '@opensumi/mock-socket';
 
 import { WSChannelHandler } from '../../src/browser/ws-channel-handler';
-import { stringify, parse } from '../../src/common/utils';
+import { stringify, parse } from '../../src/common/ws-channel';
 (global as any).WebSocket = WebSocket;
 
+const randomPortFn = () => Math.floor(Math.random() * 10000) + 10000;
+
+const randomPort = randomPortFn();
+
 describe('connection browser', () => {
   it('init connection', async () => {
     jest.setTimeout(20000);
 
-    const fakeWSURL = 'ws://localhost:8089';
+    const fakeWSURL = `ws://localhost:${randomPort}`;
     const mockServer = new Server(fakeWSURL);
 
     let receivedHeartbeat = false;
     mockServer.on('connection', (socket) => {
       socket.on('message', (msg) => {
-        const msgObj = parse(msg as string);
+        const msgObj = parse(msg as Uint8Array);
         if (msgObj.kind === 'open') {
           socket.send(
             stringify({
               id: msgObj.id,
-              kind: 'ready',
+              kind: 'server-ready',
             }),
           );
-        } else if (msgObj.kind === 'heartbeat') {
+        } else if (msgObj.kind === 'ping') {
           receivedHeartbeat = true;
         }
       });
     });
 
-    const wsChannelHandler = new WSChannelHandler(fakeWSURL, console);
+    const wsChannelHandler = new WSChannelHandler(ReconnectingWebSocketConnection.forURL(fakeWSURL), console);
 
     await wsChannelHandler.initHandler();
     await new Promise<void>((resolve) => {

packages/connection/__test__/common/buffers.test.ts

@@ -0,0 +1,76 @@
+import { Buffers } from '../../src/common/connection/buffers';
+
+describe('Buffers', () => {
+  it('can append and slice', () => {
+    const list = new Buffers();
+    list.push(new Uint8Array([1, 2, 3]));
+    list.push(new Uint8Array([4, 5, 6]));
+
+    expect(list.slice(0, 0)).toEqual(new Uint8Array(0));
+    expect(list.slice(0, 2)).toEqual(new Uint8Array([1, 2]));
+    expect(list.slice(0, 7)).toEqual(new Uint8Array([1, 2, 3, 4, 5, 6]));
+
+    expect(list.slice(1, 1)).toEqual(new Uint8Array(0));
+    expect(list.slice(1, 5)).toEqual(new Uint8Array([2, 3, 4, 5]));
+    expect(list.slice(1, 7)).toEqual(new Uint8Array([2, 3, 4, 5, 6]));
+
+    expect(list.slice(2, 2)).toEqual(new Uint8Array(0));
+    expect(list.slice(2, 6)).toEqual(new Uint8Array([3, 4, 5, 6]));
+  });
+
+  it('can splice', () => {
+    const list = new Buffers();
+    list.push(new Uint8Array([1, 2, 3]));
+    list.push(new Uint8Array([4, 5, 6]));
+
+    expect(list.splice(0, 0)).toEqual({ buffers: [], size: 0 });
+    expect(list.byteLength).toEqual(6);
+    expect(list.splice(0, 1)).toEqual({ buffers: [new Uint8Array([1])], size: 1 });
+    expect(list.byteLength).toEqual(5);
+    expect(list.splice(0, 2)).toEqual({ buffers: [new Uint8Array([2, 3])], size: 2 });
+    expect(list.byteLength).toEqual(3);
+
+    expect(list.splice(0, 0, new Uint8Array([1]))).toEqual({ buffers: [], size: 0 });
+    expect(list.byteLength).toEqual(4);
+    expect(list.splice(0, 0, new Uint8Array([2, 3]))).toEqual({ buffers: [], size: 0 });
+    expect(list.byteLength).toEqual(6);
+    expect(list.splice(0, 0, new Uint8Array([4, 5, 6]))).toEqual({ buffers: [], size: 0 });
+    expect(list.byteLength).toEqual(9);
+
+    expect(list.buffers).toEqual([
+      new Uint8Array([4, 5, 6]),
+      new Uint8Array([2, 3]),
+      new Uint8Array([1]),
+      new Uint8Array([4, 5, 6]),
+    ]);
+  });
+  it('can copy', () => {
+    const list = new Buffers();
+    list.push(new Uint8Array([1, 2, 3]));
+
+    list.push(new Uint8Array([4, 5, 6]));
+
+    const target = new Uint8Array(7);
+
+    list.copy(target, 0);
+
+    expect(target).toEqual(new Uint8Array([1, 2, 3, 4, 5, 6, 0]));
+  });
+
+  it('other', () => {
+    const list = new Buffers();
+    list.push(new Uint8Array([1, 2, 3]));
+    list.push(new Uint8Array([4, 5, 6]));
+    expect(list.byteLength).toEqual(6);
+    expect(list.pos(0)).toEqual({ buf: 0, offset: 0 });
+    expect(list.pos(1)).toEqual({ buf: 0, offset: 1 });
+    expect(list.pos(2)).toEqual({ buf: 0, offset: 2 });
+    expect(list.get(0)).toEqual(1);
+    expect(list.get(1)).toEqual(2);
+
+    list.set(0, 9);
+    list.set(1, 10);
+    expect(list.get(0)).toEqual(9);
+    expect(list.get(1)).toEqual(10);
+  });
+});

packages/connection/__test__/common/stream-decoder.test.ts

@@ -0,0 +1,105 @@
+/* eslint-disable no-console */
+
+import { BinaryReader } from '@furyjs/fury/dist/lib/reader';
+
+import {
+  StreamPacketDecoder,
+  createStreamPacket,
+  kMagicNumber,
+} from '../../src/common/connection/drivers/stream-decoder';
+
+const reader = BinaryReader({});
+
+function round(x: number, count: number) {
+  return Math.round(x * 10 ** count) / 10 ** count;
+}
+
+function createPayload(size: number) {
+  const payload = new Uint8Array(size);
+  for (let i = 0; i < size; i++) {
+    payload[i] = i % 256;
+  }
+
+  return payload;
+}
+
+console.time('createPayload');
+const p1k = createPayload(1024);
+const p64k = createPayload(64 * 1024);
+const p128k = createPayload(128 * 1024);
+const p1m = createPayload(1024 * 1024);
+const p5m = createPayload(5 * 1024 * 1024);
+const p10m = createPayload(10 * 1024 * 1024);
+
+const h1m = createPayload(1024 + p1m.byteLength);
+h1m.set(p1m, 1024);
+
+const h5m = createPayload(1024 + p5m.byteLength + 233);
+h5m.set(p5m, 1024);
+
+console.timeEnd('createPayload');
+
+// 1m
+const pressure = 1024 * 1024;
+
+const purePackets = [p1k, p64k, p128k, p5m, p10m].map((v) => [createStreamPacket(v), v] as const);
+
+const mixedPackets = [p1m, p5m].map((v) => {
+  const sumiPacket = createStreamPacket(v);
+  const newPacket = createPayload(1024 + sumiPacket.byteLength);
+  newPacket.set(sumiPacket, 1024);
+  return [newPacket, v] as const;
+});
+
+const packets = [...purePackets, ...mixedPackets];
+
+describe('stream-packet', () => {
+  it('can create sumi stream packet', () => {
+    const content = new Uint8Array([1, 2, 3]);
+    const packet = createStreamPacket(content);
+
+    reader.reset(packet);
+    expect(reader.uint32()).toBe(kMagicNumber);
+    expect(reader.varUInt32()).toBe(content.byteLength);
+    expect(Uint8Array.from(reader.buffer(content.byteLength))).toEqual(content);
+  });
+
+  packets.forEach(([packet, expected]) => {
+    it(`can decode stream packet: ${round(packet.byteLength / 1024 / 1024, 2)}m`, (done) => {
+      const decoder = new StreamPacketDecoder();
+
+      decoder.onData((data) => {
+        expect(data.byteLength).toEqual(expected.byteLength);
+        for (let i = 0; i < 10; i++) {
+          // 随机选一些数据(<= 100字节)，对比是否正确，对比整个数组的话，超大 buffer 会很耗时
+          const start = Math.floor(Math.random() * data.byteLength);
+          const end = Math.floor(Math.random() * 1024);
+
+          expect(data.subarray(start, end)).toEqual(expected.subarray(start, end));
+        }
+        decoder.dispose();
+        done();
+      });
+
+      console.log('write chunk', packet.byteLength);
+      // write chunk by ${pressure} bytes
+      for (let i = 0; i < packet.byteLength; i += pressure) {
+        decoder.push(packet.subarray(i, i + pressure));
+        logMemoryUsage();
+      }
+
+      logMemoryUsage();
+    });
+  });
+});
+
+function logMemoryUsage() {
+  const used = process.memoryUsage();
+  let text = new Date().toLocaleString('zh') + ' Memory usage:\n';
+  // eslint-disable-next-line guard-for-in
+  for (const key in used) {
+    text += `${key} ${Math.round((used[key] / 1024 / 1024) * 100) / 100} MB\n`;
+  }
+
+  console.log(text);
+}