@yuants/protocol - npm Package Compare versions

Comparing version 0.23.2 to 0.23.3

dist/terminal.js

@@ -5,3 +5,3 @@ import { UUID, formatTime } from '@yuants/data-model';
 import { isNode } from 'browser-or-node';
-import { EMPTY, Observable, ReplaySubject, Subject, catchError, debounceTime, defer, distinctUntilChanged, exhaustMap, filter, first, firstValueFrom, from, groupBy, interval, map, mergeAll, mergeMap, of, repeat, retry, share, shareReplay, switchMap, takeWhile, tap, timeout, timer, toArray, withLatestFrom, } from 'rxjs';
+import { EMPTY, Observable, ReplaySubject, Subject, catchError, debounceTime, defer, distinctUntilChanged, exhaustMap, filter, first, firstValueFrom, from, fromEvent, groupBy, interval, map, mergeAll, mergeMap, mergeWith, of, partition, repeat, retry, share, shareReplay, switchMap, takeWhile, tap, timeout, timer, toArray, withLatestFrom, } from 'rxjs';
 import { createConnectionJson } from './create-connection';
@@ -386,2 +386,3 @@ import { PromRegistry } from './services/metrics';
         this._subscriptions.push(this._output$.subscribe((msg) => {
+            var _a;
             // Local Loop Back Tunnel
@@ -411,7 +412,36 @@ if (msg.target_terminal_id === this.terminal_id) {
             }
-            if (peerInfo && peerInfo.peer.connected) {
-                console.info(formatTime(Date.now()), 'Terminal', 'WebRTC', 'sent', JSON.stringify(msg));
+            // NOTE: reserve 32KB for other purpose
+            const reservedSize = 32 * 1024;
+            if (peerInfo && peerInfo.peer.connected && ((_a = peerInfo.maxMessageSize) !== null && _a !== void 0 ? _a : 0) > reservedSize) {
+                const stringified = JSON.stringify(msg);
                 setTimeout(() => {
                     try {
-                        peerInfo.peer.send(JSON.stringify(msg));
+                        const trace_id = UUID();
+                        const chunkSize = peerInfo.maxMessageSize - reservedSize;
+                        if (stringified.length > chunkSize) {
+                            for (let i = 0; i < stringified.length; i += chunkSize) {
+                                const body = stringified.slice(i, i + chunkSize);
+                                console.info(formatTime(Date.now()), 'Terminal', 'WebRTC', 'sent', `chunkSize: ${chunkSize}`, body.length);
+                                peerInfo.peer.send(JSON.stringify({
+                                    trace_id,
+                                    method: `WebRTC/Chunk`,
+                                    source_terminal_id: msg.source_terminal_id,
+                                    target_terminal_id: msg.target_terminal_id,
+                                    frame: {
+                                        seq: i / chunkSize,
+                                        body,
+                                    },
+                                }));
+                            }
+                            peerInfo.peer.send(JSON.stringify({
+                                trace_id,
+                                method: `WebRTC/Chunk`,
+                                source_terminal_id: msg.source_terminal_id,
+                                target_terminal_id: msg.target_terminal_id,
+                                res: { code: 200, message: 'OK' },
+                            }));
+                        }
+                        else {
+                            peerInfo.peer.send(stringified);
+                        }
                     }
@@ -431,2 +461,3 @@ catch (err) {
         const { session_id, direction, remote_terminal_id, onSignal, onDestroy } = config;
+        const subs = [];
         const peer = getSimplePeerInstance({
@@ -440,2 +471,3 @@ initiator: direction === 'Active',
             session_id,
+            maxMessageSize: 65536,
             peer,
@@ -445,6 +477,35 @@ };
             console.info(formatTime(Date.now()), 'Terminal', 'WebRTC', direction, 'signal', session_id, remote_terminal_id, data);
-            onSignal(data);
+            try {
+                onSignal(data);
+            }
+            catch (err) {
+                console.error(formatTime(Date.now()), 'Error', err);
+                this._mapTerminalIdToPeer[remote_terminal_id] = undefined;
+                for (const sub of subs) {
+                    sub.unsubscribe();
+                }
+                onDestroy();
+            }
         });
-        peer.on('data', (data) => {
-            console.info(formatTime(Date.now()), 'Terminal', 'WebRTC', direction, 'data', session_id, remote_terminal_id, data.toString());
+        const data$ = fromEvent(peer, 'data').pipe(
+        //
+        tap((data) => {
+            console.info(formatTime(Date.now()), 'Terminal', 'WebRTC', direction, 'data', session_id, remote_terminal_id, data.length);
+        }), map((v) => JSON.parse(v.toString())), share());
+        const [chunkData$, completeData$] = partition(data$, (v) => v.method === 'WebRTC/Chunk');
+        const resembledData$ = chunkData$.pipe(
+        //
+        groupBy((v) => v.trace_id), mergeMap((group) => group.pipe(
+        //
+        takeWhile((data) => data.res === undefined), map((v) => v.frame), toArray(), map((v) => v
+            .sort((a, b) => a.seq - b.seq)
+            .map((x) => x.body)
+            .join('')), map((v) => JSON.parse(v)), timeout({ each: 15000, meta: `WebRTC/Chunk Timeout: trace_id=${group.key}` }), catchError((err) => {
+            console.error('Error', err);
+            return EMPTY;
+        }))));
+        subs.push(completeData$
+            .pipe(
+        //
+        mergeWith(resembledData$), tap((data) => {
             if (data.method) {
@@ -466,6 +527,11 @@ TerminalReceiveMassageTotal.inc({
             }
-            this._input$.next(JSON.parse(data.toString()));
-        });
+        }))
+            .subscribe((msg) => this._input$.next(msg)));
         peer.on('connect', () => {
+            var _a, _b;
             console.info(formatTime(Date.now()), 'Terminal', 'WebRTC', direction, 'connected', session_id, remote_terminal_id);
+            // @ts-ignore
+            const maxMessageSize = (_b = (_a = peer._pc.sctp) === null || _a === void 0 ? void 0 : _a.maxMessageSize) !== null && _b !== void 0 ? _b : 65536;
+            this._mapTerminalIdToPeer[remote_terminal_id] = { session_id, maxMessageSize, peer };
+            console.info(formatTime(Date.now()), 'Terminal', 'WebRTC', direction, 'maxMessageSize', maxMessageSize);
         });
@@ -475,2 +541,5 @@ peer.on('close', () => {
             this._mapTerminalIdToPeer[remote_terminal_id] = undefined;
+            for (const sub of subs) {
+                sub.unsubscribe();
+            }
             onDestroy();
@@ -481,2 +550,5 @@ });
             this._mapTerminalIdToPeer[remote_terminal_id] = undefined;
+            for (const sub of subs) {
+                sub.unsubscribe();
+            }
             onDestroy();
@@ -483,0 +555,0 @@ });

lib/terminal.js

@@ -391,2 +391,3 @@ "use strict";
         this._subscriptions.push(this._output$.subscribe((msg) => {
+            var _a;
             // Local Loop Back Tunnel
@@ -416,7 +417,36 @@ if (msg.target_terminal_id === this.terminal_id) {
             }
-            if (peerInfo && peerInfo.peer.connected) {
-                console.info((0, data_model_1.formatTime)(Date.now()), 'Terminal', 'WebRTC', 'sent', JSON.stringify(msg));
+            // NOTE: reserve 32KB for other purpose
+            const reservedSize = 32 * 1024;
+            if (peerInfo && peerInfo.peer.connected && ((_a = peerInfo.maxMessageSize) !== null && _a !== void 0 ? _a : 0) > reservedSize) {
+                const stringified = JSON.stringify(msg);
                 setTimeout(() => {
                     try {
-                        peerInfo.peer.send(JSON.stringify(msg));
+                        const trace_id = (0, data_model_1.UUID)();
+                        const chunkSize = peerInfo.maxMessageSize - reservedSize;
+                        if (stringified.length > chunkSize) {
+                            for (let i = 0; i < stringified.length; i += chunkSize) {
+                                const body = stringified.slice(i, i + chunkSize);
+                                console.info((0, data_model_1.formatTime)(Date.now()), 'Terminal', 'WebRTC', 'sent', `chunkSize: ${chunkSize}`, body.length);
+                                peerInfo.peer.send(JSON.stringify({
+                                    trace_id,
+                                    method: `WebRTC/Chunk`,
+                                    source_terminal_id: msg.source_terminal_id,
+                                    target_terminal_id: msg.target_terminal_id,
+                                    frame: {
+                                        seq: i / chunkSize,
+                                        body,
+                                    },
+                                }));
+                            }
+                            peerInfo.peer.send(JSON.stringify({
+                                trace_id,
+                                method: `WebRTC/Chunk`,
+                                source_terminal_id: msg.source_terminal_id,
+                                target_terminal_id: msg.target_terminal_id,
+                                res: { code: 200, message: 'OK' },
+                            }));
+                        }
+                        else {
+                            peerInfo.peer.send(stringified);
+                        }
                     }
@@ -436,2 +466,3 @@ catch (err) {
         const { session_id, direction, remote_terminal_id, onSignal, onDestroy } = config;
+        const subs = [];
         const peer = (0, webrtc_1.getSimplePeerInstance)({
@@ -445,2 +476,3 @@ initiator: direction === 'Active',
             session_id,
+            maxMessageSize: 65536,
             peer,
@@ -450,6 +482,35 @@ };
             console.info((0, data_model_1.formatTime)(Date.now()), 'Terminal', 'WebRTC', direction, 'signal', session_id, remote_terminal_id, data);
-            onSignal(data);
+            try {
+                onSignal(data);
+            }
+            catch (err) {
+                console.error((0, data_model_1.formatTime)(Date.now()), 'Error', err);
+                this._mapTerminalIdToPeer[remote_terminal_id] = undefined;
+                for (const sub of subs) {
+                    sub.unsubscribe();
+                }
+                onDestroy();
+            }
         });
-        peer.on('data', (data) => {
-            console.info((0, data_model_1.formatTime)(Date.now()), 'Terminal', 'WebRTC', direction, 'data', session_id, remote_terminal_id, data.toString());
+        const data$ = (0, rxjs_1.fromEvent)(peer, 'data').pipe(
+        //
+        (0, rxjs_1.tap)((data) => {
+            console.info((0, data_model_1.formatTime)(Date.now()), 'Terminal', 'WebRTC', direction, 'data', session_id, remote_terminal_id, data.length);
+        }), (0, rxjs_1.map)((v) => JSON.parse(v.toString())), (0, rxjs_1.share)());
+        const [chunkData$, completeData$] = (0, rxjs_1.partition)(data$, (v) => v.method === 'WebRTC/Chunk');
+        const resembledData$ = chunkData$.pipe(
+        //
+        (0, rxjs_1.groupBy)((v) => v.trace_id), (0, rxjs_1.mergeMap)((group) => group.pipe(
+        //
+        (0, rxjs_1.takeWhile)((data) => data.res === undefined), (0, rxjs_1.map)((v) => v.frame), (0, rxjs_1.toArray)(), (0, rxjs_1.map)((v) => v
+            .sort((a, b) => a.seq - b.seq)
+            .map((x) => x.body)
+            .join('')), (0, rxjs_1.map)((v) => JSON.parse(v)), (0, rxjs_1.timeout)({ each: 15000, meta: `WebRTC/Chunk Timeout: trace_id=${group.key}` }), (0, rxjs_1.catchError)((err) => {
+            console.error('Error', err);
+            return rxjs_1.EMPTY;
+        }))));
+        subs.push(completeData$
+            .pipe(
+        //
+        (0, rxjs_1.mergeWith)(resembledData$), (0, rxjs_1.tap)((data) => {
             if (data.method) {
@@ -471,6 +532,11 @@ TerminalReceiveMassageTotal.inc({
             }
-            this._input$.next(JSON.parse(data.toString()));
-        });
+        }))
+            .subscribe((msg) => this._input$.next(msg)));
         peer.on('connect', () => {
+            var _a, _b;
             console.info((0, data_model_1.formatTime)(Date.now()), 'Terminal', 'WebRTC', direction, 'connected', session_id, remote_terminal_id);
+            // @ts-ignore
+            const maxMessageSize = (_b = (_a = peer._pc.sctp) === null || _a === void 0 ? void 0 : _a.maxMessageSize) !== null && _b !== void 0 ? _b : 65536;
+            this._mapTerminalIdToPeer[remote_terminal_id] = { session_id, maxMessageSize, peer };
+            console.info((0, data_model_1.formatTime)(Date.now()), 'Terminal', 'WebRTC', direction, 'maxMessageSize', maxMessageSize);
         });
@@ -480,2 +546,5 @@ peer.on('close', () => {
             this._mapTerminalIdToPeer[remote_terminal_id] = undefined;
+            for (const sub of subs) {
+                sub.unsubscribe();
+            }
             onDestroy();
@@ -486,2 +555,5 @@ });
             this._mapTerminalIdToPeer[remote_terminal_id] = undefined;
+            for (const sub of subs) {
+                sub.unsubscribe();
+            }
             onDestroy();
@@ -488,0 +560,0 @@ });

package.json

 {
 	"name": "@yuants/protocol",
-	"version": "0.23.2",
+	"version": "0.23.3",
 	"main": "lib/index.js",
@@ -5,0 +5,0 @@ "module": "dist/index.js",

temp/package-deps.json

 {
-    "libraries/protocol/CHANGELOG.json": "b9af9baa75a5b7d6ee9268521fa623a6f33bf521",
-    "libraries/protocol/CHANGELOG.md": "33f01f3ea7527eb04564bc28411354380c070990",
+    "libraries/protocol/CHANGELOG.json": "90805305a1d428139e44d8dabcdc82f06b211d93",
+    "libraries/protocol/CHANGELOG.md": "ef5a1f6ab25a4fa0daaa3539ea6a106c47f3fee2",
     "libraries/protocol/api-extractor.json": "62f4fd324425b9a235f0c117975967aab09ced0c",
@@ -9,3 +9,3 @@ "libraries/protocol/config/jest.config.json": "4bb17bde3ee911163a3edb36a6eb71491d80b1bd",
     "libraries/protocol/etc/protocol.api.md": "2275d7a334a65f54530dd3d79109c947f45553b0",
-    "libraries/protocol/package.json": "ec1a26ebe88535261fb31fff6cba6b468bfb6360",
+    "libraries/protocol/package.json": "c331807af451ccd0710b624213c9a4bc19b0d4b1",
     "libraries/protocol/src/create-connection.ts": "4fe9f0e005d62910c1bc76abe31df79e4ab64f1e",
@@ -23,3 +23,3 @@ "libraries/protocol/src/index.ts": "d4d17468883fe9a097862e3a24e6836898ee610b",
     "libraries/protocol/src/terminal.test.ts": "a88d7681da88e37ded88571fb5374a9e94a4d428",
-    "libraries/protocol/src/terminal.ts": "dd5322e3ebaac36f0b81dd4f4549cc5ca0900a57",
+    "libraries/protocol/src/terminal.ts": "be96b9f0f79e63f965e41f39b76751690023889e",
     "libraries/protocol/src/utils/DataRecord.ts": "5a942dea5beee36c7f823c8aa70edad6e72ec9ef",
@@ -26,0 +26,0 @@ "libraries/protocol/src/utils/Order.ts": "7b2b6b2659a8953aa95def726b1d02cd4fa4654c",

No alert changes

Improved metrics

Total package byte prevSize

614371

increased by3.53%

Lines of code

7448

increased by1.97%

No dependency changes