Provider loop discovery almost finished

Provider lookup will lookup on startup, do the healthcheck until the
stream starts and also restart on stream closure or provider stall

Docs/Provider_Discovery.md

@@ -0,0 +1,31 @@
+# Provider Discovery
+```
+NewControlPanel()
+       ↓
+go telemService.FindProvider() - has a callback for onFind
+       |
+       |iterates over the known providers
+       |
+     onFind
+       ↓
+telemService.SwitchProvider() - activates the found provider
+       |
+       |-→ Create a background job while the stream hasn't initiated to listened
+       |   for data, otherwise the connection might die before we start streaming
+       |                                 ↓
+       |                     go telemService.ProviderMonitor()
+       ↓                     |                              |           
+  /-→waits--\    if the provider stops we        this healthcheck is stopped 
+  \_________/    we clear the provider and       once the stream starts
+                 go back to the                   
+                             ↓
+                       FindProvider()
+```
+The provider discovery routine starts on `tui/tui.go`.
+`FindProvider` is called here and it starts a background job.
+On successful discovery the background job calls the `SwitchProvider` method
+and dies.
+
+## Provider stalls
+A provider stalls once there is no new data.
+After stalling

providers/beamng/beamng.go

@@ -79,6 +79,13 @@ func NewBeamNGProvider(ip string, port int) (*BeamNG, error) {
 	return provider, nil
 }
 
+func (b *BeamNG) Close() {
+}
+
+func (b *BeamNG) IsAlive(timeout time.Duration) bool {
+	return true
+}
+
 func (b *BeamNG) StopStream() {
 	if b.streamCancel == nil {
 		return

providers/beamng/utils.go

@@ -31,9 +31,3 @@ func IsRunning() bool {
 	// Think of a better number or something
 	return n >= 80
 }
-
-// Stalled
-// TODO: needs to be implemented
-func (i *BeamNG) Stalled() bool {
-	return false
-}

providers/iracing/iracing.go

@@ -108,6 +108,13 @@ func NewIRacingProvider(
 	return provider, nil
 }
 
+func (i *IRacing) Close() {
+	// Need to find a way of gracefully closing the channel
+	// close(i.streamCh)
+	i.SDK.Close()
+	i.ticker.Stop()
+}
+
 func (i *IRacing) isDataAvailable() bool {
 	// Its offline telemetry, data must be available
 	if i.SDK.File == nil {
@@ -126,6 +133,8 @@ func (i *IRacing) stream(ctx context.Context) {
 	i.data.InitialTime = time.Now()
 
 	go func() {
+		defer close(i.streamCh)
+
 		// Put this into the configuration file
 		consecutiveTimeouts := 0
 		maxTimeouts := 30
@@ -140,6 +149,7 @@ func (i *IRacing) stream(ctx context.Context) {
 
 			if i.SDK.CheckForDataEvent(time.Duration(dataEvTimeout) * time.Millisecond) {
 				consecutiveTimeouts = 0
+				i.logger.Debug("sending data", "timeouts", consecutiveTimeouts)
 				i.readData()
 
 				// Publish data
@@ -253,3 +263,11 @@ func (i *IRacing) Subscribe(requestFields map[int16]telemetry.FieldID) {
 
 	i.logger.Debug(fmt.Sprintf("Subscribed: %+v\n", i.data.ActiveBinds))
 }
+
+func (i *IRacing) IsAlive(timeout time.Duration) bool {
+	if !i.SDK.CheckForDataEvent(timeout) {
+		return false
+	}
+
+	return true
+}

providers/iracing/utils.go

@@ -31,7 +31,3 @@ func IsRunning() bool {
 
 	return true
 }
-
-func (i *IRacing) Stalled() bool {
-	return i.Stalled()
-}

services/telemetry.go

@@ -25,8 +25,10 @@ type TelemetryService struct {
 	// Output window
 	Messages chan string
 	// Cancel looking for providers
-	CtxMonitor    context.Context
+	CtxMonitor        context.Context
-	cancelMonitor context.CancelFunc
+	cancelMonitor     context.CancelFunc
+	CtxHealthcheck    context.Context
+	healthCheckCancel context.CancelFunc
 }
 
 func NewTelemetryService(logger *slog.Logger, devServo *DeviceService) *TelemetryService {
@@ -43,15 +45,51 @@ func NewTelemetryService(logger *slog.Logger, devServo *DeviceService) *Telemetr
 	return newService
 }
 
-func (t *TelemetryService) OnFindProvider(prov providers.Provider) {
+func (t *TelemetryService) ProviderMonitor(ctx context.Context) {
+	ticker := time.NewTicker(2 * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			slog.Debug("checking if provider is still running")
+			if !t.activeProvider.IsAlive(500 * time.Millisecond) {
+				slog.Debug("provider healthcheck failed")
+				t.dropActiveProvider()
+				t.onProviderHealthCheckFailed()
+				return
+			}
+		}
+	}
+}
+
+func (t *TelemetryService) onProviderHealthCheckFailed() {
+	// Just restart the whole lookup process
+	go t.FindProvider(t.CtxMonitor)
+}
+
+func (t *TelemetryService) onFindProvider(prov providers.Provider) {
+	// Attach to the provider
+	t.logger.Info("found provider for " + prov.Name)
 	t.SwitchProvider(prov.NewProvider(t.logger))
-	t.logger.Debug("Found provider for " + prov.Name)
+
+	// Create a routine to poll this provider while we wait to start the stream or pause it
+	t.CtxHealthcheck, t.healthCheckCancel = context.WithCancel(context.Background())
+	go t.ProviderMonitor(t.CtxHealthcheck)
+}
+
+func (t *TelemetryService) onProviderStopsMidStream() {
+	// clear the current provider
+	t.logger.Info("cleaning dropped provider and restarting lookup service")
+	t.dropActiveProvider()
+	go t.FindProvider(t.CtxMonitor)
 }
 
 // TODO: add some way of retriggering this. Currently it should:
 // start monitoring on startup -> find provider -> stop monitoring (when game closes for example)
-func (t *TelemetryService) FindProvider(ctx context.Context, callback func(providers.Provider),
+func (t *TelemetryService) FindProvider(ctx context.Context) {
-) {
 	ticker := time.NewTicker(2 * time.Second)
 	defer ticker.Stop()
 
@@ -64,7 +102,7 @@ func (t *TelemetryService) FindProvider(ctx context.Context, callback func(provi
 			for _, prov := range providers.Providers {
 				t.logger.Debug("checking provider: " + prov.Name)
 				if prov.IsRunning() {
-					callback(prov)
+					t.onFindProvider(prov)
 					return
 				}
 				t.logger.Debug("  wasn't read")
@@ -95,6 +133,8 @@ func (t *TelemetryService) multiplexData(ctx context.Context, dataCh <-chan tele
 			return
 		case data, ok := <-dataCh:
 			if !ok {
+				t.logger.Debug("something happened on the provider - stream closed")
+				t.onProviderStopsMidStream()
 				return
 			}
 
@@ -118,6 +158,8 @@ func (t *TelemetryService) dropActiveProvider() {
 	}
 
 	t.activeProvider.StopStream()
+	t.activeProvider.Close()
+	t.activeProvider = nil
 }
 
 func (t *TelemetryService) SubscribeListener(id string, bufferSize int) <-chan telem.TelemetryData {
@@ -160,6 +202,10 @@ func (t *TelemetryService) StartStream() {
 		slog.Debug("there's no active provider. not starting the stream")
 		return
 	}
+
+	// Stop the provider healthcheck
+	t.healthCheckCancel()
+
 	simInCh, _ := t.activeProvider.Stream()
 	// TODO: the provider needs to be able to tell the data has stopped
 	// so we can restart the provider lookup routine

telemetry/provider.go

@@ -1,9 +1,12 @@
 // Package telemetry is our interface with our data sources
 package telemetry
 
+import "time"
+
 type TelemetryProvider interface {
 	StopStream()
 	Stream() (<-chan TelemetryData, error)
 	Subscribe(map[int16]FieldID)
-	Stalled() bool // Return true if no fresh data is coming
+	IsAlive(time.Duration) bool
+	Close()
 }

tui/tui.go

@@ -31,7 +31,7 @@ func NewControlPanel(logger *slog.Logger) *ControlPanel {
 		panic("failed to create the telemetry service")
 	}
 
-	go telemService.FindProvider(telemService.CtxMonitor, telemService.OnFindProvider)
+	go telemService.FindProvider(telemService.CtxMonitor)
 
 	return &ControlPanel{
 		Controller:       baseController,