has way too many things
In short: added the ability to detect when a device has disconnected. From this we have to achieve a way of: - reconnecting the device - re-subscribing to the fields it needs (it should already be subscribed since the ESDI didn't stop tho - but for the sake of it, or in case a device joins later) - start sending data to it (which should be automatic given how we are handling the devices)
This commit is contained in:
+30
-72
@@ -2,32 +2,21 @@ package services
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"esdi/devices"
|
||||
"esdi/peripheral"
|
||||
"esdi/telemetry"
|
||||
)
|
||||
|
||||
var ErrPeripheralAlreadyRegistered = errors.New("peripheral is already registered")
|
||||
|
||||
// DeviceService will handle sending the data from the telemetry service to the
|
||||
// actual devices
|
||||
// NOTE: create a virtual device and make it be the output window or something so
|
||||
// we can just add it as a device or whatever instead of being a custom made thing
|
||||
// that would be pretty cool I think
|
||||
// DeviceService is the API for the peripherals
|
||||
type DeviceService struct {
|
||||
Logger *slog.Logger
|
||||
// Device discovery
|
||||
mu sync.RWMutex
|
||||
PSS *PeripheralStateStore // Store to track peripheral state
|
||||
ctxDiscovery context.Context
|
||||
ctxDiscoveryCancel context.CancelFunc
|
||||
Devices map[string]peripheral.Peripheral
|
||||
// Strem handling
|
||||
streamCancel context.CancelFunc
|
||||
TelemCh <-chan telemetry.TelemetryData
|
||||
@@ -39,7 +28,7 @@ func NewDeviceService(logger *slog.Logger) *DeviceService {
|
||||
sharedChannel := make(chan string, 10)
|
||||
|
||||
dev := &DeviceService{
|
||||
Devices: make(map[string]peripheral.Peripheral),
|
||||
PSS: NewPeripheralStateStore(logger.With("Service", "PeripheralStateStore"), devices.List),
|
||||
Logger: logger,
|
||||
Messages: sharedChannel,
|
||||
}
|
||||
@@ -52,68 +41,33 @@ func NewDeviceService(logger *slog.Logger) *DeviceService {
|
||||
return dev
|
||||
}
|
||||
|
||||
func (ds *DeviceService) FindDevices() {
|
||||
// Need to define a list of devices to search for
|
||||
// For now lets just try to find our cdashdisplay - will think about the rest later
|
||||
ticker := time.NewTicker(2 * time.Second)
|
||||
defer ticker.Stop()
|
||||
// Getters [START] -------------------------------------------------------------
|
||||
func (ds *DeviceService) GetDevices() []peripheral.Peripheral {
|
||||
snapshot := ds.PSS.GetStates()
|
||||
peripherals := make([]peripheral.Peripheral, 0, len(snapshot))
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ds.ctxDiscovery.Done():
|
||||
// If requested to cancel we cancel background discovery
|
||||
return
|
||||
case <-ticker.C:
|
||||
for pName, peripheral := range devices.List {
|
||||
if ds.DeviceExists(pName) {
|
||||
// We already discovered this device
|
||||
continue
|
||||
}
|
||||
|
||||
ds.Logger.Debug("looking for device", "name", pName)
|
||||
dev, err := peripheral.Discover()
|
||||
if err != nil {
|
||||
ds.Logger.Debug("didn't find device", "name", pName)
|
||||
continue
|
||||
}
|
||||
|
||||
// Register the device we just found
|
||||
ds.RegisterDevice(dev)
|
||||
}
|
||||
for _, state := range snapshot {
|
||||
if state.State != DeviceIsConnected {
|
||||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (ds *DeviceService) RegisterDevice(dev peripheral.Peripheral) error {
|
||||
ds.mu.Lock()
|
||||
defer ds.mu.Unlock()
|
||||
|
||||
if ds.DeviceExists(dev.Name()) {
|
||||
return ErrPeripheralAlreadyRegistered
|
||||
peripherals = append(peripherals, state.Peripheral)
|
||||
}
|
||||
|
||||
ds.Devices[dev.Name()] = dev
|
||||
|
||||
return nil
|
||||
return peripherals
|
||||
}
|
||||
|
||||
func (ds *DeviceService) GetDevice(name string) (peripheral.Peripheral, error) {
|
||||
val, ok := ds.Devices[name]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("device `%s` couldn't be found", name)
|
||||
}
|
||||
|
||||
return val, nil
|
||||
func (ds *DeviceService) GetPeripheral(pname string) (peripheral.Peripheral, error) {
|
||||
return ds.PSS.GetPeripheral(pname)
|
||||
}
|
||||
|
||||
func (ds *DeviceService) DeviceExists(name string) bool {
|
||||
if _, ok := ds.Devices[name]; !ok {
|
||||
return false
|
||||
}
|
||||
|
||||
return true
|
||||
func (ds *DeviceService) PeripheralExists(pname string) bool {
|
||||
_, err := ds.PSS.GetPeripheral(pname)
|
||||
return err == nil
|
||||
}
|
||||
|
||||
// Getters [END] ---------------------------------------------------------------
|
||||
|
||||
// Actions [START] -------------------------------------------------------------
|
||||
func (ds *DeviceService) StartStream() {
|
||||
// NOTE: i'm using this pattern a whole lot. Maybe I can create a struct to handle this
|
||||
var ctx context.Context
|
||||
@@ -136,6 +90,8 @@ func (ds *DeviceService) SetTelemetryChannel(ch <-chan telemetry.TelemetryData)
|
||||
ds.TelemCh = ch
|
||||
}
|
||||
|
||||
// Actions [END] ---------------------------------------------------------------
|
||||
|
||||
// transmit will send the data to the devices themselves
|
||||
func (ds *DeviceService) transmit(ctx context.Context) {
|
||||
var isSending atomic.Bool
|
||||
@@ -157,15 +113,17 @@ func (ds *DeviceService) transmit(ctx context.Context) {
|
||||
|
||||
// TODO: make a copy of the data and send that copy instead of keeping
|
||||
// the data locked
|
||||
ds.mu.RLock()
|
||||
for _, dev := range ds.Devices {
|
||||
err := dev.SendData(&data)
|
||||
for _, dev := range ds.PSS.GetStates() {
|
||||
if dev.State != DeviceIsConnected {
|
||||
continue
|
||||
}
|
||||
|
||||
err := dev.Peripheral.SendData(&data)
|
||||
|
||||
if err == peripheral.ErrDeviceTimedOut {
|
||||
// What do we do here?
|
||||
// TODO: somehow we need to handle reconnection
|
||||
ds.onDeviceTimedOut(dev.device.Name)
|
||||
}
|
||||
}
|
||||
ds.mu.RUnlock()
|
||||
|
||||
isSending.Store(false)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,222 @@
|
||||
package services
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"log/slog"
|
||||
"maps"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"esdi/devices"
|
||||
"esdi/peripheral"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrPeripheralAlreadyRegistered = errors.New("peripheral is already registered")
|
||||
ErrNoSuchDevice = errors.New("device doesn't exist")
|
||||
ErrDeviceIsNotConnected = errors.New("device isn't connected")
|
||||
)
|
||||
|
||||
type DeviceState = uint8
|
||||
|
||||
const (
|
||||
DeviceTimedOut uint8 = iota
|
||||
DeviceIsConnected
|
||||
DeviceIsDisconnected
|
||||
DeviceReconnected
|
||||
)
|
||||
|
||||
type PeripheralState struct {
|
||||
device *devices.Device
|
||||
Peripheral peripheral.Peripheral
|
||||
State DeviceState
|
||||
}
|
||||
|
||||
func NewPeripheralState(
|
||||
dev *devices.Device,
|
||||
peripheral peripheral.Peripheral,
|
||||
state DeviceState,
|
||||
) *PeripheralState {
|
||||
perState := PeripheralState{
|
||||
device: dev,
|
||||
Peripheral: peripheral,
|
||||
State: state,
|
||||
}
|
||||
|
||||
return &perState
|
||||
}
|
||||
|
||||
type PeripheralStateStore struct {
|
||||
Logger *slog.Logger
|
||||
mu sync.RWMutex
|
||||
store map[string]*PeripheralState
|
||||
}
|
||||
|
||||
func NewPeripheralStateStore(
|
||||
nLogger *slog.Logger,
|
||||
devList map[string]*devices.Device,
|
||||
) *PeripheralStateStore {
|
||||
store := PeripheralStateStore{
|
||||
Logger: nLogger,
|
||||
store: make(map[string]*PeripheralState),
|
||||
}
|
||||
|
||||
for _, dev := range devList {
|
||||
store.AddDevice(dev)
|
||||
}
|
||||
|
||||
return &store
|
||||
}
|
||||
|
||||
func (pss *PeripheralStateStore) GetStates() map[string]*PeripheralState {
|
||||
pss.mu.RLock()
|
||||
defer pss.mu.RUnlock()
|
||||
|
||||
return maps.Clone(pss.store)
|
||||
}
|
||||
|
||||
func (pss *PeripheralStateStore) GetState(pname string) (*PeripheralState, error) {
|
||||
if !pss.DeviceExists(pname) {
|
||||
return nil, ErrNoSuchDevice
|
||||
}
|
||||
|
||||
pss.mu.RLock()
|
||||
defer pss.mu.RUnlock()
|
||||
return pss.store[pname], nil
|
||||
}
|
||||
|
||||
func (pss *PeripheralStateStore) GetPeripheral(pname string) (peripheral.Peripheral, error) {
|
||||
state, err := pss.GetState(pname)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if state.State != DeviceIsConnected {
|
||||
return nil, ErrDeviceIsNotConnected
|
||||
}
|
||||
|
||||
return state.Peripheral, nil
|
||||
}
|
||||
|
||||
// AddDevice adds a new device for tracking
|
||||
func (pss *PeripheralStateStore) AddDevice(dev *devices.Device) error {
|
||||
if pss.DeviceExists(dev.Name) {
|
||||
return ErrPeripheralAlreadyRegistered
|
||||
}
|
||||
|
||||
pss.mu.Lock()
|
||||
defer pss.mu.Unlock()
|
||||
pss.store[dev.Name] = NewPeripheralState(dev, nil, DeviceIsDisconnected)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// DeviceExists returns whether the store is already tracking `pname`
|
||||
func (pss *PeripheralStateStore) DeviceExists(pname string) bool {
|
||||
pss.mu.RLock()
|
||||
defer pss.mu.RUnlock()
|
||||
|
||||
if _, ok := pss.store[pname]; ok {
|
||||
return true
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
// DeleteDevice deletes `pname` from tracking
|
||||
func (pss *PeripheralStateStore) DeleteDevice(pname string) error {
|
||||
if !pss.DeviceExists(pname) {
|
||||
return ErrNoSuchDevice
|
||||
}
|
||||
|
||||
pss.mu.Lock()
|
||||
defer pss.mu.Unlock()
|
||||
|
||||
delete(pss.store, pname)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// "Events" [START] ------------------------------------------------------------
|
||||
func (ds *DeviceService) onDeviceTimedOut(pname string) {
|
||||
// We need to deregister the device
|
||||
ds.PSS.setDeviceTimedOut(pname)
|
||||
}
|
||||
|
||||
// "Events" [END] --------------------------------------------------------------
|
||||
|
||||
// Device State Handling [START] -----------------------------------------------
|
||||
func (pss *PeripheralStateStore) setDeviceConnected(pname string, per peripheral.Peripheral) {
|
||||
pss.Logger.Info("found device", "device", pname)
|
||||
|
||||
pss.mu.Lock()
|
||||
defer pss.mu.Unlock()
|
||||
pss.store[pname].Peripheral = per
|
||||
pss.store[pname].State = DeviceIsConnected
|
||||
}
|
||||
|
||||
func (pss *PeripheralStateStore) setDeviceTimedOut(pname string) {
|
||||
pss.Logger.Info("device timed out", "device", pname)
|
||||
|
||||
pss.mu.Lock()
|
||||
defer pss.mu.Unlock()
|
||||
pss.store[pname].Peripheral = nil
|
||||
pss.store[pname].State = DeviceTimedOut
|
||||
}
|
||||
|
||||
func (pss *PeripheralStateStore) setDeviceReconnected(pname string, per peripheral.Peripheral) {
|
||||
pss.Logger.Info("device reconnecting", "device", pname)
|
||||
|
||||
pss.mu.Lock()
|
||||
defer pss.mu.Unlock()
|
||||
pss.store[pname].Peripheral = per
|
||||
pss.store[pname].State = DeviceReconnected
|
||||
}
|
||||
|
||||
// Device State Handling [END] -------------------------------------------------
|
||||
|
||||
// Device Handling [START] -----------------------------------------------------
|
||||
func (pss *PeripheralStateStore) HandleDeviceState() {
|
||||
peripherals := pss.GetStates()
|
||||
|
||||
for pName, pState := range peripherals {
|
||||
switch pState.State {
|
||||
case DeviceIsDisconnected:
|
||||
pss.Logger.Debug("looking for device", "name", pName)
|
||||
dev, err := pState.device.Discover()
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
// Register the device we just found
|
||||
pss.setDeviceConnected(pName, dev)
|
||||
case DeviceIsConnected:
|
||||
// Need to check if its streaming, if its not streaming than we have to do a healthcheck
|
||||
pss.Logger.Debug("Device is connected. Normal", "device", pName)
|
||||
case DeviceReconnected:
|
||||
// If the device has reconnected we need to reset the device and then set it as connected
|
||||
pss.Logger.Debug("Device has reconnected. Clearing up state", "device", pName)
|
||||
case DeviceTimedOut:
|
||||
// If the device has timed out we need to re-discover it or something
|
||||
pss.Logger.Debug("Device is timed out. Attempting to recconect", "device", pName)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Device Handling [END] -------------------------------------------------------
|
||||
|
||||
// FindDevices is a routine that goes over the devices in the PeripheralStateStore
|
||||
// and handles their state accordingly
|
||||
func (ds *DeviceService) FindDevices() {
|
||||
ticker := time.NewTicker(2 * time.Second)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ds.ctxDiscovery.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
ds.PSS.HandleDeviceState()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -102,7 +102,7 @@ func (t *TelemetryService) SubscribeToFields() []telem.FieldID {
|
||||
seen := make(map[telemetry.FieldID]struct{})
|
||||
var allFields []telemetry.FieldID
|
||||
|
||||
for _, dev := range t.devService.Devices {
|
||||
for _, dev := range t.devService.GetDevices() {
|
||||
for _, field := range dev.RequiredFields() {
|
||||
if _, exists := seen[field]; !exists {
|
||||
seen[field] = struct{}{}
|
||||
|
||||
Reference in New Issue
Block a user