experimenting with websockets

Author: fstasi

arduino-ide-extension/package.json

@@ -64,6 +64,7 @@
     "fuzzy": "^0.1.3",
     "glob": "^7.1.6",
     "google-protobuf": "^3.11.4",
+    "grpc": "^1.24.11",
     "hash.js": "^1.1.7",
     "is-valid-path": "^0.1.1",
     "js-yaml": "^3.13.1",

arduino-ide-extension/src/browser/arduino-ide-frontend-module.ts

@@ -397,9 +397,19 @@ export default new ContainerModule((bind, unbind, isBound, rebind) => {
     createWidget: () => context.container.get(MonitorWidget),
   }));
   // Frontend binding for the serial monitor service
+  // bind(MonitorServiceClient)
+  //   .toDynamicValue((context) => {
+  //     const client = context.container.get(MonitorServiceClientImpl);
+  //     WebSocketConnectionProvider.createProxy(
+  //       context.container,
+  //       MonitorServicePath,
+  //       client
+  //     );
+  //     return client;
+  //   })
+  //   .inSingletonScope();
   bind(MonitorService)
     .toDynamicValue((context) => {
-      debugger;
       const connection = context.container.get(WebSocketConnectionProvider);
       const client =
         context.container.get<MonitorServiceClient>(MonitorServiceClient);

arduino-ide-extension/src/browser/monitor/monitor-connection.ts

@@ -59,7 +59,7 @@ export class MonitorConnection {
   /**
    * This emitter forwards all read events **iff** the connection is established.
    */
-  protected readonly onReadEmitter = new Emitter<{ message: string }>();
+  protected readonly onReadEmitter = new Emitter<{ messages: string[] }>();
 
   /**
    * Array for storing previous monitor errors received from the server, and based on the number of elements in this array,
@@ -71,6 +71,31 @@ export class MonitorConnection {
 
   @postConstruct()
   protected init(): void {
+    this.monitorServiceClient.onMessage(async (port) => {
+      const w = new WebSocket(`ws://localhost:${port}`);
+      let h = 0;
+      w.onmessage = (res) => {
+        const messages = JSON.parse(res.data);
+        h += messages.length;
+
+        if (h > 1000) {
+          h = 0;
+          console.log('read 1000 messages');
+        }
+        // console.log(`received ${messages.length} messages`);
+        // this.onReadEmitter.fire({ messages });
+      };
+    });
+
+    // let i = 0;
+    // this.monitorServiceClient.onMessage(async (msg) => {
+    //   // msg received
+    //   i++;
+    //   if (i % 1000 === 0) {
+    //     // console.log(msg);
+    //   }
+    // });
+
     this.monitorServiceClient.onError(async (error) => {
       let shouldReconnect = false;
       if (this.state) {
@@ -231,15 +256,22 @@ export class MonitorConnection {
     );
     const connectStatus = await this.monitorService.connect(config);
     if (Status.isOK(connectStatus)) {
+      let j = 0;
       const requestMessage = () => {
-        this.monitorService.request().then(({ message }) => {
+        this.monitorService.request().then(({ messages }) => {
           if (this.connected) {
-            this.onReadEmitter.fire({ message });
+            // this.onReadEmitter.fire({ messages });
+            j += messages.length;
+            if (j > 1000) {
+              j = 0;
+              // console.log(`read more than 1000 messages`);
+            }
+
             requestMessage();
           }
         });
       };
-      requestMessage();
+      // requestMessage();
       this.state = { config };
       console.info(
         `<<< Serial monitor connection created for ${Board.toString(
@@ -300,7 +332,7 @@ export class MonitorConnection {
     return this.onConnectionChangedEmitter.event;
   }
 
-  get onRead(): Event<{ message: string }> {
+  get onRead(): Event<{ messages: string[] }> {
     return this.onReadEmitter.event;
   }
 

arduino-ide-extension/src/browser/monitor/monitor-service-client-impl.ts

@@ -10,7 +10,14 @@ export class MonitorServiceClientImpl implements MonitorServiceClient {
   protected readonly onErrorEmitter = new Emitter<MonitorError>();
   readonly onError = this.onErrorEmitter.event;
 
+  protected readonly onMessageEmitter = new Emitter<string>();
+  readonly onMessage = this.onMessageEmitter.event;
+
   notifyError(error: MonitorError): void {
     this.onErrorEmitter.fire(error);
   }
+
+  notifyMessage(message: string): void {
+    this.onMessageEmitter.fire(message);
+  }
 }

arduino-ide-extension/src/browser/monitor/monitor-widget.tsx

@@ -326,7 +326,9 @@ export class SerialMonitorOutput extends React.Component<
     return (
       <React.Fragment>
         <div style={{ whiteSpace: 'pre', fontFamily: 'monospace' }}>
-          {this.state.lines.map((l) => `${l}\n`)}
+          {this.state.lines.map((line, i) => (
+            <MonitorTextLine text={line} key={i} />
+          ))}
         </div>
         <div
           style={{ float: 'left', clear: 'both' }}
@@ -341,24 +343,30 @@ export class SerialMonitorOutput extends React.Component<
   componentDidMount(): void {
     this.scrollToBottom();
     this.toDisposeBeforeUnmount.pushAll([
-      this.props.monitorConnection.onRead(({ message }) => {
-        const rawLines = message.split('\n');
-        const lines: string[] = [];
-        const timestamp = () =>
-          this.state.timestamp
-            ? `${dateFormat(new Date(), 'H:M:ss.l')} -> `
-            : '';
-        for (let i = 0; i < rawLines.length; i++) {
-          if (
-            i === 0
-            // &&  this.state.content.length !== 0
-          ) {
-            lines.push(rawLines[i]);
-          } else {
-            lines.push(timestamp() + rawLines[i]);
+      this.props.monitorConnection.onRead(({ messages }) => {
+        messages.forEach((message) => {
+          const rawLines = message.split('\n');
+          const lines: string[] = [];
+          const timestamp = () =>
+            this.state.timestamp
+              ? `${dateFormat(new Date(), 'H:M:ss.l')} -> `
+              : '';
+
+          for (let i = 0; i < rawLines.length; i++) {
+            if (i === 0 && this.state.lines.length !== 0) {
+              lines.push(rawLines[i]);
+            } else {
+              lines.push(timestamp() + rawLines[i]);
+            }
           }
-        }
-        this.setState({ lines: this.state.lines.concat(lines) });
+
+          this.setState((prevState) => ({
+            lines: [...prevState.lines, lines.join('\n')],
+          }));
+
+          // const content = this.state.content + lines.join('\n');
+          // this.setState({ content });
+        });
       }),
       this.props.clearConsoleEvent(() => this.setState({ lines: [] })),
       this.props.monitorModel.onChange(({ property }) => {
@@ -386,6 +394,11 @@ export class SerialMonitorOutput extends React.Component<
   }
 }
 
+const _MonitorTextLine = ({ text }: { text: string }): React.ReactElement => {
+  return <div>{text}</div>;
+};
+export const MonitorTextLine = React.memo(_MonitorTextLine);
+
 export interface SelectOption<T> {
   readonly label: string;
   readonly value: T;

arduino-ide-extension/src/common/protocol/monitor-service.ts

@@ -24,7 +24,7 @@ export interface MonitorService extends JsonRpcServer<MonitorServiceClient> {
   connect(config: MonitorConfig): Promise<Status>;
   disconnect(): Promise<Status>;
   send(message: string): Promise<Status>;
-  request(): Promise<{ message: string }>;
+  request(): Promise<{ messages: string[] }>;
 }
 
 export interface MonitorConfig {
@@ -61,8 +61,10 @@ export namespace MonitorConfig {
 
 export const MonitorServiceClient = Symbol('MonitorServiceClient');
 export interface MonitorServiceClient {
-  notifyError(event: MonitorError): void;
   onError: Event<MonitorError>;
+  onMessage: Event<string>;
+  notifyError(event: MonitorError): void;
+  notifyMessage(message: string): void;
 }
 
 export interface MonitorError {

arduino-ide-extension/src/node/arduino-ide-backend-module.ts

@@ -200,6 +200,7 @@ export default new ContainerModule((bind, unbind, isBound, rebind) => {
       bind(MonitorClientProvider).toSelf().inSingletonScope();
       bind(MonitorServiceImpl).toSelf().inSingletonScope();
       bind(MonitorService).toService(MonitorServiceImpl);
+
       bindBackendService<MonitorService, MonitorServiceClient>(
         MonitorServicePath,
         MonitorService,

arduino-ide-extension/src/node/monitor/monitor-service-impl.ts

@@ -1,5 +1,5 @@
 import { ClientDuplexStream } from '@grpc/grpc-js';
-import { TextDecoder, TextEncoder } from 'util';
+import { TextEncoder } from 'util';
 import { injectable, inject, named } from 'inversify';
 import { Struct } from 'google-protobuf/google/protobuf/struct_pb';
 import { Emitter } from '@theia/core/lib/common/event';
@@ -18,6 +18,7 @@ import {
 } from '../cli-protocol/cc/arduino/cli/monitor/v1/monitor_pb';
 import { MonitorClientProvider } from './monitor-client-provider';
 import { Board, Port } from '../../common/protocol/boards-service';
+import * as WebSocket from 'ws';
 
 interface ErrorWithCode extends Error {
   readonly code: number;
@@ -74,6 +75,13 @@ export class MonitorServiceImpl implements MonitorService {
   protected messages: string[] = [];
   protected onMessageDidReadEmitter = new Emitter<void>();
 
+  private rand = 0;
+
+  constructor() {
+    this.rand = Math.floor(1000 * Math.random());
+    console.log(`Spawning ${this.rand}`);
+  }
+
   setClient(client: MonitorServiceClient | undefined): void {
     this.client = client;
   }
@@ -122,14 +130,36 @@ export class MonitorServiceImpl implements MonitorService {
       }).bind(this)
     );
 
+    const ws = new WebSocket.Server({ port: 0 });
+    const address: any = ws.address();
+    this.client?.notifyMessage(address.port);
+    let wsConn: WebSocket | null = null;
+    ws.on('connection', (ws) => {
+      wsConn = ws;
+    });
+
+    const emptyTheQueue = () => {
+      if (this.messages.length) {
+        wsConn?.send(JSON.stringify(this.messages));
+        this.messages = [];
+      }
+    };
+
+    // empty the queue every 16ms (~60fps)
+    setInterval(emptyTheQueue, 32);
+
     duplex.on(
       'data',
       ((resp: StreamingOpenResponse) => {
+        // eslint-disable-next-line unused-imports/no-unused-vars
         const raw = resp.getData();
+
         const message =
           typeof raw === 'string' ? raw : new TextDecoder('utf8').decode(raw);
+        // this.client?.notifyMessage(message);
         this.messages.push(message);
-        this.onMessageDidReadEmitter.fire();
+        // this.onMessageDidReadEmitter.fire();
+        // wsConn?.send(message);
       }).bind(this)
     );
 
@@ -207,12 +237,14 @@ export class MonitorServiceImpl implements MonitorService {
     });
   }
 
-  async request(): Promise<{ message: string }> {
-    const message = this.messages.shift();
-    if (message) {
-      return { message };
+  async request(): Promise<{ messages: string[] }> {
+    // const messages = this.messages.shift();
+    const messages = this.messages;
+    if (messages.length) {
+      this.messages = [];
+      return { messages };
     }
-    return new Promise<{ message: string }>((resolve) => {
+    return new Promise<{ messages: string[] }>((resolve) => {
       const toDispose = this.onMessageDidReadEmitter.event(() => {
         toDispose.dispose();
         resolve(this.request());