applied the same start-stop scheme to the BeamNG provider

Reviewed-on: #11

Docs/Streaming_Flow.md

@@ -0,0 +1 @@
+# Streaming Flow

providers/beamng/beamng.go

@@ -28,7 +28,7 @@ type BeamNG struct {
 	updaters [telemetry.MaxFields]func(*telemetry.TelemetryField)
 
 	// stream control
-	streamCh     chan telemetry.TelemetryData
+	wg           sync.WaitGroup
 	streamCancel context.CancelFunc
 
 	// timing
@@ -47,7 +47,6 @@ func NewBeamNGProvider(logger *slog.Logger, opts *bngsdk.Options) (*BeamNG, erro
 
 	provider := &BeamNG{
 		logger: logger.With("TelemetryProvider", NAME),
-		streamCh: make(chan telemetry.TelemetryData, 1),
 		data:   telemetry.NewTelemetryData(),
 		SDK:    beam,
 		og:     &bngsdk.Outgauge{},
@@ -110,9 +109,9 @@ func (b *BeamNG) Stream() (<-chan telemetry.TelemetryData, error) {
 	ctx, b.streamCancel = context.WithCancel(context.Background())
 
 	// Start the stream
-	b.stream(ctx)
+	ch := b.stream(ctx)
 
-	return b.streamCh, nil
+	return ch, nil
 }
 
 func (b *BeamNG) Subscribe(requestFields []telemetry.FieldID) {
@@ -162,7 +161,6 @@ func (b *BeamNG) Subscribe(requestFields []telemetry.FieldID) {
 
 func (b *BeamNG) readData() {
 	slog.Debug("READING THIS DATA")
-	// BUG: getting stuck in here
 	ogSnapshot, err := b.SDK.Update()
 	slog.Debug("THE DATA WAS READ")
 	if err != nil {
@@ -193,10 +191,15 @@ func (b *BeamNG) readData() {
 	b.data.LastDataPoll = time.Now()
 }
 
-func (b *BeamNG) stream(ctx context.Context) {
+func (b *BeamNG) stream(ctx context.Context) <-chan telemetry.TelemetryData {
 	b.data.InitialTime = time.Now()
+	outCh := make(chan telemetry.TelemetryData)
+	b.wg.Add(1)
 
 	go func() {
+		defer b.wg.Done()
+		defer close(outCh)
+
 		for {
 			// Explicitly intercept cancellation
 			select {
@@ -205,8 +208,6 @@ func (b *BeamNG) stream(ctx context.Context) {
 			default:
 			}
 
-			// NOTE: add a method to check if there's data available, or make this happen
-
 			select {
 			case <-ctx.Done():
 				return
@@ -217,7 +218,7 @@ func (b *BeamNG) stream(ctx context.Context) {
 
 				// Publish data
 				select {
-				case b.streamCh <- *b.data:
+				case outCh <- *b.data:
 					slog.Debug("PUBLISHED DATA")
 				default:
 					// skip this data, don't allow publishers to lag behind
@@ -225,4 +226,6 @@ func (b *BeamNG) stream(ctx context.Context) {
 			}
 		}
 	}()
+
+	return outCh
 }

providers/iracing/iracing.go

@@ -33,7 +33,8 @@ type IRacing struct {
 	ticker *time.Ticker // ticker will keep polling intervals constant
 
 	// Stream
-	streamCh     chan telemetry.TelemetryData
+	wg sync.WaitGroup
+	// streamCh     chan telemetry.TelemetryData
 	streamCancel context.CancelFunc
 }
 
@@ -53,7 +54,7 @@ func NewIRacingProvider(
 		logger: logger,
 		SDK:    sdk,
 		data:   telemetry.NewTelemetryData(),
-		streamCh: make(chan telemetry.TelemetryData, 1),
+		// streamCh: make(chan telemetry.TelemetryData, 1),
 		// NOTE: This is because I stupidly recorded a test IBT file in 240
 		// TODO: make this configurable from the user side
 		ticker: time.NewTicker(time.Second / 240),
@@ -129,11 +130,14 @@ func (i *IRacing) isDataAvailable() bool {
 	return true
 }
 
-func (i *IRacing) stream(ctx context.Context) {
+func (i *IRacing) stream(ctx context.Context) <-chan telemetry.TelemetryData {
 	i.data.InitialTime = time.Now()
+	outCh := make(chan telemetry.TelemetryData)
+	i.wg.Add(1)
 
 	go func() {
-		defer close(i.streamCh)
+		defer i.wg.Done()
+		defer close(outCh)
 
 		// Put this into the configuration file
 		consecutiveTimeouts := 0
@@ -149,12 +153,11 @@ 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
 				select {
-				case i.streamCh <- *i.data:
+				case outCh <- *i.data:
 				default:
 					// skip this data, don't allow publishers to lag behind
 				}
@@ -170,6 +173,8 @@ func (i *IRacing) stream(ctx context.Context) {
 			}
 		}
 	}()
+
+	return outCh
 }
 
 func (i *IRacing) readData() {
@@ -206,9 +211,9 @@ func (i *IRacing) Stream() (<-chan telemetry.TelemetryData, error) {
 	ctx, i.streamCancel = context.WithCancel(context.Background())
 
 	// Start the stream
-	i.stream(ctx)
+	ch := i.stream(ctx)
 
-	return i.streamCh, nil
+	return ch, nil
 }
 
 func (i *IRacing) StopStream() {
@@ -217,6 +222,7 @@ func (i *IRacing) StopStream() {
 	}
 
 	i.streamCancel()
+	i.wg.Wait()
 	i.streamCancel = nil
 }
 

services/devices.go

@@ -84,14 +84,6 @@ func (ds *DeviceService) FindDevices() {
 	}
 }
 
-// func (ds *DeviceService) SubscribeFields() error {
-// 	for _, dev := range ds.Devices {
-// 		fields := dev.RequiredFields()
-// 	}
-//
-// 	return nil
-// }
-
 func (ds *DeviceService) RegisterDevice(dev peripheral.Peripheral) error {
 	ds.mu.Lock()
 	defer ds.mu.Unlock()

services/telemetry.go

@@ -16,6 +16,8 @@ import (
 type TelemetryService struct {
 	logger     *slog.Logger
 	devService *DeviceService
+	// Streaming
+	isStreaming bool
 	// Concurrency protection
 	mut            sync.RWMutex
 	activeProvider telem.TelemetryProvider
@@ -66,6 +68,92 @@ func (t *TelemetryService) ProviderMonitor(ctx context.Context) {
 	}
 }
 
+// Listener Control [START] ----------------------------------------------------
+
+func (t *TelemetryService) SubscribeListener(id string, bufferSize int) <-chan telem.TelemetryData {
+	t.mut.Lock()
+	defer t.mut.Unlock()
+
+	// NOTE: is this truly necessary?
+	// return the channel if it already exists
+	if ch, exists := t.listeners[id]; exists {
+		return ch
+	}
+
+	ch := make(chan telem.TelemetryData, bufferSize)
+	t.listeners[id] = ch
+
+	t.logger.Info("New stream subscriber registered", "id", id)
+	return ch
+}
+
+func (t *TelemetryService) UnsubscribeListener(id string) {
+	t.mut.Lock()
+	defer t.mut.Unlock()
+
+	if ch, exists := t.listeners[id]; exists {
+		close(ch)
+		delete(t.listeners, id)
+		t.logger.Info("Stream subscriber removed", "id", id)
+	}
+}
+
+func (t *TelemetryService) SubscribeToFields() []telem.FieldID {
+	seen := make(map[telemetry.FieldID]struct{})
+	var allFields []telemetry.FieldID
+
+	for _, dev := range t.devService.Devices {
+		for _, field := range dev.RequiredFields() {
+			if _, exists := seen[field]; !exists {
+				seen[field] = struct{}{}
+				allFields = append(allFields, field)
+			}
+		}
+	}
+
+	t.logger.Debug("requested fields", "fields", allFields)
+	t.activeProvider.Subscribe(allFields)
+
+	return allFields
+}
+
+// Listener Control [END] ------------------------------------------------------
+
+// Provider Control [START] ----------------------------------------------------
+
+func (t *TelemetryService) HasActiveProvider() bool {
+	if t.activeProvider == nil {
+		return false
+	}
+
+	return true
+}
+
+func (t *TelemetryService) dropActiveProvider() {
+	if t.cancelForward != nil {
+		t.cancelForward()
+	}
+
+	t.activeProvider.StopStream()
+	t.activeProvider.Close()
+	t.activeProvider = nil
+}
+
+func (t *TelemetryService) SwitchProvider(newProvider telem.TelemetryProvider) error {
+	t.mut.Lock()
+	defer t.mut.Unlock()
+
+	// Clean up the current to be old provider
+	if t.activeProvider != nil {
+		t.dropActiveProvider()
+	}
+
+	// Assign the new provider
+	t.activeProvider = newProvider
+
+	return nil
+}
+
 func (t *TelemetryService) onProviderHealthCheckFailed() {
 	// Just restart the whole lookup process
 	go t.FindProvider(t.CtxMonitor)
@@ -123,19 +211,46 @@ func (t *TelemetryService) FindProvider(ctx context.Context) {
 	}
 }
 
-func (t *TelemetryService) SwitchProvider(newProvider telem.TelemetryProvider) error {
+// Provider Control [END] ------------------------------------------------------
+
+// Streaming Control [START] ---------------------------------------------------
+
+func (t *TelemetryService) StopStream() {
 	t.mut.Lock()
 	defer t.mut.Unlock()
 
-	// Clean up the current to be old provider
+	if t.cancelForward != nil {
-	if t.activeProvider != nil {
+		t.cancelForward()
-		t.dropActiveProvider()
+		t.cancelForward = nil
 	}
 
-	// Assign the new provider
+	t.activeProvider.StopStream()
-	t.activeProvider = newProvider
+	t.isStreaming = false
+}
 
-	return nil
+func (t *TelemetryService) StartStream() {
+	slog.Debug("Stream started")
+
+	// Start the new stream
+	if t.activeProvider == nil {
+		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
+
+	// Create the context so we can control the lifecycle
+	ctx, cancel := context.WithCancel(context.Background())
+	t.cancelForward = cancel
+
+	// Multiplex this data
+	go t.multiplexData(ctx, simInCh)
+	t.isStreaming = true
 }
 
 func (t *TelemetryService) multiplexData(ctx context.Context, dataCh <-chan telem.TelemetryData) {
@@ -165,101 +280,8 @@ func (t *TelemetryService) multiplexData(ctx context.Context, dataCh <-chan tele
 	}
 }
 
-func (t *TelemetryService) dropActiveProvider() {
+func (t *TelemetryService) IsStreaming() bool {
-	if t.cancelForward != nil {
+	return t.isStreaming
-		t.cancelForward()
-	}
-
-	t.activeProvider.StopStream()
-	t.activeProvider.Close()
-	t.activeProvider = nil
 }
 
-func (t *TelemetryService) SubscribeListener(id string, bufferSize int) <-chan telem.TelemetryData {
+// Streaming Control [END] -----------------------------------------------------
-	t.mut.Lock()
-	defer t.mut.Unlock()
-
-	// NOTE: is this truly necessary?
-	// return the channel if it already exists
-	if ch, exists := t.listeners[id]; exists {
-		return ch
-	}
-
-	ch := make(chan telem.TelemetryData, bufferSize)
-	t.listeners[id] = ch
-
-	t.logger.Info("New stream subscriber registered", "id", id)
-	return ch
-}
-
-func (t *TelemetryService) UnsubscribeListener(id string) {
-	t.mut.Lock()
-	defer t.mut.Unlock()
-
-	if ch, exists := t.listeners[id]; exists {
-		close(ch)
-		delete(t.listeners, id)
-		t.logger.Info("Stream subscriber removed", "id", id)
-	}
-}
-
-func (t *TelemetryService) SubscribeToFields() {
-	seen := make(map[telemetry.FieldID]struct{})
-	var allFields []telemetry.FieldID
-
-	for _, dev := range t.devService.Devices {
-		for _, field := range dev.RequiredFields() {
-			if _, exists := seen[field]; !exists {
-				seen[field] = struct{}{}
-				allFields = append(allFields, field)
-			}
-		}
-	}
-
-	t.logger.Debug("requested fields", "fields", allFields)
-	t.activeProvider.Subscribe(allFields)
-}
-
-func (t *TelemetryService) StartStream() {
-	slog.Debug("Stream started")
-
-	// Start the new stream
-	if t.activeProvider == nil {
-		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
-
-	// Create the context so we can control the lifecycle
-	ctx, cancel := context.WithCancel(context.Background())
-	t.cancelForward = cancel
-
-	// Multiplex this data
-	go t.multiplexData(ctx, simInCh)
-}
-
-func (t *TelemetryService) StopStream() {
-	t.mut.Lock()
-	defer t.mut.Unlock()
-
-	if t.cancelForward != nil {
-		t.cancelForward()
-		t.cancelForward = nil
-	}
-
-	t.activeProvider.StopStream()
-}
-
-func (t *TelemetryService) HasActiveProvider() bool {
-	if t.activeProvider == nil {
-		return false
-	}
-
-	return true
-}

tui/internal/controllers/streaming.go

@@ -18,7 +18,7 @@ import (
 
 type StreamingCtrl struct {
 	*Controller
-	Service     *services.DeviceService
+	DevService  *services.DeviceService
 	StreamView  *views.StreamToolView
 	Messages    chan string
 	Internal    chan string
@@ -45,18 +45,16 @@ func NewStreamingCtrl(
 
 	ctrl := &StreamingCtrl{
 		Controller:  base,
-		Service:     devService,
+		DevService:  devService,
 		TelemServ:   serTelem,
 		Messages:    make(chan string, 10),
 		Internal:    make(chan string, 10),
 		TelemetryCh: make(chan telemetry.TelemetryData, 1),
 		Run:         false,
 		StreamView:  streamView,
-		isRunning:   false,
 	}
 
 	ctrl.registerHooks()
-	// ctrl.subscribeListeners()
 
 	return ctrl
 }
@@ -96,11 +94,11 @@ func (sc *StreamingCtrl) registerHooks() {
 }
 
 func (sc *StreamingCtrl) StartStop() {
-	if sc.isRunning {
+	if sc.TelemServ.IsStreaming() {
 		slog.Info("stopping stream")
 
 		sc.TelemServ.StopStream()
-		sc.Service.StopStream()
+		sc.DevService.StopStream()
 
 		sc.isRunning = false
 		return
@@ -110,9 +108,9 @@ func (sc *StreamingCtrl) StartStop() {
 	// NOTE:
 	// Subscribe the only existing device - needs to be discovered by now
 	slog.Debug("setting the data stream for device servie")
-	sc.Service.SetTelemetryChannel(sc.TelemServ.SubscribeListener("DeviceService", 1))
+	sc.DevService.SetTelemetryChannel(sc.TelemServ.SubscribeListener("DeviceService", 1))
 
-	dev, err := sc.Service.GetDevice(uidevice.NAME)
+	dev, err := sc.DevService.GetDevice(uidevice.NAME)
 	if err == nil {
 		if uiDev, ok := dev.(*uidevice.UIDevice); ok {
 			sc.TelemetryCh = uiDev.DataChannel()
@@ -121,7 +119,7 @@ func (sc *StreamingCtrl) StartStop() {
 	}
 
 	slog.Debug("starting services")
-	sc.Service.StartStream()
+	sc.DevService.StartStream()
 	sc.TelemServ.StartStream()
 
 	sc.isRunning = true
@@ -161,24 +159,11 @@ func (sc *StreamingCtrl) updateStream() {
 // Performance reasoning: this is not used during the high frequency data transmission
 // so we can get away with using a map for convenience here
 func (sc *StreamingCtrl) SetInternalState() {
-	// Acquire the cdashdisplay
+	fields := sc.TelemServ.SubscribeToFields()
-	// displayIF, err := sc.Service.GetDevice(cdashdisplay.NAME)
-	// if err != nil {
-	// 	sc.Messages <- "failed to get " + cdashdisplay.NAME
-	// 	return
-	// }
-	// display, ok := displayIF.(*cdashdisplay.CDashDisplay)
-	// if !ok {
-	// 	sc.Messages <- "failed to acquire " + cdashdisplay.NAME
-	// 	return
-	// }
-	// ---
-
-	sc.TelemServ.SubscribeToFields()
 
 	// sc.Messages <- fmt.Sprintf("Subscribed Fields: %+v [%d]\n", fields, len(fields))
 	// Should I update this?
-	sc.Messages <- fmt.Sprintf("Subscribed to fields\n")
+	sc.Messages <- fmt.Sprintf("Subscribed to fields: %+v", fields)
 }
 
 func (sc *StreamingCtrl) listenToUIStream() {
@@ -190,8 +175,6 @@ func (sc *StreamingCtrl) listenToUIStream() {
 		}
 		isDrawing.Store(true)
 
-		// sc.Logger.Debug("got data", "data", msg)
-
 		// Capture locally
 		telemetryMsg := msg