WIP: feat: send data by chunk in websocket

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

@@ -35,7 +35,9 @@ console.timeEnd('createPayload');
 // 1m
 const pressure = 1024 * 1024;
 
-const purePackets = [p1k, p64k, p128k, p5m, p10m].map((v) => [LengthFieldBasedFrameDecoder.construct(v), v] as const);
+const purePackets = [p1k, p64k, p128k, p5m, p10m].map(
+  (v) => [LengthFieldBasedFrameDecoder.construct(v).dump(), v] as const,
+);
 
 const size = purePackets.reduce((acc, v) => acc + v[0].byteLength, 0);
 
@@ -48,7 +50,7 @@ purePackets.forEach((v) => {
 });
 
 const mixedPackets = [p1m, p5m].map((v) => {
-  const sumiPacket = LengthFieldBasedFrameDecoder.construct(v);
+  const sumiPacket = LengthFieldBasedFrameDecoder.construct(v).dump();
   const newPacket = createPayload(1024 + sumiPacket.byteLength);
   newPacket.set(sumiPacket, 1024);
   return [newPacket, v] as const;
@@ -59,7 +61,7 @@ const packets = [...purePackets, ...mixedPackets];
 describe('frame decoder', () => {
   it('can create frame', () => {
     const content = new Uint8Array([1, 2, 3]);
-    const packet = LengthFieldBasedFrameDecoder.construct(content);
+    const packet = LengthFieldBasedFrameDecoder.construct(content).dump();
     const reader = BinaryReader({});
 
     reader.reset(packet);
@@ -116,7 +118,7 @@ describe('frame decoder', () => {
 
   it('can decode a stream it has no valid length info', (done) => {
     const v = createPayload(1024);
-    const sumiPacket = LengthFieldBasedFrameDecoder.construct(v);
+    const sumiPacket = LengthFieldBasedFrameDecoder.construct(v).dump();
 
     const decoder = new LengthFieldBasedFrameDecoder();
     decoder.onData((data) => {

packages/connection/src/common/connection/drivers/frame-decoder.ts

@@ -1,6 +1,6 @@
 import { BinaryWriter } from '@furyjs/fury/dist/lib/writer';
 
-import { Emitter, readUInt32LE } from '@opensumi/ide-core-common';
+import { MaybeNull, readUInt32LE } from '@opensumi/ide-core-common';
 
 import { Buffers } from '../../buffers/buffers';
 
@@ -14,8 +14,15 @@ export const indicator = new Uint8Array([0x0d, 0x0a, 0x0d, 0x0a]);
  * we use a length field to represent the length of the data, and then read the data according to the length
  */
 export class LengthFieldBasedFrameDecoder {
-  protected dataEmitter = new Emitter<Uint8Array>();
-  onData = this.dataEmitter.event;
+  private _onDataListener: MaybeNull<(data: Uint8Array) => void>;
+  onData(listener: (data: Uint8Array) => void) {
+    this._onDataListener = listener;
+    return {
+      dispose: () => {
+        this._onDataListener = null;
+      },
+    };
+  }
 
   protected buffers = new Buffers();
   protected cursor = this.buffers.cursor();
@@ -57,7 +64,9 @@ export class LengthFieldBasedFrameDecoder {
 
       const binary = this.buffers.slice(start, end);
 
-      this.dataEmitter.fire(binary);
+      if (this._onDataListener) {
+        this._onDataListener(binary);
+      }
 
       if (this.buffers.byteLength > end) {
         this.contentLength = -1;
@@ -154,7 +163,7 @@ export class LengthFieldBasedFrameDecoder {
   }
 
   dispose() {
-    this.dataEmitter.dispose();
+    this._onDataListener = undefined;
     this.buffers.dispose();
   }
 
@@ -165,6 +174,6 @@ export class LengthFieldBasedFrameDecoder {
     LengthFieldBasedFrameDecoder.writer.buffer(indicator);
     LengthFieldBasedFrameDecoder.writer.uint32(content.byteLength);
     LengthFieldBasedFrameDecoder.writer.buffer(content);
-    return LengthFieldBasedFrameDecoder.writer.dump();
+    return LengthFieldBasedFrameDecoder.writer;
   }
 }

packages/connection/src/common/connection/drivers/reconnecting-websocket.ts

@@ -4,17 +4,30 @@ import ReconnectingWebSocket, {
   UrlProvider,
 } from '@opensumi/reconnecting-websocket';
 
+import { chunkSize } from '../../constants';
+
 import { BaseConnection } from './base';
+import { LengthFieldBasedFrameDecoder } from './frame-decoder';
 
 import type { ErrorEvent } from '@opensumi/reconnecting-websocket';
 
 export class ReconnectingWebSocketConnection extends BaseConnection<Uint8Array> {
+  protected decoder = new LengthFieldBasedFrameDecoder();
+
   constructor(private socket: ReconnectingWebSocket) {
     super();
+
+    this.socket.addEventListener('message', this.dataHandler);
   }
 
   send(data: Uint8Array): void {
-    this.socket.send(data);
+    const handle = LengthFieldBasedFrameDecoder.construct(data).dumpAndOwn();
+    const packet = handle.get();
+    for (let i = 0; i < packet.byteLength; i += chunkSize) {
+      this.socket.send(packet.subarray(i, i + chunkSize));
+    }
+
+    handle.dispose();
   }
 
   isOpen(): boolean {
@@ -31,27 +44,7 @@ export class ReconnectingWebSocketConnection extends BaseConnection<Uint8Array>
   }
 
   onMessage(cb: (data: Uint8Array) => void): IDisposable {
-    const handler = (e: MessageEvent) => {
-      let buffer: Promise<ArrayBuffer>;
-      if (e.data instanceof Blob) {
-        buffer = e.data.arrayBuffer();
-      } else if (e.data instanceof ArrayBuffer) {
-        buffer = Promise.resolve(e.data);
-      } else if (e.data?.constructor?.name === 'Buffer') {
-        // Compatibility with nodejs Buffer in test environment
-        buffer = Promise.resolve(e.data);
-      } else {
-        throw new Error('unknown message type, expect Blob or ArrayBuffer, received: ' + typeof e.data);
-      }
-      buffer.then((v) => cb(new Uint8Array(v, 0, v.byteLength)));
-    };
-
-    this.socket.addEventListener('message', handler);
-    return {
-      dispose: () => {
-        this.socket.removeEventListener('message', handler);
-      },
-    };
+    return this.decoder.onData(cb);
   }
   onceClose(cb: (code?: number, reason?: string) => void): IDisposable {
     const disposable = this.onClose(wrapper);
@@ -91,8 +84,23 @@ export class ReconnectingWebSocketConnection extends BaseConnection<Uint8Array>
     };
   }
 
+  private dataHandler = (e: MessageEvent) => {
+    let buffer: Promise<ArrayBuffer>;
+    if (e.data instanceof Blob) {
+      buffer = e.data.arrayBuffer();
+    } else if (e.data instanceof ArrayBuffer) {
+      buffer = Promise.resolve(e.data);
+    } else if (e.data?.constructor?.name === 'Buffer') {
+      // Compatibility with nodejs Buffer in test environment
+      buffer = Promise.resolve(e.data);
+    } else {
+      throw new Error('unknown message type, expect Blob or ArrayBuffer, received: ' + typeof e.data);
+    }
+    buffer.then((v) => this.decoder.push(new Uint8Array(v, 0, v.byteLength)));
+  };
+
   dispose(): void {
-    // do nothing
+    this.socket.removeEventListener('message', this.dataHandler);
   }
 
   static forURL(url: UrlProvider, protocols?: string | string[], options?: ReconnectingWebSocketOptions) {

packages/connection/src/common/connection/drivers/stream.ts

@@ -21,10 +21,11 @@ export class StreamConnection extends BaseConnection<Uint8Array> {
   }
 
   send(data: Uint8Array): void {
-    const result = LengthFieldBasedFrameDecoder.construct(data);
-    this.writable.write(result, () => {
+    const handle = LengthFieldBasedFrameDecoder.construct(data).dumpAndOwn();
+    this.writable.write(handle.get(), () => {
       // TODO: logger error
     });
+    handle.dispose();
   }
 
   onMessage(cb: (data: Uint8Array) => void): IDisposable {

packages/connection/src/common/connection/drivers/ws-websocket.ts

@@ -1,24 +1,34 @@
 import { IDisposable } from '@opensumi/ide-core-common';
 
+import { chunkSize } from '../../constants';
+
 import { BaseConnection } from './base';
+import { LengthFieldBasedFrameDecoder } from './frame-decoder';
 
 import type WebSocket from 'ws';
 
 export class WSWebSocketConnection extends BaseConnection<Uint8Array> {
+  protected decoder = new LengthFieldBasedFrameDecoder();
+
   constructor(public socket: WebSocket) {
     super();
+    this.socket.on('message', (data: Buffer) => {
+      this.decoder.push(data);
+    });
   }
+
   send(data: Uint8Array): void {
-    this.socket.send(data);
+    const handle = LengthFieldBasedFrameDecoder.construct(data).dumpAndOwn();
+    const packet = handle.get();
+    for (let i = 0; i < packet.byteLength; i += chunkSize) {
+      this.socket.send(packet.subarray(i, i + chunkSize));
+    }
+
+    handle.dispose();
   }
 
   onMessage(cb: (data: Uint8Array) => void): IDisposable {
-    this.socket.on('message', cb);
-    return {
-      dispose: () => {
-        this.socket.off('message', cb);
-      },
-    };
+    return this.decoder.onData(cb);
   }
   onceClose(cb: () => void): IDisposable {
     this.socket.once('close', cb);

packages/connection/src/common/constants.ts

@@ -1 +1,6 @@
 export const METHOD_NOT_REGISTERED = '$$METHOD_NOT_REGISTERED';
+
+/**
+ * 分片大小, 8MB
+ */
+export const chunkSize = 8 * 1024 * 1024;

packages/connection/src/node/common-channel-handler.ts

@@ -42,8 +42,7 @@ export class CommonChannelHandler extends BaseCommonChannelHandler implements We
       ...this.options.wsServerOptions,
     });
     this.wsServer.on('connection', (connection: WebSocket) => {
-      const wsConnection = new WSWebSocketConnection(connection);
-      this.receiveConnection(wsConnection);
+      this.receiveConnection(new WSWebSocketConnection(connection));
     });
   }