@@ -64,6 +64,9 @@ export class MonitorService extends CoreClientAware implements Disposable {
64
64
protected _initialized = new Deferred < void > ( ) ;
65
65
protected creating : Deferred < Status > ;
66
66
67
+ MAX_WRITE_TO_STREAM_TRIES = 10 ;
68
+ WRITE_TO_STREAM_TIMEOUT_MS = 30000 ;
69
+
67
70
constructor (
68
71
@inject ( ILogger )
69
72
@named ( MonitorServiceName )
@@ -134,15 +137,6 @@ export class MonitorService extends CoreClientAware implements Disposable {
134
137
return ! ! this . duplex ;
135
138
}
136
139
137
- setDuplexHandlers (
138
- duplex : ClientDuplexStream < MonitorRequest , MonitorResponse > ,
139
- handlers : DuplexHandler [ ]
140
- ) : void {
141
- for ( const handler of handlers ) {
142
- duplex . on ( handler . key , handler . callback ) ;
143
- }
144
- }
145
-
146
140
/**
147
141
* Start and connects a monitor using currently set board and port.
148
142
* If a monitor is already started or board fqbn, port address and/or protocol
@@ -199,16 +193,16 @@ export class MonitorService extends CoreClientAware implements Disposable {
199
193
const coreClient = await this . coreClient ( ) ;
200
194
201
195
const { instance } = coreClient ;
202
- const req = new MonitorRequest ( ) ;
203
- req . setInstance ( instance ) ;
196
+ const monitorRequest = new MonitorRequest ( ) ;
197
+ monitorRequest . setInstance ( instance ) ;
204
198
if ( this . board ?. fqbn ) {
205
- req . setFqbn ( this . board . fqbn ) ;
199
+ monitorRequest . setFqbn ( this . board . fqbn ) ;
206
200
}
207
201
if ( this . port ?. address && this . port ?. protocol ) {
208
202
const port = new gRPCPort ( ) ;
209
203
port . setAddress ( this . port . address ) ;
210
204
port . setProtocol ( this . port . protocol ) ;
211
- req . setPort ( port ) ;
205
+ monitorRequest . setPort ( port ) ;
212
206
}
213
207
const config = new MonitorPortConfiguration ( ) ;
214
208
for ( const id in this . settings . pluggableMonitorSettings ) {
@@ -217,81 +211,11 @@ export class MonitorService extends CoreClientAware implements Disposable {
217
211
s . setValue ( this . settings . pluggableMonitorSettings [ id ] . selectedValue ) ;
218
212
config . addSettings ( s ) ;
219
213
}
220
- req . setPortConfiguration ( config ) ;
221
-
222
- // Promise executor
223
- const writeToStream = ( resolve : ( value : boolean ) => void ) => {
224
- this . duplex = coreClient . client . monitor ( ) ;
225
-
226
- const duplexHandlers : DuplexHandler [ ] = [
227
- {
228
- key : 'close' ,
229
- callback : ( ) => {
230
- this . duplex = null ;
231
- this . updateClientsSettings ( {
232
- monitorUISettings : { connected : false } ,
233
- } ) ;
234
- this . logger . info (
235
- `monitor to ${ this . port ?. address } using ${ this . port ?. protocol } closed by client`
236
- ) ;
237
- } ,
238
- } ,
239
- {
240
- key : 'end' ,
241
- callback : ( ) => {
242
- this . duplex = null ;
243
- this . updateClientsSettings ( {
244
- monitorUISettings : { connected : false } ,
245
- } ) ;
246
- this . logger . info (
247
- `monitor to ${ this . port ?. address } using ${ this . port ?. protocol } closed by server`
248
- ) ;
249
- } ,
250
- } ,
251
- {
252
- key : 'error' ,
253
- callback : ( err : Error ) => {
254
- this . logger . error ( err ) ;
255
- resolve ( false ) ;
256
- // TODO
257
- // this.theiaFEClient?.notifyError()
258
- } ,
259
- } ,
260
- {
261
- key : 'data' ,
262
- callback : ( res : MonitorResponse ) => {
263
- if ( res . getError ( ) ) {
264
- // TODO: Maybe disconnect
265
- this . logger . error ( res . getError ( ) ) ;
266
- return ;
267
- }
268
- if ( res . getSuccess ( ) ) {
269
- resolve ( true ) ;
270
- return ;
271
- }
272
- const data = res . getRxData ( ) ;
273
- const message =
274
- typeof data === 'string'
275
- ? data
276
- : new TextDecoder ( 'utf8' ) . decode ( data ) ;
277
- this . messages . push ( ...splitLines ( message ) ) ;
278
- } ,
279
- } ,
280
- ] ;
281
-
282
- this . setDuplexHandlers ( this . duplex , duplexHandlers ) ;
283
- this . duplex . write ( req ) ;
284
- } ;
285
-
286
- let attemptsRemaining = 10 ;
287
- let wroteToStreamSuccessfully = false ;
288
- while ( attemptsRemaining > 0 ) {
289
- wroteToStreamSuccessfully = await new Promise ( writeToStream ) ;
290
- if ( wroteToStreamSuccessfully ) break ;
291
- attemptsRemaining -= 1 ;
292
- await new Promise ( ( r ) => setTimeout ( r , 2000 ) ) ;
293
- }
214
+ monitorRequest . setPortConfiguration ( config ) ;
294
215
216
+ const wroteToStreamSuccessfully = await this . pollWriteToStream (
217
+ monitorRequest
218
+ ) ;
295
219
if ( wroteToStreamSuccessfully ) {
296
220
this . startMessagesHandlers ( ) ;
297
221
this . logger . info (
@@ -311,6 +235,144 @@ export class MonitorService extends CoreClientAware implements Disposable {
311
235
}
312
236
}
313
237
238
+ async createDuplex ( ) : Promise <
239
+ ClientDuplexStream < MonitorRequest , MonitorResponse >
240
+ > {
241
+ const coreClient = await this . coreClient ( ) ;
242
+ return coreClient . client . monitor ( ) ;
243
+ }
244
+
245
+ setDuplexHandlers (
246
+ duplex : ClientDuplexStream < MonitorRequest , MonitorResponse > ,
247
+ additionalHandlers : DuplexHandler [ ]
248
+ ) : void {
249
+ // default handlers
250
+ duplex
251
+ . on ( 'close' , ( ) => {
252
+ this . duplex = null ;
253
+ this . updateClientsSettings ( {
254
+ monitorUISettings : { connected : false } ,
255
+ } ) ;
256
+ this . logger . info (
257
+ `monitor to ${ this . port ?. address } using ${ this . port ?. protocol } closed by client`
258
+ ) ;
259
+ } )
260
+ . on ( 'end' , ( ) => {
261
+ this . duplex = null ;
262
+ this . updateClientsSettings ( {
263
+ monitorUISettings : { connected : false } ,
264
+ } ) ;
265
+ this . logger . info (
266
+ `monitor to ${ this . port ?. address } using ${ this . port ?. protocol } closed by server`
267
+ ) ;
268
+ } ) ;
269
+
270
+ for ( const handler of additionalHandlers ) {
271
+ duplex . on ( handler . key , handler . callback ) ;
272
+ }
273
+ }
274
+
275
+ pollWriteToStream ( request : MonitorRequest ) : Promise < boolean > {
276
+ let attemptsRemaining = this . MAX_WRITE_TO_STREAM_TRIES ;
277
+ const writeTimeoutMs = this . WRITE_TO_STREAM_TIMEOUT_MS ;
278
+
279
+ const createWriteToStreamExecutor =
280
+ ( duplex : ClientDuplexStream < MonitorRequest , MonitorResponse > ) =>
281
+ ( resolve : ( value : boolean ) => void , reject : ( ) => void ) => {
282
+ const resolvingDuplexHandlers : DuplexHandler [ ] = [
283
+ {
284
+ key : 'error' ,
285
+ callback : async ( err : Error ) => {
286
+ this . logger . error ( err ) ;
287
+ resolve ( false ) ;
288
+ // TODO
289
+ // this.theiaFEClient?.notifyError()
290
+ } ,
291
+ } ,
292
+ {
293
+ key : 'data' ,
294
+ callback : async ( monitorResponse : MonitorResponse ) => {
295
+ if ( monitorResponse . getError ( ) ) {
296
+ // TODO: Maybe disconnect
297
+ this . logger . error ( monitorResponse . getError ( ) ) ;
298
+ return ;
299
+ }
300
+ if ( monitorResponse . getSuccess ( ) ) {
301
+ resolve ( true ) ;
302
+ return ;
303
+ }
304
+ const data = monitorResponse . getRxData ( ) ;
305
+ const message =
306
+ typeof data === 'string'
307
+ ? data
308
+ : new TextDecoder ( 'utf8' ) . decode ( data ) ;
309
+ this . messages . push ( ...splitLines ( message ) ) ;
310
+ } ,
311
+ } ,
312
+ ] ;
313
+
314
+ this . setDuplexHandlers ( duplex , resolvingDuplexHandlers ) ;
315
+
316
+ setTimeout ( ( ) => {
317
+ reject ( ) ;
318
+ } , writeTimeoutMs ) ;
319
+ duplex . write ( request ) ;
320
+ } ;
321
+
322
+ const pollWriteToStream = new Promise < boolean > ( ( resolve ) => {
323
+ const startPolling = async ( ) => {
324
+ // here we create a new duplex but we don't yet
325
+ // set "this.duplex", nor do we use "this.duplex" in our poll
326
+ // as duplex 'end' / 'close' events (which we do not "await")
327
+ // will set "this.duplex" to null
328
+ const createdDuplex = await this . createDuplex ( ) ;
329
+
330
+ let pollingIsSuccessful ;
331
+ // attempt a "writeToStream" and "await" CLI response: success (true) or error (false)
332
+ // if we get neither within WRITE_TO_STREAM_TIMEOUT_MS or an error we get undefined
333
+ try {
334
+ const writeToStream = createWriteToStreamExecutor ( createdDuplex ) ;
335
+ pollingIsSuccessful = await new Promise ( writeToStream ) ;
336
+ } catch ( error ) {
337
+ this . logger . error ( error ) ;
338
+ }
339
+
340
+ // CLI confirmed port opened successfully
341
+ if ( pollingIsSuccessful ) {
342
+ this . duplex = createdDuplex ;
343
+ resolve ( true ) ;
344
+ return ;
345
+ }
346
+
347
+ // if "pollingIsSuccessful" is false
348
+ // the CLI gave us an error, lets try again
349
+ // after waiting 2 seconds if we've not already
350
+ // reached MAX_WRITE_TO_STREAM_TRIES
351
+ if ( pollingIsSuccessful === false ) {
352
+ attemptsRemaining -= 1 ;
353
+ if ( attemptsRemaining > 0 ) {
354
+ setTimeout ( startPolling , 2000 ) ;
355
+ return ;
356
+ } else {
357
+ resolve ( false ) ;
358
+ return ;
359
+ }
360
+ }
361
+
362
+ // "pollingIsSuccessful" remains undefined:
363
+ // we got no response from the CLI within 30 seconds
364
+ // resolve to false and end the duplex connection
365
+ resolve ( false ) ;
366
+ createdDuplex . end ( ) ;
367
+ return ;
368
+ } ;
369
+
370
+ startPolling ( ) ;
371
+ } ) ;
372
+
373
+ return pollWriteToStream ;
374
+ }
375
+
314
376
/**
315
377
* Pauses the currently running monitor, it still closes the gRPC connection
316
378
* with the underlying monitor process but it doesn't stop the message handlers
0 commit comments