Skip to content

Commit 116b3d5

Browse files
silvanocerzaAlberto Iannaccone
silvanocerza
authored and
Alberto Iannaccone
committed
Moved some interfaces
1 parent 750796d commit 116b3d5

File tree

2 files changed

+328
-5
lines changed

2 files changed

+328
-5
lines changed

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

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -86,11 +86,7 @@ import { ArduinoLocalizationContribution } from './arduino-localization-contribu
8686
import { LocalizationContribution } from '@theia/core/lib/node/i18n/localization-contribution';
8787
import { MonitorManagerProxyImpl } from './monitor-manager-proxy-impl';
8888
import { MonitorManager } from './monitor-manager';
89-
import {
90-
MonitorManagerProxy,
91-
MonitorManagerProxyClient,
92-
MonitorManagerProxyPath,
93-
} from '../common/monitor-manager-proxy';
89+
import { MonitorManagerProxy, MonitorManagerProxyClient, MonitorManagerProxyPath } from '../common/protocol/monitor-service';
9490

9591
export default new ContainerModule((bind, unbind, isBound, rebind) => {
9692
bind(BackendApplication).toSelf().inSingletonScope();

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

Lines changed: 327 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,327 @@
1+
import { ClientDuplexStream } from "@grpc/grpc-js";
2+
import { Disposable, Emitter, ILogger } from "@theia/core";
3+
import { inject, named } from "@theia/core/shared/inversify";
4+
import { Board, Port, Status, MonitorSettings } from "../common/protocol";
5+
import { MonitorPortConfiguration, MonitorPortSetting, MonitorRequest, MonitorResponse } from "./cli-protocol/cc/arduino/cli/commands/v1/monitor_pb";
6+
import { CoreClientAware } from "./core-client-provider";
7+
import { WebSocketProvider } from "./web-socket/web-socket-provider";
8+
import { Port as gRPCPort } from 'arduino-ide-extension/src/node/cli-protocol/cc/arduino/cli/commands/v1/port_pb'
9+
import WebSocketProviderImpl from "./web-socket/web-socket-provider-impl";
10+
11+
export class MonitorService extends CoreClientAware implements Disposable {
12+
// Bidirectional gRPC stream used to receive and send data from the running
13+
// pluggable monitor managed by the Arduino CLI.
14+
protected duplex: ClientDuplexStream<MonitorRequest, MonitorResponse> | null;
15+
16+
// Settings used by the currently running pluggable monitor.
17+
// They can be freely modified while running.
18+
protected settings: MonitorSettings;
19+
20+
// List of messages received from the running pluggable monitor.
21+
// These are flushed from time to time to the frontend.
22+
protected messages: string[] = [];
23+
24+
// Handles messages received from the frontend via websocket.
25+
protected onMessageReceived?: Disposable;
26+
27+
// Sends messages to the frontend from time to time.
28+
protected flushMessagesInterval?: NodeJS.Timeout;
29+
30+
// Triggered each time the number of clients connected
31+
// to the this service WebSocket changes.
32+
protected onWSClientsNumberChanged?: Disposable;
33+
34+
// Used to notify that the monitor is being disposed
35+
protected readonly onDisposeEmitter = new Emitter<void>();
36+
readonly onDispose = this.onDisposeEmitter.event;
37+
38+
protected readonly webSocketProvider: WebSocketProvider = new WebSocketProviderImpl();
39+
40+
constructor(
41+
@inject(ILogger)
42+
@named("monitor-service")
43+
protected readonly logger: ILogger,
44+
45+
private readonly board: Board,
46+
private readonly port: Port,
47+
) {
48+
super();
49+
50+
this.onWSClientsNumberChanged = this.webSocketProvider.onClientsNumberChanged(async (clients: number) => {
51+
if (clients === 0) {
52+
// There are no more clients that want to receive
53+
// data from this monitor, we can freely close
54+
// and dispose it.
55+
this.dispose();
56+
}
57+
});
58+
}
59+
60+
getWebsocketAddress(): number {
61+
return this.webSocketProvider.getAddress().port;
62+
}
63+
64+
dispose(): void {
65+
this.stop();
66+
this.onDisposeEmitter.fire();
67+
}
68+
69+
/**
70+
* isStarted is used to know if the currently running pluggable monitor is started.
71+
* @returns true if pluggable monitor communication duplex is open,
72+
* false in all other cases.
73+
*/
74+
isStarted(): boolean {
75+
return !!this.duplex;
76+
}
77+
78+
/**
79+
* Start and connects a monitor using currently set board and port.
80+
* If a monitor is already started or board fqbn, port address and/or protocol
81+
* are missing nothing happens.
82+
* @returns a status to verify connection has been established.
83+
*/
84+
async start(): Promise<Status> {
85+
if (this.duplex) {
86+
return Status.ALREADY_CONNECTED;
87+
}
88+
89+
if (!this.board?.fqbn || !this.port?.address || !this.port?.protocol) {
90+
return Status.CONFIG_MISSING
91+
}
92+
93+
this.logger.info("starting monitor");
94+
const coreClient = await this.coreClient();
95+
const { client, instance } = coreClient;
96+
97+
this.duplex = client.monitor()
98+
this.duplex
99+
.on('close', () => {
100+
this.logger.info(`monitor to ${this.port?.address} using ${this.port?.protocol} closed by client`)
101+
})
102+
.on('end', () => {
103+
this.logger.info(`monitor to ${this.port?.address} using ${this.port?.protocol} closed by server`)
104+
})
105+
.on('error', (err: Error) => {
106+
this.logger.error(err);
107+
// TODO
108+
// this.theiaFEClient?.notifyError()
109+
})
110+
.on('data', ((res: MonitorResponse) => {
111+
if (res.getError()) {
112+
// TODO: Maybe disconnect
113+
this.logger.error(res.getError());
114+
return;
115+
}
116+
const data = res.getRxData()
117+
const message =
118+
typeof data === 'string' ? data : new TextDecoder('utf8').decode(data);
119+
this.messages.push(...splitLines(message))
120+
}).bind(this));
121+
122+
const req = new MonitorRequest();
123+
req.setInstance(instance);
124+
if (this.board?.fqbn) {
125+
req.setFqbn(this.board.fqbn)
126+
}
127+
if (this.port?.address && this.port?.protocol) {
128+
const port = new gRPCPort()
129+
port.setAddress(this.port.address);
130+
port.setProtocol(this.port.protocol);
131+
req.setPort(port);
132+
}
133+
const config = new MonitorPortConfiguration();
134+
for (const id in this.settings) {
135+
const s = new MonitorPortSetting();
136+
s.setSettingId(id);
137+
s.setValue(this.settings[id].selectedValue);
138+
config.addSettings(s);
139+
}
140+
req.setPortConfiguration(config)
141+
142+
const connect = new Promise<Status>(resolve => {
143+
if (this.duplex?.write(req)) {
144+
this.startMessagesHandlers();
145+
this.logger.info(`started monitor to ${this.port?.address} using ${this.port?.protocol}`)
146+
resolve(Status.OK);
147+
}
148+
this.logger.warn(`failed starting monitor to ${this.port?.address} using ${this.port?.protocol}`)
149+
resolve(Status.NOT_CONNECTED);
150+
});
151+
152+
const connectTimeout = new Promise<Status>(resolve => {
153+
setTimeout(async () => {
154+
this.logger.warn(`timeout starting monitor to ${this.port?.address} using ${this.port?.protocol}`)
155+
resolve(Status.NOT_CONNECTED);
156+
}, 1000);
157+
});
158+
// Try opening a monitor connection with a timeout
159+
return await Promise.race([
160+
connect,
161+
connectTimeout,
162+
])
163+
}
164+
165+
/**
166+
* Pauses the currently running monitor, it still closes the gRPC connection
167+
* with the underlying monitor process but it doesn't stop the message handlers
168+
* currently running.
169+
* This is mainly used to handle upload when to the board/port combination
170+
* the monitor is listening to.
171+
* @returns
172+
*/
173+
async pause(): Promise<void> {
174+
return new Promise(resolve => {
175+
if (!this.duplex) {
176+
this.logger.warn(`monitor to ${this.port?.address} using ${this.port?.protocol} already stopped`)
177+
return resolve();
178+
}
179+
// It's enough to close the connection with the client
180+
// to stop the monitor process
181+
this.duplex.cancel();
182+
this.duplex = null;
183+
this.logger.info(`stopped monitor to ${this.port?.address} using ${this.port?.protocol}`)
184+
resolve();
185+
})
186+
}
187+
188+
/**
189+
* Stop the monitor currently running
190+
*/
191+
async stop(): Promise<void> {
192+
return this.pause().finally(
193+
this.stopMessagesHandlers
194+
);
195+
}
196+
197+
/**
198+
* Send a message to the running monitor, a well behaved monitor
199+
* will then send that message to the board.
200+
* We MUST NEVER send a message that wasn't a user's input to the board.
201+
* @param message string sent to running monitor
202+
* @returns a status to verify message has been sent.
203+
*/
204+
async send(message: string): Promise<Status> {
205+
if (!this.duplex) {
206+
return Status.NOT_CONNECTED;
207+
}
208+
const coreClient = await this.coreClient();
209+
const { instance } = coreClient;
210+
211+
const req = new MonitorRequest();
212+
req.setInstance(instance);
213+
req.setTxData(new TextEncoder().encode(message));
214+
return new Promise<Status>(resolve => {
215+
if (this.duplex) {
216+
this.duplex?.write(req, () => {
217+
resolve(Status.OK);
218+
});
219+
return;
220+
}
221+
this.stop().then(() => resolve(Status.NOT_CONNECTED));
222+
})
223+
}
224+
225+
/**
226+
* Set monitor settings, if there is a running monitor they'll be sent
227+
* to it, otherwise they'll be used when starting one.
228+
* Only values in settings parameter will be change, other values won't
229+
* be changed in any way.
230+
* @param settings map of monitor settings to change
231+
* @returns a status to verify settings have been sent.
232+
*/
233+
async changeSettings(settings: MonitorSettings): Promise<Status> {
234+
const config = new MonitorPortConfiguration();
235+
for (const id in settings) {
236+
const s = new MonitorPortSetting();
237+
s.setSettingId(id);
238+
s.setValue(settings[id].selectedValue);
239+
config.addSettings(s);
240+
this.settings[id] = settings[id];
241+
}
242+
243+
if (!this.duplex) {
244+
return Status.NOT_CONNECTED;
245+
}
246+
const coreClient = await this.coreClient();
247+
const { instance } = coreClient;
248+
249+
const req = new MonitorRequest();
250+
req.setInstance(instance);
251+
req.setPortConfiguration(config)
252+
this.duplex.write(req);
253+
return Status.OK
254+
}
255+
256+
/**
257+
* Starts the necessary handlers to send and receive
258+
* messages to and from the frontend and the running monitor
259+
*/
260+
private startMessagesHandlers(): void {
261+
if (!this.flushMessagesInterval) {
262+
const flushMessagesToFrontend = () => {
263+
if (this.messages.length) {
264+
this.webSocketProvider.sendMessage(JSON.stringify(this.messages));
265+
this.messages = [];
266+
}
267+
};
268+
this.flushMessagesInterval = setInterval(flushMessagesToFrontend, 32);
269+
}
270+
271+
if (!this.onMessageReceived) {
272+
this.onMessageReceived = this.webSocketProvider.onMessageReceived(
273+
(msg: string) => {
274+
const message: SerialPlotter.Protocol.Message = JSON.parse(msg);
275+
276+
switch (message.command) {
277+
case SerialPlotter.Protocol.Command.PLOTTER_SEND_MESSAGE:
278+
this.send(message.data);
279+
break;
280+
281+
case SerialPlotter.Protocol.Command.PLOTTER_SET_BAUDRATE:
282+
this.theiaFEClient?.notifyBaudRateChanged(
283+
parseInt(message.data, 10) as SerialConfig.BaudRate
284+
);
285+
break;
286+
287+
case SerialPlotter.Protocol.Command.PLOTTER_SET_LINE_ENDING:
288+
this.theiaFEClient?.notifyLineEndingChanged(message.data);
289+
break;
290+
291+
case SerialPlotter.Protocol.Command.PLOTTER_SET_INTERPOLATE:
292+
this.theiaFEClient?.notifyInterpolateChanged(message.data);
293+
break;
294+
295+
default:
296+
break;
297+
}
298+
}
299+
)
300+
}
301+
}
302+
303+
/**
304+
* Stops the necessary handlers to send and receive messages to
305+
* and from the frontend and the running monitor
306+
*/
307+
private stopMessagesHandlers(): void {
308+
if (this.flushMessagesInterval) {
309+
clearInterval(this.flushMessagesInterval);
310+
this.flushMessagesInterval = undefined;
311+
}
312+
if (this.onMessageReceived) {
313+
this.onMessageReceived.dispose();
314+
this.onMessageReceived = undefined;
315+
}
316+
}
317+
318+
}
319+
320+
/**
321+
* Splits a string into an array without removing newline char.
322+
* @param s string to split into lines
323+
* @returns an lines array
324+
*/
325+
function splitLines(s: string): string[] {
326+
return s.split(/(?<=\n)/);
327+
}

0 commit comments

Comments
 (0)