Implemented serial-monitoring for the backend.

Signed-off-by: Akos Kitta <kittaakos@typefox.io>

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

@@ -19,6 +19,9 @@ import { DefaultWorkspaceServerExt } from './default-workspace-server-ext';
 import { WorkspaceServer } from '@theia/workspace/lib/common';
 import { SketchesServiceImpl } from './sketches-service-impl';
 import { SketchesService, SketchesServicePath } from '../common/protocol/sketches-service';
+import { MonitorServiceImpl } from './monitor/monitor-service-impl';
+import { MonitorService, MonitorServicePath, MonitorServiceClient } from '../common/protocol/monitor-service';
+import { MonitorClientProvider } from './monitor/monitor-client-provider';
 
 export default new ContainerModule((bind, unbind, isBound, rebind) => {
     bind(ArduinoDaemon).toSelf().inSingletonScope();
@@ -104,4 +107,25 @@ export default new ContainerModule((bind, unbind, isBound, rebind) => {
     // If nothing was set previously.
     bind(DefaultWorkspaceServerExt).toSelf().inSingletonScope();
     rebind(WorkspaceServer).toService(DefaultWorkspaceServerExt);
+
+    // Shared monitor client provider service for the backend.
+    bind(MonitorClientProvider).toSelf().inSingletonScope();
+
+    // Connection scoped service for the serial monitor.
+    const monitorServiceConnectionModule = ConnectionContainerModule.create(({ bind, bindBackendService }) => {
+        bind(MonitorServiceImpl).toSelf().inSingletonScope();
+        bind(MonitorService).toService(MonitorServiceImpl);
+        bindBackendService<MonitorService, MonitorServiceClient>(MonitorServicePath, MonitorService, (service, client) => {
+            service.setClient(client);
+            client.onDidCloseConnection(() => service.dispose());
+            return service;
+        });
+    });
+    bind(ConnectionContainerModule).toConstantValue(monitorServiceConnectionModule);
+
+    // Logger for the monitor service.
+    bind(ILogger).toDynamicValue(ctx => {
+        const parentLogger = ctx.container.get<ILogger>(ILogger);
+        return parentLogger.child('monitor-service');
+    }).inSingletonScope().whenTargetNamed('monitor-service');
 });

arduino-ide-extension/src/node/monitor/monitor-client-provider.ts

@@ -0,0 +1,20 @@
+import * as grpc from '@grpc/grpc-js';
+import { injectable, postConstruct } from 'inversify';
+import { Deferred } from '@theia/core/lib/common/promise-util';
+import { MonitorClient } from '../cli-protocol/monitor/monitor_grpc_pb';
+
+@injectable()
+export class MonitorClientProvider {
+
+    readonly deferred = new Deferred<MonitorClient>();
+
+    @postConstruct()
+    protected init(): void {
+        this.deferred.resolve(new MonitorClient('localhost:50051', grpc.credentials.createInsecure()));
+    }
+
+    get client(): Promise<MonitorClient> {
+        return this.deferred.promise;
+    }
+
+}

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

@@ -0,0 +1,164 @@
+import { v4 } from 'uuid';
+import * as grpc from '@grpc/grpc-js';
+import { TextDecoder, TextEncoder } from 'util';
+import { injectable, inject, named } from 'inversify';
+import { ILogger, Disposable, DisposableCollection } from '@theia/core';
+import { MonitorService, MonitorServiceClient, ConnectionConfig, ConnectionType } from '../../common/protocol/monitor-service';
+import { StreamingOpenReq, StreamingOpenResp, MonitorConfig } from '../cli-protocol/monitor/monitor_pb';
+import { MonitorClientProvider } from './monitor-client-provider';
+
+export interface MonitorDuplex {
+    readonly toDispose: Disposable;
+    readonly duplex: grpc.ClientDuplexStream<StreamingOpenReq, StreamingOpenResp>;
+}
+
+type ErrorCode = { code: number };
+type MonitorError = Error & ErrorCode;
+namespace MonitorError {
+
+    export function is(error: Error & Partial<ErrorCode>): error is MonitorError {
+        return typeof error.code === 'number';
+    }
+
+    /**
+     * The frontend has refreshed the browser, for instance.
+     */
+    export function isClientCancelledError(error: MonitorError): boolean {
+        return error.code === 1 && error.message === 'Cancelled on client';
+    }
+
+    /**
+     * When detaching a physical device when the duplex channel is still opened.
+     */
+    export function isDeviceNotConfiguredError(error: MonitorError): boolean {
+        return error.code === 2 && error.message === 'device not configured';
+    }
+
+}
+
+@injectable()
+export class MonitorServiceImpl implements MonitorService {
+
+    @inject(ILogger)
+    @named('monitor-service')
+    protected readonly logger: ILogger;
+
+    @inject(MonitorClientProvider)
+    protected readonly monitorClientProvider: MonitorClientProvider;
+
+    protected client?: MonitorServiceClient;
+    protected readonly connections = new Map<string, MonitorDuplex>();
+
+    setClient(client: MonitorServiceClient | undefined): void {
+        this.client = client;
+    }
+
+    dispose(): void {
+        for (const [connectionId, duplex] of this.connections.entries()) {
+            this.doDisconnect(connectionId, duplex);
+        }
+    }
+
+    async connect(config: ConnectionConfig): Promise<{ connectionId: string }> {
+        const client = await this.monitorClientProvider.client;
+        const duplex = client.streamingOpen();
+        const connectionId = v4();
+        const toDispose = new DisposableCollection(
+            Disposable.create(() => this.disconnect(connectionId))
+        );
+
+        duplex.on('error', ((error: Error) => {
+            if (MonitorError.is(error) && (
+                MonitorError.isClientCancelledError(error)
+                || MonitorError.isDeviceNotConfiguredError(error)
+            )) {
+                if (this.client) {
+                    this.client.notifyError(error);
+                }
+            }
+            this.logger.error(`Error occurred for connection ${connectionId}.`, error);
+            toDispose.dispose();
+        }).bind(this));
+
+        duplex.on('data', ((resp: StreamingOpenResp) => {
+            if (this.client) {
+                const raw = resp.getData();
+                const data = typeof raw === 'string' ? raw : new TextDecoder('utf8').decode(raw);
+                this.client.notifyRead({ connectionId, data });
+            }
+        }).bind(this));
+
+        const { type, port } = config;
+        const req = new StreamingOpenReq();
+        const monitorConfig = new MonitorConfig();
+        monitorConfig.setType(this.mapType(type));
+        monitorConfig.setTarget(port);
+        if (config.baudRate !== undefined) {
+            monitorConfig.setAdditionalconfig({ 'BaudRate': config.baudRate });
+        }
+        req.setMonitorconfig(monitorConfig);
+
+        return new Promise<{ connectionId: string }>(resolve => {
+            duplex.write(req, () => {
+                this.connections.set(connectionId, { toDispose, duplex });
+                resolve({ connectionId });
+            });
+        });
+    }
+
+    async disconnect(connectionId: string): Promise<boolean> {
+        this.logger.info(`>>> Received disconnect request for connection: ${connectionId}`);
+        const disposable = this.connections.get(connectionId);
+        if (!disposable) {
+            this.logger.warn(`<<< No connection was found for ID: ${connectionId}`);
+            return false;
+        }
+        const result = await this.doDisconnect(connectionId, disposable);
+        if (result) {
+            this.logger.info(`<<< Successfully disconnected from ${connectionId}.`);
+        } else {
+            this.logger.info(`<<< Could not disconnected from ${connectionId}.`);
+        }
+        return result;
+    }
+
+    protected async doDisconnect(connectionId: string, duplex: MonitorDuplex): Promise<boolean> {
+        const { toDispose } = duplex;
+        this.logger.info(`>>> Disposing monitor connection: ${connectionId}...`);
+        try {
+            toDispose.dispose();
+            this.logger.info(`<<< Connection disposed: ${connectionId}.`);
+            return true;
+        } catch (e) {
+            this.logger.error(`<<< Error occurred when disposing monitor connection: ${connectionId}. ${e}`);
+            return false;
+        }
+    }
+
+    async send(connectionId: string, data: string): Promise<void> {
+        const duplex = this.duplex(connectionId);
+        if (duplex) {
+            const req = new StreamingOpenReq();
+            req.setData(new TextEncoder().encode(data));
+            return new Promise<void>(resolve => duplex.duplex.write(req, resolve));
+        } else {
+            throw new Error(`No connection with ID: ${connectionId}.`);
+        }
+    }
+
+    protected mapType(type?: ConnectionType): MonitorConfig.TargetType {
+        switch (type) {
+            case ConnectionType.SERIAL: return MonitorConfig.TargetType.SERIAL;
+            default: return MonitorConfig.TargetType.SERIAL;
+        }
+    }
+
+    protected duplex(connectionId: string): MonitorDuplex | undefined {
+        const monitorClient = this.connections.get(connectionId);
+        if (!monitorClient) {
+            this.logger.warn(`Could not find monitor client for connection ID: ${connectionId}`);
+        }
+        return monitorClient;
+    }
+
+}