2024-11-21 14:21:30 +01:00

794 lines
26 KiB
TypeScript

import { ClientDuplexStream, status } from '@grpc/grpc-js';
import {
ApplicationError,
Disposable,
Emitter,
ILogger,
nls,
} from '@theia/core';
import { inject, named, postConstruct } from '@theia/core/shared/inversify';
import {
Board,
Port,
Monitor,
createAlreadyConnectedError,
createMissingConfigurationError,
createNotConnectedError,
createConnectionFailedError,
isMonitorConnected,
MonitorSettings,
PluggableMonitorSettings,
} from '../common/protocol';
import {
EnumerateMonitorPortSettingsRequest,
EnumerateMonitorPortSettingsResponse,
MonitorPortOpenRequest,
MonitorRequest,
MonitorResponse,
} from './cli-protocol/cc/arduino/cli/commands/v1/monitor_pb';
import { CoreClientAware } from './core-client-provider';
import { WebSocketProvider } from './web-socket/web-socket-provider';
import { Port as RpcPort } from './cli-protocol/cc/arduino/cli/commands/v1/port_pb';
import { MonitorSettingsProvider } from './monitor-settings/monitor-settings-provider';
import {
Deferred,
retry,
timeoutReject,
} from '@theia/core/lib/common/promise-util';
import { MonitorServiceFactoryOptions } from './monitor-service-factory';
import { ServiceError } from './service-error';
import {
MonitorPortConfiguration,
MonitorPortSetting,
} from './cli-protocol/cc/arduino/cli/commands/v1/common_pb';
export const MonitorServiceName = 'monitor-service';
type DuplexHandlerKeys =
| 'close'
| 'end'
| 'error'
| 'data'
| 'status'
| 'metadata';
interface DuplexHandler {
key: DuplexHandlerKeys;
callback: (...args: any) => void;
}
const MAX_WRITE_TO_STREAM_TRIES = 10;
const WRITE_TO_STREAM_TIMEOUT_MS = 30000;
export class MonitorService extends CoreClientAware implements Disposable {
@inject(ILogger)
@named(MonitorServiceName)
private readonly logger: ILogger;
@inject(MonitorSettingsProvider)
private readonly monitorSettingsProvider: MonitorSettingsProvider;
@inject(WebSocketProvider)
private readonly webSocketProvider: WebSocketProvider;
// Bidirectional gRPC stream used to receive and send data from the running
// pluggable monitor managed by the Arduino CLI.
private duplex: ClientDuplexStream<MonitorRequest, MonitorResponse> | null;
// Settings used by the currently running pluggable monitor.
// They can be freely modified while running.
private settings: MonitorSettings = {};
// List of messages received from the running pluggable monitor.
// These are flushed from time to time to the frontend.
private messages: string[] = [];
// Handles messages received from the frontend via websocket.
private onMessageReceived?: Disposable;
// Sends messages to the frontend from time to time.
private flushMessagesInterval?: NodeJS.Timeout;
// Triggered each time the number of clients connected
// to the this service WebSocket changes.
private onWSClientsNumberChanged?: Disposable;
// Used to notify that the monitor is being disposed
private readonly onDisposeEmitter = new Emitter<void>();
readonly onDispose = this.onDisposeEmitter.event;
private _initialized = new Deferred<void>();
private creating: Deferred<void>;
private readonly board: Board;
private readonly port: Port;
private readonly monitorID: string;
private readonly streamingTextDecoder = new TextDecoder('utf8');
/**
* The lightweight representation of the port configuration currently in use for the running monitor.
* IDE2 stores this object after starting the monitor. On every monitor settings change request, IDE2 compares
* the current config with the new settings, and only sends the diff as the new config to overcome https://github.com/arduino/arduino-ide/issues/375.
*/
private currentPortConfigSnapshot:
| MonitorPortConfiguration.AsObject
| undefined;
constructor(
@inject(MonitorServiceFactoryOptions) options: MonitorServiceFactoryOptions
) {
super();
this.board = options.board;
this.port = options.port;
this.monitorID = options.monitorID;
}
@postConstruct()
protected init(): void {
this.onWSClientsNumberChanged =
this.webSocketProvider.onClientsNumberChanged(async (clients: number) => {
if (clients === 0) {
// There are no more clients that want to receive
// data from this monitor, we can freely close
// and dispose it.
this.dispose();
return;
}
this.updateClientsSettings(this.settings);
});
this.portMonitorSettings(this.port.protocol, this.board.fqbn!, true).then(
async (settings) => {
this.settings = {
...this.settings,
pluggableMonitorSettings:
await this.monitorSettingsProvider.getSettings(
this.monitorID,
settings
),
};
this._initialized.resolve();
}
);
}
get initialized(): Promise<void> {
return this._initialized.promise;
}
getWebsocketAddressPort(): number {
return this.webSocketProvider.getAddress().port;
}
dispose(): void {
this.stop();
this.onDisposeEmitter.fire();
this.onWSClientsNumberChanged?.dispose();
}
/**
* isStarted is used to know if the currently running pluggable monitor is started.
* @returns true if pluggable monitor communication duplex is open,
* false in all other cases.
*/
isStarted(): boolean {
return !!this.duplex;
}
/**
* Start and connects a monitor using currently set board and port.
* If a monitor is already started, the promise will reject with an `AlreadyConnectedError`.
* If the board fqbn, port address and/or protocol are missing, the promise rejects with a `MissingConfigurationError`.
*/
async start(): Promise<void> {
if (this.creating?.state === 'unresolved') return this.creating.promise;
this.creating = new Deferred();
if (this.duplex) {
this.updateClientsSettings({
monitorUISettings: {
connectionStatus: 'connected',
connected: true, // TODO: should be removed when plotter app understand the `connectionStatus` message
serialPort: this.port.address,
},
});
this.creating.reject(createAlreadyConnectedError(this.port));
return this.creating.promise;
}
if (!this.board?.fqbn || !this.port?.address || !this.port?.protocol) {
this.updateClientsSettings({
monitorUISettings: {
connectionStatus: 'not-connected',
connected: false, // TODO: should be removed when plotter app understand the `connectionStatus` message
},
});
this.creating.reject(createMissingConfigurationError(this.port));
return this.creating.promise;
}
this.logger.info('starting monitor');
try {
// get default monitor settings from the CLI
const defaultSettings = await this.portMonitorSettings(
this.port.protocol,
this.board.fqbn
);
this.updateClientsSettings({
monitorUISettings: { connectionStatus: 'connecting' },
});
// get actual settings from the settings provider
this.settings = {
...this.settings,
pluggableMonitorSettings: {
...this.settings.pluggableMonitorSettings,
...(await this.monitorSettingsProvider.getSettings(
this.monitorID,
defaultSettings
)),
},
};
const coreClient = await this.coreClient;
const { instance } = coreClient;
const openPortRequest = new MonitorPortOpenRequest();
openPortRequest.setInstance(instance);
if (this.board?.fqbn) {
openPortRequest.setFqbn(this.board.fqbn);
}
if (this.port?.address && this.port?.protocol) {
const rpcPort = new RpcPort();
rpcPort.setAddress(this.port.address);
rpcPort.setProtocol(this.port.protocol);
openPortRequest.setPort(rpcPort);
}
const config = new MonitorPortConfiguration();
for (const id in this.settings.pluggableMonitorSettings) {
const s = new MonitorPortSetting();
s.setSettingId(id);
s.setValue(this.settings.pluggableMonitorSettings[id].selectedValue);
config.addSettings(s);
}
openPortRequest.setPortConfiguration(config);
await this.pollWriteToStream(openPortRequest);
// Only store the config, if the monitor has successfully started.
this.currentPortConfigSnapshot = MonitorPortConfiguration.toObject(
false,
config
);
this.logger.info(
`Using port configuration for ${this.port.protocol}:${
this.port.address
}: ${JSON.stringify(this.currentPortConfigSnapshot)}`
);
this.startMessagesHandlers();
this.logger.info(
`started monitor to ${this.port?.address} using ${this.port?.protocol}`
);
this.updateClientsSettings({
monitorUISettings: {
connectionStatus: 'connected',
connected: true, // TODO: should be removed when plotter app understand the `connectionStatus` message
serialPort: this.port.address,
},
});
this.creating.resolve();
return this.creating.promise;
} catch (err) {
this.logger.warn(
`failed starting monitor to ${this.port?.address} using ${this.port?.protocol}`
);
const appError = ApplicationError.is(err)
? err
: createConnectionFailedError(
this.port,
ServiceError.is(err)
? err.details
: err instanceof Error
? err.message
: String(err)
);
this.creating.reject(appError);
this.updateClientsSettings({
monitorUISettings: {
connectionStatus: { errorMessage: appError.message },
},
});
return this.creating.promise;
}
}
async createDuplex(): Promise<
ClientDuplexStream<MonitorRequest, MonitorResponse>
> {
const coreClient = await this.coreClient;
return coreClient.client.monitor();
}
setDuplexHandlers(
duplex: ClientDuplexStream<MonitorRequest, MonitorResponse>,
additionalHandlers: DuplexHandler[]
): void {
// default handlers
duplex
.on('close', () => {
if (duplex === this.duplex) {
this.duplex = null;
this.updateClientsSettings({
monitorUISettings: {
connected: false, // TODO: should be removed when plotter app understand the `connectionStatus` message
connectionStatus: 'not-connected',
},
});
}
this.logger.info(
`monitor to ${this.port?.address} using ${this.port?.protocol} closed by client`
);
})
.on('end', () => {
if (duplex === this.duplex) {
this.duplex = null;
this.updateClientsSettings({
monitorUISettings: {
connected: false, // TODO: should be removed when plotter app understand the `connectionStatus` message
connectionStatus: 'not-connected',
},
});
}
this.logger.info(
`monitor to ${this.port?.address} using ${this.port?.protocol} closed by server`
);
});
for (const handler of additionalHandlers) {
duplex.on(handler.key, handler.callback);
}
}
pollWriteToStream(request: MonitorPortOpenRequest): Promise<void> {
const createWriteToStreamExecutor =
(duplex: ClientDuplexStream<MonitorRequest, MonitorResponse>) =>
(resolve: () => void, reject: (reason?: unknown) => void) => {
const resolvingDuplexHandlers: DuplexHandler[] = [
{
key: 'error',
callback: async (err: Error) => {
this.logger.error(err);
const details = ServiceError.is(err) ? err.details : err.message;
reject(createConnectionFailedError(this.port, details));
},
},
{
key: 'data',
callback: async (monitorResponse: MonitorResponse) => {
if (monitorResponse.getError()) {
// TODO: Maybe disconnect
this.logger.error(monitorResponse.getError());
return;
}
if (monitorResponse.getSuccess()) {
resolve();
return;
}
const data = monitorResponse.getRxData();
const message =
typeof data === 'string'
? data
: this.streamingTextDecoder.decode(data, { stream: true });
this.messages.push(...splitLines(message));
},
},
];
this.setDuplexHandlers(duplex, resolvingDuplexHandlers);
duplex.write(new MonitorRequest().setOpenRequest(request));
};
return Promise.race([
retry(
async () => {
let createdDuplex = undefined;
try {
createdDuplex = await this.createDuplex();
await new Promise<void>(createWriteToStreamExecutor(createdDuplex));
this.duplex = createdDuplex;
} catch (err) {
createdDuplex?.end();
throw err;
}
},
2_000,
MAX_WRITE_TO_STREAM_TRIES
),
timeoutReject(
WRITE_TO_STREAM_TIMEOUT_MS,
nls.localize(
'arduino/monitor/connectionTimeout',
"Timeout. The IDE has not received the 'success' message from the monitor after successfully connecting to it"
)
),
]) as Promise<unknown> as Promise<void>;
}
private endingDuplex: Promise<void> | undefined;
/**
* Pauses the currently running monitor, it still closes the gRPC connection
* with the underlying monitor process but it doesn't stop the message handlers
* currently running.
* This is mainly used to handle upload with the board/port combination
* the monitor is listening to.
* @returns
*/
async pause(): Promise<void> {
const duplex = this.duplex;
if (!duplex) {
this.logger.warn(
`monitor to ${this.port?.address} using ${this.port?.protocol} already stopped`
);
return;
}
if (this.endingDuplex) {
return this.endingDuplex;
}
const deferredEnd = new Deferred<void>();
this.endingDuplex = deferredEnd.promise;
// to terminate the monitor connection, send a close request, and wait for the end event
duplex.once('end', () => {
deferredEnd.resolve();
});
try {
await new Promise((resolve) =>
duplex.write(new MonitorRequest().setClose(true), resolve)
);
await this.endingDuplex;
} finally {
this.endingDuplex = undefined;
}
// Sanity check
// Duplexes are allowed to be half open, check whether the monitor server (the readable) has ended
if (!duplex.readableEnded) {
throw new Error('Could not end the monitor connection');
}
}
/**
* Stop the monitor currently running
*/
async stop(): Promise<void> {
return this.pause().finally(() => this.stopMessagesHandlers());
}
/**
* Send a message to the running monitor, a well behaved monitor
* will then send that message to the board.
* We MUST NEVER send a message that wasn't a user's input to the board.
* @param message string sent to running monitor
* @returns a status to verify message has been sent.
*/
async send(message: string): Promise<void> {
if (!this.duplex) {
throw createNotConnectedError(this.port);
}
const req = new MonitorRequest();
req.setTxData(new TextEncoder().encode(message));
return new Promise<void>((resolve, reject) => {
if (this.duplex) {
this.duplex?.write(req, resolve);
return;
}
this.stop().then(() => reject(createNotConnectedError(this.port)));
});
}
/**
*
* @returns map of current monitor settings
*/
async currentSettings(): Promise<MonitorSettings> {
await this.initialized;
return this.settings;
}
// TODO: move this into MonitorSettingsProvider
/**
* Returns the possible configurations used to connect a monitor
* to the board specified by fqbn using the specified protocol
* @param protocol the protocol of the monitor we want get settings for
* @param fqbn the fqbn of the board we want to monitor
* @returns a map of all the settings supported by the monitor
*/
private async portMonitorSettings(
protocol: string,
fqbn: string,
swallowsPlatformNotFoundError = false
): Promise<PluggableMonitorSettings> {
const coreClient = await this.coreClient;
const { client, instance } = coreClient;
const req = new EnumerateMonitorPortSettingsRequest();
req.setInstance(instance);
req.setPortProtocol(protocol);
req.setFqbn(fqbn);
const resp = await new Promise<
EnumerateMonitorPortSettingsResponse | undefined
>((resolve, reject) => {
client.enumerateMonitorPortSettings(req, async (err, resp) => {
if (err) {
// Check whether the platform is installed: https://github.com/arduino/arduino-ide/issues/1974.
// No error codes. Look for `Unknown FQBN: platform arduino:mbed_nano is not installed` message similarities: https://github.com/arduino/arduino-cli/issues/1762.
if (
swallowsPlatformNotFoundError &&
ServiceError.is(err) &&
err.code === status.NOT_FOUND &&
err.details.includes('FQBN') &&
err.details.includes(fqbn.split(':', 2).join(':')) // create a platform ID from the FQBN
) {
resolve(undefined);
}
reject(err);
}
resolve(resp);
});
});
const settings: PluggableMonitorSettings = {};
if (!resp) {
return settings;
}
for (const iterator of resp.getSettingsList()) {
settings[iterator.getSettingId()] = {
id: iterator.getSettingId(),
label: iterator.getLabel(),
type: iterator.getType(),
values: iterator.getEnumValuesList(),
selectedValue: iterator.getValue(),
};
}
return settings;
}
/**
* Set monitor settings, if there is a running monitor they'll be sent
* to it, otherwise they'll be used when starting one.
* Only values in settings parameter will be change, other values won't
* be changed in any way.
* @param settings map of monitor settings to change
* @returns a status to verify settings have been sent.
*/
async changeSettings(settings: MonitorSettings): Promise<void> {
const config = new MonitorPortConfiguration();
const { pluggableMonitorSettings } = settings;
const reconciledSettings = await this.monitorSettingsProvider.setSettings(
this.monitorID,
pluggableMonitorSettings || {}
);
if (reconciledSettings) {
for (const id in reconciledSettings) {
const s = new MonitorPortSetting();
s.setSettingId(id);
s.setValue(reconciledSettings[id].selectedValue);
config.addSettings(s);
}
}
const connectionStatus = Boolean(this.duplex)
? 'connected'
: 'not-connected';
this.updateClientsSettings({
monitorUISettings: {
...settings.monitorUISettings,
connectionStatus,
serialPort: this.port.address,
connected: isMonitorConnected(connectionStatus), // TODO: should be removed when plotter app understand the `connectionStatus` message
},
pluggableMonitorSettings: reconciledSettings,
});
if (!this.duplex) {
// instead of throwing an error, return silently like the original logic
// https://github.com/arduino/arduino-ide/blob/9b49712669b06c97bda68a1e5f04eee4664c13f8/arduino-ide-extension/src/node/monitor-service.ts#L540
return;
}
const diffConfig = await this.maybeUpdatePortConfigSnapshot(config);
if (!diffConfig) {
this.logger.info(
`No port configuration changes have been detected. No need to send configure commands to the running monitor ${this.port.protocol}:${this.port.address}.`
);
return;
}
this.logger.info(
`Sending monitor request with new port configuration: ${JSON.stringify(
MonitorPortConfiguration.toObject(false, diffConfig)
)}`
);
const req = new MonitorRequest();
req.setUpdatedConfiguration(diffConfig);
this.duplex.write(req);
}
/**
* Function to calculate a diff between the `otherPortConfig` argument and the `currentPortConfigSnapshot`.
*
* If the current config snapshot and the snapshot derived from `otherPortConfig` are the same, no snapshot update happens,
* and the function returns with undefined. Otherwise, the current snapshot config value will be updated from the snapshot
* derived from the `otherPortConfig` argument, and this function returns with a `MonitorPortConfiguration` instance
* representing only the difference between the two snapshot configs to avoid sending unnecessary monitor to configure commands to the CLI.
* See [#1703 (comment)](https://github.com/arduino/arduino-ide/pull/1703#issuecomment-1327913005) for more details.
*/
private async maybeUpdatePortConfigSnapshot(
otherPortConfig: MonitorPortConfiguration
): Promise<MonitorPortConfiguration | undefined> {
const otherPortConfigSnapshot = MonitorPortConfiguration.toObject(
false,
otherPortConfig
);
if (!this.currentPortConfigSnapshot) {
throw new Error(
`The current port configuration object was undefined when tried to merge in ${JSON.stringify(
otherPortConfigSnapshot
)}.`
);
}
const justDiff = await import('just-diff');
const snapshotDiff = justDiff.diff(
this.currentPortConfigSnapshot,
otherPortConfigSnapshot
);
if (!snapshotDiff.length) {
return undefined;
}
const diffConfig = snapshotDiff.reduce((acc, curr) => {
if (!this.isValidMonitorPortSettingChange(curr)) {
throw new Error(
`Expected only 'replace' operation: a 'value' change in the 'settingsList'. Calculated diff a ${JSON.stringify(
snapshotDiff
)} between ${JSON.stringify(
this.currentPortConfigSnapshot
)} and ${JSON.stringify(
otherPortConfigSnapshot
)} snapshots. Current JSON-patch entry was ${JSON.stringify(curr)}.`
);
}
const { path, value } = curr;
const [, index] = path;
if (!this.currentPortConfigSnapshot?.settingsList) {
throw new Error(
`'settingsList' is missing from current port config snapshot: ${JSON.stringify(
this.currentPortConfigSnapshot
)}`
);
}
const changedSetting = this.currentPortConfigSnapshot.settingsList[index];
const setting = new MonitorPortSetting();
setting.setValue(value);
setting.setSettingId(changedSetting.settingId);
acc.addSettings(setting);
return acc;
}, new MonitorPortConfiguration());
this.currentPortConfigSnapshot = otherPortConfigSnapshot;
this.logger.info(
`Updated the port configuration for ${this.port.protocol}:${
this.port.address
}: ${JSON.stringify(this.currentPortConfigSnapshot)}`
);
return diffConfig;
}
private isValidMonitorPortSettingChange(entry: {
op: string;
path: (string | number)[];
value: unknown;
}): entry is {
op: 'replace';
path: ['settingsList', number, string];
value: string;
} {
const { op, path, value } = entry;
return (
op === 'replace' &&
path.length === 3 &&
path[0] === 'settingsList' &&
typeof path[1] === 'number' &&
path[2] === 'value' &&
typeof value === 'string'
);
}
/**
* Starts the necessary handlers to send and receive
* messages to and from the frontend and the running monitor
*/
private startMessagesHandlers(): void {
if (!this.flushMessagesInterval) {
const flushMessagesToFrontend = () => {
if (this.messages.length) {
this.webSocketProvider.sendMessage(JSON.stringify(this.messages));
this.messages = [];
}
};
this.flushMessagesInterval = setInterval(flushMessagesToFrontend, 32);
}
if (!this.onMessageReceived) {
this.onMessageReceived = this.webSocketProvider.onMessageReceived(
(msg: string) => {
const message: Monitor.Message = JSON.parse(msg);
switch (message.command) {
case Monitor.ClientCommand.SEND_MESSAGE:
this.send(message.data as string);
break;
case Monitor.ClientCommand.CHANGE_SETTINGS:
this.changeSettings(message.data as MonitorSettings);
break;
}
}
);
}
}
updateClientsSettings(settings: MonitorSettings): void {
this.settings = { ...this.settings, ...settings };
if (
settings.monitorUISettings?.connectionStatus &&
!('connected' in settings.monitorUISettings)
) {
// Make sure the deprecated `connected` prop is set.
settings.monitorUISettings.connected = isMonitorConnected(
settings.monitorUISettings.connectionStatus
);
}
if (
typeof settings.monitorUISettings?.connected === 'boolean' &&
!('connectionStatus' in settings.monitorUISettings)
) {
// Set the connectionStatus if the message was sent by the plotter which does not handle the new protocol. Assuming that the plotter can send anything.
// https://github.com/arduino/arduino-serial-plotter-webapp#monitor-settings
settings.monitorUISettings.connectionStatus = settings.monitorUISettings
.connected
? 'connected'
: 'not-connected';
}
const command: Monitor.Message = {
command: Monitor.MiddlewareCommand.ON_SETTINGS_DID_CHANGE,
data: settings,
};
this.webSocketProvider.sendMessage(JSON.stringify(command));
}
/**
* Stops the necessary handlers to send and receive messages to
* and from the frontend and the running monitor
*/
private stopMessagesHandlers(): void {
if (this.flushMessagesInterval) {
clearInterval(this.flushMessagesInterval);
this.flushMessagesInterval = undefined;
}
if (this.onMessageReceived) {
this.onMessageReceived.dispose();
this.onMessageReceived = undefined;
}
}
}
/**
* Splits a string into an array without removing newline char.
* @param s string to split into lines
* @returns an lines array
*/
function splitLines(s: string): string[] {
return s.split(/(?<=\n)/);
}