From dbd4086bf96ba85833302ce1c21a2d14cdc5e901 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Sat, 12 Sep 2026 23:28:15 +0100 Subject: [PATCH] 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 | 31 +++++++++++++++++++ providers/beamng/beamng.go | 7 +++++ providers/beamng/utils.go | 6 ---- providers/iracing/iracing.go | 18 +++++++++++ providers/iracing/utils.go | 4 --- services/telemetry.go | 60 +++++++++++++++++++++++++++++++----- telemetry/provider.go | 5 ++- tui/tui.go | 2 +- 8 files changed, 114 insertions(+), 19 deletions(-) create mode 100644 Docs/Provider_Discovery.md diff --git a/Docs/Provider_Discovery.md b/Docs/Provider_Discovery.md new file mode 100644 index 0000000..d107f7c --- /dev/null +++ b/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 diff --git a/providers/beamng/beamng.go b/providers/beamng/beamng.go index d653073..3a7b769 100644 --- a/providers/beamng/beamng.go +++ b/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 diff --git a/providers/beamng/utils.go b/providers/beamng/utils.go index 98eb459..2854b74 100644 --- a/providers/beamng/utils.go +++ b/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 -} diff --git a/providers/iracing/iracing.go b/providers/iracing/iracing.go index 268cd6a..333364a 100644 --- a/providers/iracing/iracing.go +++ b/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 +} diff --git a/providers/iracing/utils.go b/providers/iracing/utils.go index c8b0de8..b524a80 100644 --- a/providers/iracing/utils.go +++ b/providers/iracing/utils.go @@ -31,7 +31,3 @@ func IsRunning() bool { return true } - -func (i *IRacing) Stalled() bool { - return i.Stalled() -} diff --git a/services/telemetry.go b/services/telemetry.go index f9f8566..c6c44c7 100644 --- a/services/telemetry.go +++ b/services/telemetry.go @@ -25,8 +25,10 @@ type TelemetryService struct { // Output window Messages chan string // Cancel looking for providers - CtxMonitor context.Context - cancelMonitor context.CancelFunc + CtxMonitor context.Context + 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 diff --git a/telemetry/provider.go b/telemetry/provider.go index ff3faca..e2cfeb5 100644 --- a/telemetry/provider.go +++ b/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() } diff --git a/tui/tui.go b/tui/tui.go index 68feb50..374d21e 100644 --- a/tui/tui.go +++ b/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,