feat: add sumi rpc (#3284)

* feat: add sumi rpc

* test: fix testcase

* chore: update code

* fix: forward binary in ext host

* feat: optimize proxy create function

* feat: using an incrementing request id

* feat: support one of

* fix: protocol repo not defined

* feat: use text decoder

* refactor: update code

* refactor: update protocol register

* test: fix testcase

* test: update testcase

* chore: update code

* chore: update types

* feat: connection add dispose method

* feat: avoid create function multiple times

* refactor: simplify ext connection implementation

* feat(rpc): support cancellation

* refactor: update code

* feat(rpc): support any serializer

* refactor: update code

* feat: any serializer can reviver uri

* feat: ext rpc protocol use channel

* fix: on request not found args

* fix: rpc timeout not work

* refactor: update code

* docs: add more docs

* fix: rpc broken

* docs: update docs

* fix: message port on message not work

* test: fix types

* refactor: use new style MessagePort event listener

* fix: any serializer cannot handle Uint8Array

* test: fix testcase

* refactor: update code

* test: remove only case

* feat: compatible with legacy

* refactor: rpc use sumi connection

* refactor: add sumi connection multiplexer

* test: fix testcases

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__/common/fury-extends/one-of.test.ts

@@ -0,0 +1,64 @@
+/* eslint-disable no-console */
+
+import { PingMessage, PongMessage, parse, stringify } from '../../../src/common/ws-channel';
+
+describe('oneOf', () => {
+  function testIt(obj: any) {
+    const bytes = stringify(obj);
+    const obj2 = parse(bytes);
+    expect(obj2).toEqual(obj);
+    const str = JSON.stringify(obj);
+
+    console.log('bytes.length', bytes.byteLength);
+    console.log('json length', str.length);
+  }
+
+  it('should serialize and deserialize', () => {
+    const obj = {
+      kind: 'ping',
+      clientId: '123',
+      id: '456',
+    } as PingMessage;
+
+    testIt(obj);
+
+    const obj2 = {
+      kind: 'pong',
+      clientId: '123',
+      id: '456',
+    } as PongMessage;
+
+    testIt(obj2);
+
+    const obj3 = {
+      kind: 'open',
+      clientId: '123',
+      id: '456',
+      path: '/test',
+    };
+
+    testIt(obj3);
+
+    const obj4 = {
+      kind: 'server-ready',
+      id: '456',
+    };
+
+    testIt(obj4);
+
+    const obj5 = {
+      kind: 'data',
+      id: '456',
+      content: 'hello',
+    };
+
+    testIt(obj5);
+
+    const obj6 = {
+      kind: 'binary',
+      id: '456',
+      binary: Buffer.from([1, 2, 3]),
+    };
+    testIt(obj6);
+  });
+});

packages/connection/__test__/common/rpc/common-tester.ts

@@ -0,0 +1,34 @@
+import { IConnectionPair } from './utils';
+
+export const test = (
+  name: string,
+  options: {
+    pairFactory: () => IConnectionPair;
+    factory: (pair: IConnectionPair) => any;
+  },
+) => {
+  const { factory, pairFactory } = options;
+  describe(name, () => {
+    let pair: IConnectionPair;
+    beforeEach(() => {
+      pair = pairFactory();
+    });
+
+    afterEach(() => {
+      pair && pair.close();
+    });
+
+    it('can call method', async () => {
+      const { invoker1, invoker2 } = factory(pair);
+
+      const result = await invoker1.add(1, 2);
+      expect(result).toBe(3);
+
+      const result2 = await invoker2.shortUrl('1234567890abcdefg');
+      expect(result2).toBe('1234567890');
+
+      const result3 = await invoker2.returnUndefined();
+      expect(result3).toBeUndefined();
+    });
+  });
+};

packages/connection/__test__/common/rpc/index.test.ts

@@ -0,0 +1,9 @@
+import { test } from './common-tester';
+import { createLegacyRPCClientPair, createMessageConnectionPair } from './utils';
+
+const factory = (pair: any) => createLegacyRPCClientPair(pair);
+
+test('legacy json rpc', {
+  factory,
+  pairFactory: createMessageConnectionPair,
+});

packages/connection/__test__/common/rpc/sumi-rpc.test.ts

@@ -0,0 +1,77 @@
+import { Type } from '@furyjs/fury';
+
+import { METHOD_NOT_REGISTERED } from '@opensumi/ide-connection/lib/common/constants';
+
+import { test } from './common-tester';
+import { createConnectionPair, createSumiRPCClientPair } from './utils';
+
+test('sumi rpc', {
+  factory: createSumiRPCClientPair,
+  pairFactory: createConnectionPair,
+});
+
+describe('sumi rpc only', () => {
+  let pair: ReturnType<typeof createConnectionPair>;
+  beforeEach(() => {
+    pair = createConnectionPair();
+  });
+
+  afterEach(() => {
+    pair && pair.close();
+  });
+
+  it('can throw error', async () => {
+    const { invoker1, invoker2, client1, client2 } = createSumiRPCClientPair(pair);
+
+    client1.listenService({
+      shortUrl: (url: string) => {
+        if (!url) {
+          throw new Error('url is empty');
+        }
+        return url.slice(0, 10);
+      },
+    });
+    client2.listenService({
+      add: (a: number, b: number) => {
+        if (a === 0) {
+          throw new Error('a is zero');
+        }
+        return a + b;
+      },
+    });
+
+    await expect(invoker1.add(0, 2)).rejects.toThrow('a is zero');
+    await expect(invoker2.shortUrl('')).rejects.toThrow('url is empty');
+
+    await expect(invoker1.throwAString()).resolves.toEqual(METHOD_NOT_REGISTERED);
+    await expect(invoker2.throwAString()).resolves.toEqual(METHOD_NOT_REGISTERED);
+
+    client2.listenService({
+      throwAString: () => {
+        // eslint-disable-next-line no-throw-literal
+        throw 'a string';
+      },
+    });
+
+    try {
+      await invoker1.throwAString();
+    } catch (error) {
+      expect(typeof error).toBe('string');
+      expect(error).toBe('a string');
+    }
+  });
+
+  it('can throw error when method not found', async () => {
+    const { invoker1, invoker2, repo } = createSumiRPCClientPair(pair);
+    repo.loadProtocolMethod({
+      method: 'notFound',
+      request: [],
+      response: {
+        type: Type.any(),
+      },
+    });
+
+    await expect(invoker1.notFound()).resolves.toEqual(METHOD_NOT_REGISTERED);
+    await expect(invoker2.notFound()).resolves.toEqual(METHOD_NOT_REGISTERED);
+  });
+});