@@ -95,7 +95,6 @@ func New(id string, args ...string) (*PluggableDiscovery, error) {
95
95
process : proc ,
96
96
incomingMessagesChan : messageChan ,
97
97
outgoingCommandsPipe : stdin ,
98
- alive : true ,
99
98
}
100
99
go disc .jsonDecodeLoop (stdout , messageChan )
101
100
return disc , nil
@@ -118,6 +117,7 @@ func (disc *PluggableDiscovery) jsonDecodeLoop(in io.Reader, outChan chan<- *dis
118
117
disc .incomingMessagesError = err
119
118
disc .statusMutex .Unlock ()
120
119
close (outChan )
120
+ // TODO: Try restarting process some times before closing it completely
121
121
}
122
122
123
123
for {
@@ -202,6 +202,9 @@ func (disc *PluggableDiscovery) runProcess() error {
202
202
if err := disc .process .Start (); err != nil {
203
203
return err
204
204
}
205
+ disc .statusMutex .Lock ()
206
+ defer disc .statusMutex .Unlock ()
207
+ disc .alive = true
205
208
return nil
206
209
}
207
210
@@ -257,6 +260,13 @@ func (disc *PluggableDiscovery) Stop() error {
257
260
} else if msg .Message != "OK" || msg .Error {
258
261
return errors .Errorf ("command failed: %s" , msg .Message )
259
262
}
263
+ disc .statusMutex .Lock ()
264
+ defer disc .statusMutex .Unlock ()
265
+ if disc .eventChan != nil {
266
+ close (disc .eventChan )
267
+ disc .eventChan = nil
268
+ }
269
+ disc .eventsMode = false
260
270
return nil
261
271
}
262
272
@@ -272,6 +282,13 @@ func (disc *PluggableDiscovery) Quit() error {
272
282
} else if msg .Message != "OK" || msg .Error {
273
283
return errors .Errorf ("command failed: %s" , msg .Message )
274
284
}
285
+ disc .statusMutex .Lock ()
286
+ defer disc .statusMutex .Unlock ()
287
+ if disc .eventChan != nil {
288
+ close (disc .eventChan )
289
+ disc .eventChan = nil
290
+ }
291
+ disc .alive = false
275
292
return nil
276
293
}
277
294
@@ -296,9 +313,14 @@ func (disc *PluggableDiscovery) List() ([]*Port, error) {
296
313
// The event channel must be consumed as quickly as possible since it may block the
297
314
// discovery if it becomes full. The channel size is configurable.
298
315
func (disc * PluggableDiscovery ) EventChannel (size int ) <- chan * Event {
299
- c := make (chan * Event , size )
300
316
disc .statusMutex .Lock ()
301
317
defer disc .statusMutex .Unlock ()
318
+ if disc .eventChan != nil {
319
+ // In case there is already an existing event channel in use we close it
320
+ // before creating a new one.
321
+ close (disc .eventChan )
322
+ }
323
+ c := make (chan * Event , size )
302
324
disc .eventChan = c
303
325
return c
304
326
}
0 commit comments