partial subscriptions

not really partial. they just happen per peripheral now

constants/constants.go

@@ -1,6 +1,13 @@
 package constants
 
+// Provider Names
 const (
 	IRacingProviderName = "iRacing"
 	BeamNGProviderName  = "BeamNG.drive"
 )
+
+// Virtual Field Names
+const (
+	FuelCalculatorName = "FuelCalculator"
+	RPMLightsName      = "RPMLights"
+)

providers/beamng/beamng.go

@@ -28,6 +28,9 @@ type BeamNG struct {
 	data     *telemetry.TelemetryData
 	updaters [telemetry.MaxFields]func(*telemetry.TelemetryField)
 
+	// Field Subscription management
+	boundFields map[telemetry.FieldID]bool
+
 	// stream control
 	wg           sync.WaitGroup
 	streamCancel context.CancelFunc
@@ -51,7 +54,9 @@ func NewBeamNGProvider(logger *slog.Logger, opts *bngsdk.Options) (*BeamNG, erro
 		data:   telemetry.NewTelemetryData(),
 		SDK:    beam,
 		og:     &bngsdk.Outgauge{},
-		ticker: time.NewTicker(time.Second / 60),
+		// Field subscription management
+		boundFields: make(map[telemetry.FieldID]bool, telemetry.MaxFields),
+		ticker:      time.NewTicker(time.Second / 60),
 	}
 
 	provider.updaters = [telemetry.MaxFields]func(*telemetry.TelemetryField){
@@ -115,47 +120,68 @@ func (b *BeamNG) Stream() (<-chan telemetry.TelemetryData, error) {
 	return ch, nil
 }
 
-func (b *BeamNG) Subscribe(requestFields []telemetry.FieldID) {
-	// NOTE: document how the Subscribe funtion works
-	slog.Debug(fmt.Sprintf("Len Req: %d\n", len(requestFields)))
+// TODO: this function is exactly the same in BeamNG drive now, and I reckon it will be the same
+// In plenty other things. I should make it TelemetryData method
+func (i *BeamNG) subscribe(fields []telemetry.FieldID) []string {
+	newSubscriptions := make([]string, 0, telemetry.MaxFields)
 
-	b.data.ActiveBinds = make([]telemetry.BoundField, 0, len(requestFields))
+	i.mut.Lock()
+	defer i.mut.Unlock()
 
-	// First we must add the virtual fields
-	// we will add their dependencies and the primitives to a slice
-	pendingBinds := make([]telemetry.FieldID, telemetry.MaxFields)
-
-	for _, id := range requestFields {
-		switch id {
-		case telemetry.RPMStateColour:
-			b.data.VirtualBinds = append(b.data.VirtualBinds, telemetry.NewRPMLights())
-		case telemetry.FCCurrentLap:
-			b.data.VirtualBinds = append(b.data.VirtualBinds,
-				telemetry.NewFuelCalculator(slog.Default().WithGroup("FUEL CALC")))
-		default:
-			// primitive telemetry field
-			pendingBinds = append(pendingBinds, id)
-		}
-	}
-
-	boundCheck := make(map[telemetry.FieldID]bool)
-
-	// Now that we know all the fields we need to bind we follow the binding procedure
-	for _, id := range pendingBinds {
-		// Check if we already bound this FieldID
-		if boundCheck[id] {
+	for _, id := range fields {
+		if i.boundFields[id] {
 			continue
 		}
 
-		binding := telemetry.BoundField{
+		i.data.ActiveBinds[id] = telemetry.BoundField{
 			ID: id,
 		}
-
-		b.data.ActiveBinds = append(b.data.ActiveBinds, binding)
-		boundCheck[id] = true
+		i.boundFields[id] = true
+		newSubscriptions = append(newSubscriptions, telemetry.FieldNames[id])
 	}
 
-	slog.Debug(fmt.Sprintf("Subscribed: %+v\n", b.data.ActiveBinds))
+	// unsubscribe from fields we many not need anymore
+	for key, bound := range i.boundFields {
+		if _, ok := i.data.ActiveBinds[key]; bound && !ok {
+			delete(i.data.ActiveBinds, key)
+			i.boundFields[key] = false
+		}
+	}
+
+	return newSubscriptions
+}
+
+func (b *BeamNG) Subscribe(requestFields []telemetry.FieldID) []string {
+	// NOTE: document how the Subscribe funtion works
+	slog.Debug(fmt.Sprintf("Len Req: %d\n", len(requestFields)))
+
+	// First we must add the virtual fields
+	// we will add their dependencies and the primitives to a slice
+	toBind := make([]telemetry.FieldID, telemetry.MaxFields)
+
+	b.mut.Lock()
+	for _, id := range requestFields {
+		switch id {
+		case telemetry.RPMStateColour:
+			rpmLights := telemetry.NewRPMLights()
+			b.data.VirtualBinds[rpmLights.Name()] = rpmLights
+			toBind = append(toBind, rpmLights.EnsureSubscribed()...)
+		case telemetry.FCCurrentLap:
+			fuelCalc := telemetry.NewFuelCalculator(b.logger.WithGroup("FUEL CALC"))
+			b.data.VirtualBinds[fuelCalc.Name()] = fuelCalc
+			toBind = append(toBind, fuelCalc.EnsureSubscribed()...)
+		default:
+			// primitive telemetry field
+			toBind = append(toBind, id)
+		}
+	}
+	b.mut.Unlock()
+
+	newSubs := b.subscribe(toBind)
+
+	slog.Debug(fmt.Sprintf("Subscribed: %+v\n", toBind))
+
+	return newSubs
 }
 
 // Internal

providers/iracing/iracing.go

@@ -29,13 +29,14 @@ type IRacing struct {
 	mut      sync.Mutex
 	data     *telemetry.TelemetryData
 	updaters [telemetry.MaxFields]func(*telemetry.TelemetryField)
+	// Field Subscription management
+	boundFields map[telemetry.FieldID]bool
 
 	// Timing information
 	ticker *time.Ticker // ticker will keep polling intervals constant
 
 	// Stream
-	wg sync.WaitGroup
-	// streamCh     chan telemetry.TelemetryData
+	wg           sync.WaitGroup
 	streamCancel context.CancelFunc
 }
 
@@ -55,6 +56,8 @@ func NewIRacingProvider(
 		logger: logger,
 		SDK:    sdk,
 		data:   telemetry.NewTelemetryData(),
+		// Field subscription management
+		boundFields: make(map[telemetry.FieldID]bool, telemetry.MaxFields),
 		// TODO: make this configurable from the user side
 		ticker: time.NewTicker(time.Second / 60),
 	}
@@ -225,46 +228,67 @@ func (i *IRacing) StopStream() {
 	i.streamCancel = nil
 }
 
-func (i *IRacing) Subscribe(requestFields []telemetry.FieldID) {
-	i.logger.Debug(fmt.Sprintf("Len Req: %d\n", len(requestFields)))
+// TODO: this function is exactly the same in BeamNG drive now, and I reckon it will be the same
+// In plenty other things. I should make it TelemetryData method
+func (i *IRacing) subscribe(fields []telemetry.FieldID) []string {
+	newSubscriptions := make([]string, 0, telemetry.MaxFields)
 
-	i.data.ActiveBinds = make([]telemetry.BoundField, 0, len(requestFields))
+	i.mut.Lock()
+	defer i.mut.Unlock()
 
-	// First we must add the virtual fields
-	// we will add their dependencies and the primitives to a slice
-	pendingBinds := make([]telemetry.FieldID, 0, telemetry.MaxFields)
-
-	for _, id := range requestFields {
-		switch id {
-		case telemetry.RPMStateColour:
-			i.data.VirtualBinds = append(i.data.VirtualBinds, telemetry.NewRPMLights())
-		case telemetry.FCCurrentLap:
-			i.data.VirtualBinds = append(i.data.VirtualBinds,
-				telemetry.NewFuelCalculator(i.logger.WithGroup("FUEL CALC")))
-		default:
-			// primitive telemetry field
-			pendingBinds = append(pendingBinds, id)
-		}
-	}
-
-	boundCheck := make(map[telemetry.FieldID]bool)
-
-	// Now that we know all the fields we need to bind we follow the binding procedure
-	for _, id := range pendingBinds {
-		// Check if we already bound this FieldID
-		if boundCheck[id] {
+	for _, id := range fields {
+		if i.boundFields[id] {
 			continue
 		}
 
-		binding := telemetry.BoundField{
+		i.data.ActiveBinds[id] = telemetry.BoundField{
 			ID: id,
 		}
-
-		i.data.ActiveBinds = append(i.data.ActiveBinds, binding)
-		boundCheck[id] = true
+		i.boundFields[id] = true
+		newSubscriptions = append(newSubscriptions, telemetry.FieldNames[id])
 	}
 
+	// unsubscribe from fields we many not need anymore
+	for key, bound := range i.boundFields {
+		if _, ok := i.data.ActiveBinds[key]; bound && !ok {
+			delete(i.data.ActiveBinds, key)
+			i.boundFields[key] = false
+		}
+	}
+
+	return newSubscriptions
+}
+
+func (i *IRacing) Subscribe(requestFields []telemetry.FieldID) []string {
+	i.logger.Debug(fmt.Sprintf("Len Req: %d\n", len(requestFields)))
+
+	// First we must add the virtual fields
+	// we will add their dependencies and the primitives to a slice
+	toBind := make([]telemetry.FieldID, 0, telemetry.MaxFields)
+
+	i.mut.Lock()
+	for _, id := range requestFields {
+		switch id {
+		case telemetry.RPMStateColour:
+			rpmLights := telemetry.NewRPMLights()
+			i.data.VirtualBinds[rpmLights.Name()] = rpmLights
+			toBind = append(toBind, rpmLights.EnsureSubscribed()...)
+		case telemetry.FCCurrentLap:
+			fuelCalc := telemetry.NewFuelCalculator(i.logger.WithGroup("FUEL CALC"))
+			i.data.VirtualBinds[fuelCalc.Name()] = fuelCalc
+			toBind = append(toBind, fuelCalc.EnsureSubscribed()...)
+		default:
+			// primitive telemetry field
+			toBind = append(toBind, id)
+		}
+	}
+	i.mut.Unlock()
+
+	newSubs := i.subscribe(toBind)
+
 	i.logger.Debug(fmt.Sprintf("Subscribed: %+v\n", i.data.ActiveBinds))
+
+	return newSubs
 }
 
 func (i *IRacing) Name() string {

services/devices.go

@@ -2,6 +2,7 @@ package services
 
 import (
 	"context"
+	"fmt"
 	"log/slog"
 	"sync/atomic"
 
@@ -24,7 +25,8 @@ type DeviceService struct {
 	Messages chan string
 	// Callbacks
 	// Telemetry service data fetchers
-	telemetryProvider func() (string, error)
+	telemetryProvider        func() (string, error)
+	triggerFieldSubscription func([]telemetry.FieldID) []string
 }
 
 func NewDeviceService(logger *slog.Logger, msg chan string) *DeviceService {
@@ -43,6 +45,7 @@ func NewDeviceService(logger *slog.Logger, msg chan string) *DeviceService {
 
 	// Set the callbacks for PSS
 	dev.PSS.telemetryProvider = dev.getTelemetryProvider
+	dev.PSS.onPeripheralConfigured = dev.peripheralConfigured
 
 	return dev
 }
@@ -59,7 +62,7 @@ func (ds *DeviceService) GetDevices() []peripheral.Peripheral {
 	peripherals := make([]peripheral.Peripheral, 0, len(snapshot))
 
 	for _, state := range snapshot {
-		if state.State < DeviceIsConnected {
+		if state.State < DeviceIsConfigured {
 			continue
 		}
 		peripherals = append(peripherals, state.Peripheral)
@@ -95,6 +98,8 @@ func (ds *DeviceService) StopStream() {
 		return
 	}
 
+	ds.PSS.OnStopStream()
+
 	ds.streamCancel()
 	ds.streamCancel = nil
 }
@@ -145,5 +150,10 @@ func (ds *DeviceService) transmit(ctx context.Context) {
 }
 
 // Callbacks [START] -----------------------------------------------------------
+func (ds *DeviceService) peripheralConfigured(pname string, fields []telemetry.FieldID) {
+	// We need to retrigger field subscription here
+	subscribedTo := ds.triggerFieldSubscription(fields)
+	ds.Messages <- fmt.Sprintf("subscribed to fields: %q\n", subscribedTo)
+}
 
 // Callbacks [END] -------------------------------------------------------------

services/devices_lookup.go

@@ -10,6 +10,7 @@ import (
 
 	"esdi/devices"
 	"esdi/peripheral"
+	"esdi/telemetry"
 )
 
 var (
@@ -85,7 +86,8 @@ type PeripheralStateStore struct {
 	// Messaging for UI and stuff
 	Messages chan string
 	// Callbacks
-	telemetryProvider func() (string, error)
+	telemetryProvider      func() (string, error)
+	onPeripheralConfigured func(string, []telemetry.FieldID)
 	// Internal State
 	isStreaming bool
 }
@@ -183,6 +185,15 @@ func (pss *PeripheralStateStore) GetStreamingState() bool {
 	return pss.isStreaming
 }
 
+func (pss *PeripheralStateStore) GetPeripheralFields(pname string) []telemetry.FieldID {
+	per, err := pss.GetState(pname)
+	if err != nil {
+		return nil
+	}
+
+	return per.Peripheral.RequiredFields()
+}
+
 // "Events" [START] ------------------------------------------------------------
 func (ds *DeviceService) onDeviceTimedOut(pname string) {
 	ds.PSS.setDeviceTimedOut(pname)
@@ -204,6 +215,12 @@ func (pss *PeripheralStateStore) OnStopStream() {
 	pss.mu.Lock()
 	pss.isStreaming = false
 	pss.mu.Unlock()
+
+	for _, state := range pss.GetStates() {
+		if state.State == DeviceIsStreaming {
+			pss.setDeviceConfigured(state.device.Name)
+		}
+	}
 }
 
 // "Events" [END] --------------------------------------------------------------
@@ -270,8 +287,10 @@ func (pss *PeripheralStateStore) setDeviceConfigured(pname string) {
 	pss.Logger.Info("device is configured and ready for data", "device", pname)
 
 	pss.mu.Lock()
-	defer pss.mu.Unlock()
 	pss.store[pname].State = DeviceIsConfigured
+	pss.mu.Unlock()
+
+	pss.onPeripheralConfigured(pname, pss.GetPeripheralFields(pname))
 }
 
 func (pss *PeripheralStateStore) setDeviceIsStreaming(pname string) {
@@ -338,7 +357,6 @@ func (pss *PeripheralStateStore) configurePeripheral(
 		}
 
 		onSuccess(pname)
-		pss.setDeviceConfigured(pname)
 	}()
 }
 

services/services.go

@@ -26,9 +26,10 @@ func NewOrchestrator(logger *slog.Logger) (*Orchestrator, error) {
 
 	// Setup device service callbacks
 	devService.telemetryProvider = telemService.GetTelemetryProviderName
+	devService.triggerFieldSubscription = telemService.SubscribeToFields
 
 	// Setup telemetry service callbacks
-	telemService.peripheralProvider = devService.GetDevices
+	// telemService.getRequiredFields = devService.GetRequiredFields
 
 	return &Orchestrator{
 		DeviceService:    devService,

services/telemetry.go

@@ -7,7 +7,6 @@ import (
 	"sync"
 	"time"
 
-	"esdi/peripheral"
 	"esdi/providers"
 	"esdi/telemetry"
 	telem "esdi/telemetry"
@@ -37,7 +36,8 @@ type TelemetryService struct {
 	healthCheckCancel context.CancelFunc
 	// Callbacks
 	// Devices data request
-	peripheralProvider func() []peripheral.Peripheral
+	// peripheralProvider func() []peripheral.Peripheral
+	// getRequiredFields func() []telemetry.FieldID
 }
 
 func NewTelemetryService(
@@ -114,23 +114,11 @@ func (t *TelemetryService) UnsubscribeListener(id string) {
 	}
 }
 
-func (t *TelemetryService) SubscribeToFields() []telem.FieldID {
-	seen := make(map[telemetry.FieldID]struct{})
-	var allFields []telemetry.FieldID
+func (t *TelemetryService) SubscribeToFields(fields []telemetry.FieldID) []string {
+	t.logger.Debug("requested fields", "fields", fields)
+	subscribed := t.activeProvider.Subscribe(fields)
 
-	for _, dev := range t.peripheralProvider() {
-		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
+	return subscribed
 }
 
 // Listener Control [END] ------------------------------------------------------

telemetry/data.go

@@ -23,13 +23,6 @@ type FieldMapper struct {
 	Transform func(any) uint64
 }
 
-// VirtualField will derive data from telemetry primitives
-// So fuel per lap predictions, compound gauge lights and so on
-type VirtualField interface {
-	Process(td *TelemetryData)
-	EnsureSubscribed() []FieldID
-}
-
 // NOTE: Update the iracing SDK to write data to the same map ALWAYS, then
 // I can bind that address and read directly from there on the transform
 
@@ -228,16 +221,18 @@ func GetFieldID(name string) (FieldID, bool) {
 }
 
 // TelemetryData is
-// I need to find a way of having the values be per window or some other
 type TelemetryData struct {
 	Values              [MaxFields]TelemetryField
-	ActiveBinds         []BoundField
-	VirtualBinds        []VirtualField
+	ActiveBinds         map[FieldID]BoundField
+	VirtualBinds        map[string]VirtualField
 	InitialTime         time.Time
 	PenultimateDataPoll time.Time
 	LastDataPoll        time.Time
 }
 
 func NewTelemetryData() *TelemetryData {
-	return &TelemetryData{}
+	return &TelemetryData{
+		ActiveBinds:  make(map[FieldID]BoundField, MaxFields),
+		VirtualBinds: make(map[string]VirtualField),
+	}
 }

telemetry/fuel_calculator.go

@@ -5,6 +5,8 @@ import (
 	"log/slog"
 	"strconv"
 	"sync"
+
+	"esdi/constants"
 )
 
 // NOTE: for managing fuel consumption and predictions we need to filter out
@@ -264,3 +266,7 @@ func (fc *FuelCalculator) resetHistory() {
 func (fc *FuelCalculator) EnsureSubscribed() []FieldID {
 	return []FieldID{FuelLevel, LapNumber}
 }
+
+func (fc *FuelCalculator) Name() string {
+	return constants.FuelCalculatorName
+}

telemetry/provider.go

@@ -6,7 +6,7 @@ import "time"
 type TelemetryProvider interface {
 	StopStream()
 	Stream() (<-chan TelemetryData, error)
-	Subscribe([]FieldID)
+	Subscribe([]FieldID) []string
 	IsAlive(time.Duration) bool
 	Name() string
 	Close()

telemetry/rpm_lights.go

@@ -1,5 +1,7 @@
 package telemetry
 
+import "esdi/constants"
+
 type RPMLights struct {
 	State string
 }
@@ -28,3 +30,7 @@ func (rl *RPMLights) Process(td *TelemetryData) {
 func (rl *RPMLights) EnsureSubscribed() []FieldID {
 	return []FieldID{RPM}
 }
+
+func (rl *RPMLights) Name() string {
+	return constants.RPMLightsName
+}

telemetry/virtual_field.go

@@ -0,0 +1,9 @@
+package telemetry
+
+// VirtualField will derive data from telemetry primitives
+// So fuel per lap predictions, compound gauge lights and so on
+type VirtualField interface {
+	Name() string
+	Process(td *TelemetryData)
+	EnsureSubscribed() []FieldID
+}

tui/internal/controllers/device.go

@@ -93,7 +93,7 @@ func (mc *DeviceController) AddDeviceAPIListItems() {
 				mc.StreamCtrl.StreamView.Flex,
 			)
 
-			mc.StreamCtrl.SetInternalState()
+			// mc.StreamCtrl.SetInternalState()
 
 			mc.App.SetFocus(mc.StreamCtrl.StreamView.Options.Form)
 		})

tui/internal/controllers/streaming.go

@@ -154,8 +154,8 @@ 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() {
-	fields := sc.TelemServ.SubscribeToFields()
-	sc.Messages <- fmt.Sprintf("Subscribed to fields: %+v\n", fields)
+	// fields := sc.TelemServ.SubscribeToAllFields()
+	// sc.Messages <- fmt.Sprintf("Subscribed to fields: %+v\n", fields)
 }
 
 func (sc *StreamingCtrl) listenToUIStream() {