fix: frame broken

Author: bytemain

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

@@ -1,4 +1,4 @@
-import { Buffers, copy } from '../../src/common/connection/buffers';
+import { Buffers, copy } from '../../src/common/buffers/buffers';
 
 describe('Buffers', () => {
   it('can push and slice', () => {
@@ -232,21 +232,21 @@ describe('Buffers', () => {
     expect(cursor.lineOffset).toEqual(1);
     expect(list.pos(cursor.offset)).toEqual({ buf: cursor.line, offset: cursor.lineOffset });
 
-    cursor.move(1);
+    cursor.skip(1);
     expect(cursor.offset).toEqual(6);
     expect(cursor.value).toEqual(list.get(cursor.offset));
     expect(cursor.line).toEqual(1);
     expect(cursor.lineOffset).toEqual(2);
     expect(list.pos(cursor.offset)).toEqual({ buf: cursor.line, offset: cursor.lineOffset });
 
-    cursor.move(2);
+    cursor.skip(2);
     expect(cursor.offset).toEqual(8);
     expect(cursor.value).toEqual(list.get(cursor.offset));
     expect(cursor.line).toEqual(1);
     expect(cursor.lineOffset).toEqual(4);
     expect(list.pos(cursor.offset)).toEqual({ buf: cursor.line, offset: cursor.lineOffset });
 
-    cursor.move(3);
+    cursor.skip(3);
     expect(cursor.offset).toEqual(11);
     expect(cursor.value).toEqual(list.get(cursor.offset));
     expect(cursor.line).toEqual(2);

…n/__test__/common/stream-decoder.test.ts …on/__test__/common/frame-decoder.test.tspackages/connection/__test__/common/stream-decoder.test.ts renamed to packages/connection/__test__/common/frame-decoder.test.ts packages/connection/__test__/common/stream-decoder.test.ts renamed to packages/connection/__test__/common/frame-decoder.test.ts

@@ -3,12 +3,10 @@
 import { BinaryReader } from '@furyjs/fury/dist/lib/reader';
 
 import {
-  StreamPacketDecoder,
-  createStreamPacket,
-  kMagicNumber,
-} from '../../src/common/connection/drivers/stream-decoder';
-
-const reader = BinaryReader({});
+  LengthFieldBasedFrameDecoder,
+  indicator,
+  prependLengthField,
+} from '../../src/common/connection/drivers/frame-decoder';
 
 function round(x: number, count: number) {
   return Math.round(x * 10 ** count) / 10 ** count;
@@ -42,14 +40,7 @@ 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 purePackets = [p1k, p64k, p128k, p5m, p10m].map((v) => [prependLengthField(v), v] as const);
 
 const size = purePackets.reduce((acc, v) => acc + v[0].byteLength, 0);
 
@@ -61,22 +52,30 @@ purePackets.forEach((v) => {
   offset += sumiPacket.byteLength;
 });
 
+const mixedPackets = [p1m, p5m].map((v) => {
+  const sumiPacket = prependLengthField(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', () => {
+describe('frame decoder', () => {
+  it('can create frame', () => {
     const content = new Uint8Array([1, 2, 3]);
-    const packet = createStreamPacket(content);
+    const packet = prependLengthField(content);
+    const reader = BinaryReader({});
 
     reader.reset(packet);
-    expect(reader.uint32()).toBe(kMagicNumber);
-    expect(reader.varUInt32()).toBe(content.byteLength);
+    expect(Uint8Array.from(reader.buffer(4))).toEqual(indicator);
+    expect(reader.uint32()).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();
+    it(`can decode stream: ${round(packet.byteLength / 1024 / 1024, 2)}m`, (done) => {
+      const decoder = new LengthFieldBasedFrameDecoder();
 
       decoder.onData((data) => {
         fastExpectBufferEqual(data, expected);
@@ -95,8 +94,8 @@ describe('stream-packet', () => {
     });
   });
 
-  it('can decode a stream payload contains multiple packets', (done) => {
-    const decoder = new StreamPacketDecoder();
+  it('can decode a stream payload contains multiple frames', (done) => {
+    const decoder = new LengthFieldBasedFrameDecoder();
     const expectCount = purePackets.length;
     let count = 0;
     decoder.onData((data) => {
@@ -120,18 +119,17 @@ describe('stream-packet', () => {
     logMemoryUsage();
   });
 
-  it('can decode a stream it header and payload are separated', (done) => {
+  it('can decode a stream it has no valid length info', (done) => {
     const v = createPayload(1024);
-    const sumiPacket = createStreamPacket(v);
+    const sumiPacket = prependLengthField(v);
 
-    const decoder = new StreamPacketDecoder();
+    const decoder = new LengthFieldBasedFrameDecoder();
     decoder.onData((data) => {
       fastExpectBufferEqual(data, v);
       done();
     });
 
     console.log('write chunk', sumiPacket.byteLength);
-
     // use pressure = 2 to simulate the header and payload are separated
     const pressure = 2;
     for (let i = 0; i < sumiPacket.byteLength; i += pressure) {

packages/connection/__test__/node/ws-channel.test.ts

@@ -1,9 +1,10 @@
 import net from 'net';
 
 import { WSChannel } from '@opensumi/ide-connection';
-import { copy } from '@opensumi/ide-connection/lib/common/connection/buffers';
 import { normalizedIpcHandlerPathAsync } from '@opensumi/ide-core-common/lib/utils/ipc';
 
+import { copy } from '../../src/common/buffers/buffers';
+
 const total = 1000;
 
 describe('ws channel node', () => {

packages/connection/docs/ws-channel.md

@@ -5,3 +5,5 @@
 `WSChannel` 一开始是为了支持在浏览器环境下同时支持创建两个 RPCService 实例而出现的。我们在浏览器的 Browser 层需要和 Main 和 Ext 两个环境进行通信，建立两个 WebSocket 连接显然是不合适的。因此我们需要一个能够支持多路复用的 WebSocket 连接，然后 `WSChannel` 和 `WSChannelServer` 就应运而生。
 
 其实这种多路复用是非常有用的，在 Node.js 里我们也需要它，因为我们也会需要在一个 `net.Socket` 连接中同时传输多种信息。我遇到的情况就是需要同时支持 `@vscode/jsonrpc` 和 sumi-rpc，所以我把 Node.js 的通信连接各种东西也改成了 `WSChannel`。
+
+所以 `WSChannel` 是两个端通信的最基础的一层，需要发消息都是由 `WSChannel` 发送给对端的 `WSChannel` 来进行分发的。

…nection/src/common/connection/buffers.ts …connection/src/common/buffers/buffers.tspackages/connection/src/common/connection/buffers.ts renamed to packages/connection/src/common/buffers/buffers.ts packages/connection/src/common/connection/buffers.ts renamed to packages/connection/src/common/buffers/buffers.ts

@@ -261,9 +261,34 @@ export class Cursor {
     }
   }
 
-  move(n: number) {
-    this.offset += n;
-    this.updatePosition();
+  private skipCursor(n: number) {
+    let count = 0;
+    while (this.chunkIndex < this.buffers.buffers.length) {
+      const chunk = this.buffers.buffers[this.chunkIndex];
+      const chunkLength = chunk.byteLength;
+
+      while (this.chunkOffset < chunkLength) {
+        this.chunkOffset++;
+        this.offset++;
+
+        if (++count === n) {
+          return;
+        }
+      }
+      this.chunkOffset = 0;
+      this.chunkIndex++;
+    }
+  }
+
+  read(n: number) {
+    const end = this.offset + n;
+    const buffers = this.buffers.slice(this.offset, end);
+    this.skip(n);
+    return buffers;
+  }
+
+  skip(n: number) {
+    this.skipCursor(n);
   }
 
   moveTo(n: number) {

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

@@ -0,0 +1,170 @@
+import { BinaryWriter } from '@furyjs/fury/dist/lib/writer';
+
+import { readUInt32LE, Emitter } from '@opensumi/ide-core-common';
+
+import { Buffers } from '../../buffers/buffers';
+
+const writer = BinaryWriter({});
+
+/**
+ * You can use `Buffer.from('\r\n\r\n')` to get this indicator.
+ */
+export const indicator = new Uint8Array([0x0d, 0x0a, 0x0d, 0x0a]);
+
+/**
+ * sticky packet unpacking problems are generally problems at the transport layer.
+ * we use a length field to represent the length of the data, and then read the data according to the length
+ */
+export function prependLengthField(content: Uint8Array) {
+  writer.reset();
+  writer.buffer(indicator);
+  writer.uint32(content.byteLength);
+  writer.buffer(content);
+  return writer.dump();
+}
+
+export class LengthFieldBasedFrameDecoder {
+  protected dataEmitter = new Emitter<Uint8Array>();
+  onData = this.dataEmitter.event;
+
+  protected buffers = new Buffers();
+  protected cursor = this.buffers.cursor();
+
+  protected contentLength = -1;
+
+  protected state = 0;
+
+  /**
+   * The number of bytes in the length field.
+   *
+   * How many bytes are used to represent data length.
+   *
+   * For example, if the length field is 4 bytes, then the maximum length of the data is 2^32 = 4GB
+   */
+  lengthFieldLength = 4;
+
+  reset() {
+    this.contentLength = -1;
+    this.state = 0;
+    this.cursor.reset();
+  }
+
+  push(chunk: Uint8Array): void {
+    this.buffers.push(chunk);
+    let done = false;
+
+    while (!done) {
+      done = this.readFrame();
+    }
+  }
+
+  protected readFrame(): boolean {
+    const found = this.readLengthField();
+
+    if (found) {
+      const start = this.cursor.offset;
+      const end = start + this.contentLength;
+
+      const binary = this.buffers.slice(start, end);
+
+      this.dataEmitter.fire(binary);
+
+      if (this.buffers.byteLength > end) {
+        this.contentLength = -1;
+        this.state = 0;
+        this.cursor.moveTo(end);
+        // has more data, continue to parse
+        return false;
+      }
+
+      // delete used buffers
+      this.buffers.splice(0, end);
+      this.reset();
+    }
+
+    return true;
+  }
+
+  protected readLengthField() {
+    const bufferLength = this.buffers.byteLength;
+
+    if (this.state !== 4) {
+      if (this.cursor.offset + indicator.length > bufferLength) {
+        // Not enough data yet, wait for more data
+        return false;
+      }
+
+      this.readIndicator();
+    }
+
+    if (this.state !== 4) {
+      // Not a complete indicator yet, wait for more data
+      return false;
+    }
+
+    if (this.contentLength === -1) {
+      if (this.cursor.offset + this.lengthFieldLength > bufferLength) {
+        // Not enough data yet, wait for more data
+        return false;
+      }
+
+      // read the content length
+      const buf = this.cursor.read(this.lengthFieldLength);
+      // fury writer use little endian
+      this.contentLength = readUInt32LE(buf, 0);
+    }
+
+    if (this.cursor.offset + this.contentLength > bufferLength) {
+      // Not enough data yet, wait for more data
+      return false;
+    }
+
+    return true;
+  }
+
+  protected readIndicator() {
+    const iter = this.cursor.iterator();
+
+    let result = iter.next();
+    while (!result.done) {
+      switch (result.value) {
+        case 0x0d:
+          switch (this.state) {
+            case 0:
+              this.state = 1;
+              break;
+            case 2:
+              this.state = 3;
+              break;
+            default:
+              this.state = 0;
+              break;
+          }
+          break;
+        case 0x0a:
+          switch (this.state) {
+            case 1:
+              this.state = 2;
+              break;
+            case 3:
+              this.state = 4;
+              iter.return();
+              break;
+            default:
+              this.state = 0;
+              break;
+          }
+          break;
+        default:
+          this.state = 0;
+          break;
+      }
+      result = iter.next();
+    }
+  }
+
+  dispose() {
+    this.dataEmitter.dispose();
+    this.buffers.dispose();
+  }
+}

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

@@ -3,10 +3,10 @@ import type net from 'net';
 import { IDisposable } from '@opensumi/ide-core-common';
 
 import { BaseConnection } from './base';
-import { StreamPacketDecoder, createStreamPacket } from './stream-decoder';
+import { LengthFieldBasedFrameDecoder, prependLengthField } from './frame-decoder';
 
 export class NetSocketConnection extends BaseConnection<Uint8Array> {
-  protected decoder = new StreamPacketDecoder();
+  protected decoder = new LengthFieldBasedFrameDecoder();
 
   constructor(private socket: net.Socket) {
     super();
@@ -23,7 +23,7 @@ export class NetSocketConnection extends BaseConnection<Uint8Array> {
   }
 
   send(data: Uint8Array): void {
-    this.socket.write(createStreamPacket(data));
+    this.socket.write(prependLengthField(data));
   }
 
   onMessage(cb: (data: Uint8Array) => void): IDisposable {