Device reconnection handling #15
@@ -0,0 +1,9 @@
|
|||||||
|
package cdashdisplay
|
||||||
|
|
||||||
|
func (cds *CDashDisplay) OnLoad() error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (cds *CDashDisplay) OnTelemetryProviderFound() error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
@@ -0,0 +1,9 @@
|
|||||||
|
package uidevice
|
||||||
|
|
||||||
|
func (uid *UIDevice) OnLoad() error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (uid *UIDevice) OnTelemetryProviderFound() error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
@@ -20,6 +20,8 @@ type Peripheral interface {
|
|||||||
Setup() error
|
Setup() error
|
||||||
SendData(*telemetry.TelemetryData) error
|
SendData(*telemetry.TelemetryData) error
|
||||||
RequiredFields() []telemetry.FieldID
|
RequiredFields() []telemetry.FieldID
|
||||||
|
OnLoad() error
|
||||||
|
OnTelemetryProviderFound() error
|
||||||
Close() error
|
Close() error
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+7
-5
@@ -22,15 +22,17 @@ type DeviceService struct {
|
|||||||
TelemCh <-chan telemetry.TelemetryData
|
TelemCh <-chan telemetry.TelemetryData
|
||||||
// Output
|
// Output
|
||||||
Messages chan string
|
Messages chan string
|
||||||
|
// Callbacks
|
||||||
|
OnTelemetryProviderDiscovered func()
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewDeviceService(logger *slog.Logger) *DeviceService {
|
func NewDeviceService(logger *slog.Logger, msg chan string) *DeviceService {
|
||||||
sharedChannel := make(chan string, 10)
|
|
||||||
|
|
||||||
dev := &DeviceService{
|
dev := &DeviceService{
|
||||||
PSS: NewPeripheralStateStore(logger.With("Service", "PeripheralStateStore"), devices.List),
|
PSS: NewPeripheralStateStore(
|
||||||
|
logger.With("Service", "PeripheralStateStore"), devices.List, msg,
|
||||||
|
),
|
||||||
Logger: logger,
|
Logger: logger,
|
||||||
Messages: sharedChannel,
|
Messages: msg,
|
||||||
}
|
}
|
||||||
|
|
||||||
// Start the routine that looks for devices - should always be running in the background
|
// Start the routine that looks for devices - should always be running in the background
|
||||||
|
|||||||
@@ -0,0 +1 @@
|
|||||||
|
package services
|
||||||
@@ -2,6 +2,7 @@ package services
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"errors"
|
"errors"
|
||||||
|
"fmt"
|
||||||
"log/slog"
|
"log/slog"
|
||||||
"maps"
|
"maps"
|
||||||
"sync"
|
"sync"
|
||||||
@@ -51,15 +52,19 @@ type PeripheralStateStore struct {
|
|||||||
Logger *slog.Logger
|
Logger *slog.Logger
|
||||||
mu sync.RWMutex
|
mu sync.RWMutex
|
||||||
store map[string]*PeripheralState
|
store map[string]*PeripheralState
|
||||||
|
// Messaging for UI and stuff
|
||||||
|
Messages chan string
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewPeripheralStateStore(
|
func NewPeripheralStateStore(
|
||||||
nLogger *slog.Logger,
|
nLogger *slog.Logger,
|
||||||
devList map[string]*devices.Device,
|
devList map[string]*devices.Device,
|
||||||
|
msg chan string,
|
||||||
) *PeripheralStateStore {
|
) *PeripheralStateStore {
|
||||||
store := PeripheralStateStore{
|
store := PeripheralStateStore{
|
||||||
Logger: nLogger,
|
Logger: nLogger,
|
||||||
store: make(map[string]*PeripheralState),
|
store: make(map[string]*PeripheralState),
|
||||||
|
Messages: msg,
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, dev := range devList {
|
for _, dev := range devList {
|
||||||
@@ -154,6 +159,8 @@ func (pss *PeripheralStateStore) setDeviceConnected(pname string, per peripheral
|
|||||||
defer pss.mu.Unlock()
|
defer pss.mu.Unlock()
|
||||||
pss.store[pname].Peripheral = per
|
pss.store[pname].Peripheral = per
|
||||||
pss.store[pname].State = DeviceIsConnected
|
pss.store[pname].State = DeviceIsConnected
|
||||||
|
|
||||||
|
pss.Messages <- fmt.Sprintf("Device successfuly connected: %s\n", pname)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (pss *PeripheralStateStore) setDeviceTimedOut(pname string) {
|
func (pss *PeripheralStateStore) setDeviceTimedOut(pname string) {
|
||||||
|
|||||||
@@ -1,2 +1,34 @@
|
|||||||
// Package services interacts with the other libraries required for this UI
|
// Package services interacts with the other libraries required for this UI
|
||||||
package services
|
package services
|
||||||
|
|
||||||
|
import (
|
||||||
|
"errors"
|
||||||
|
"log/slog"
|
||||||
|
)
|
||||||
|
|
||||||
|
type Orchestrator struct {
|
||||||
|
DeviceService *DeviceService
|
||||||
|
TelemetryService *TelemetryService
|
||||||
|
Messages chan string
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewOrchestrator(logger *slog.Logger) (*Orchestrator, error) {
|
||||||
|
msg := make(chan string, 10)
|
||||||
|
|
||||||
|
devService := NewDeviceService(logger.With("service", "DeviceService"), msg)
|
||||||
|
|
||||||
|
telemService := NewTelemetryService(logger.With("service", "TelemetryService"), devService, msg)
|
||||||
|
if telemService == nil {
|
||||||
|
return nil, errors.New("failed to create telemetry service")
|
||||||
|
}
|
||||||
|
|
||||||
|
go telemService.FindProvider(telemService.CtxMonitor)
|
||||||
|
|
||||||
|
// Need to setup the callbacks on the services
|
||||||
|
|
||||||
|
return &Orchestrator{
|
||||||
|
DeviceService: devService,
|
||||||
|
TelemetryService: telemService,
|
||||||
|
Messages: msg,
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|||||||
+9
-31
@@ -32,16 +32,21 @@ type TelemetryService struct {
|
|||||||
cancelMonitor context.CancelFunc
|
cancelMonitor context.CancelFunc
|
||||||
CtxHealthcheck context.Context
|
CtxHealthcheck context.Context
|
||||||
healthCheckCancel context.CancelFunc
|
healthCheckCancel context.CancelFunc
|
||||||
|
// Callbacks
|
||||||
|
OnDevicesDiscovered func()
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewTelemetryService(logger *slog.Logger, devServo *DeviceService) *TelemetryService {
|
func NewTelemetryService(
|
||||||
sharedChannel := make(chan string, 10)
|
logger *slog.Logger,
|
||||||
|
devServo *DeviceService,
|
||||||
|
msg chan string,
|
||||||
|
) *TelemetryService {
|
||||||
newService := &TelemetryService{
|
newService := &TelemetryService{
|
||||||
logger: logger,
|
logger: logger,
|
||||||
isConnected: false,
|
isConnected: false,
|
||||||
devService: devServo,
|
devService: devServo,
|
||||||
listeners: make(map[string]chan telem.TelemetryData),
|
listeners: make(map[string]chan telem.TelemetryData),
|
||||||
Messages: sharedChannel,
|
Messages: msg,
|
||||||
}
|
}
|
||||||
newService.CtxMonitor, newService.cancelMonitor = context.WithCancel(context.Background())
|
newService.CtxMonitor, newService.cancelMonitor = context.WithCancel(context.Background())
|
||||||
|
|
||||||
@@ -59,6 +64,7 @@ func (t *TelemetryService) ProviderMonitor(ctx context.Context) {
|
|||||||
case <-ticker.C:
|
case <-ticker.C:
|
||||||
slog.Info("checking if provider is still running")
|
slog.Info("checking if provider is still running")
|
||||||
if !t.activeProvider.IsAlive(500 * time.Millisecond) {
|
if !t.activeProvider.IsAlive(500 * time.Millisecond) {
|
||||||
|
t.Messages <- "Healthcheck on provider failing. Dropping provider.\n"
|
||||||
slog.Warn("provider healthcheck failed")
|
slog.Warn("provider healthcheck failed")
|
||||||
t.dropActiveProvider()
|
t.dropActiveProvider()
|
||||||
t.onProviderHealthCheckFailed()
|
t.onProviderHealthCheckFailed()
|
||||||
@@ -154,34 +160,6 @@ func (t *TelemetryService) SwitchProvider(newProvider telem.TelemetryProvider) e
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (t *TelemetryService) onProviderHealthCheckFailed() {
|
|
||||||
// Just restart the whole lookup process
|
|
||||||
go t.FindProvider(t.CtxMonitor)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (t *TelemetryService) onFindProvider(prov telem.TelemetryProvider) {
|
|
||||||
// Attach to the provider
|
|
||||||
t.logger.Info("found provider for " + prov.Name())
|
|
||||||
err := t.SwitchProvider(prov)
|
|
||||||
if err != nil {
|
|
||||||
t.logger.Error("failed to switch to provider onFindProvider", "err", err)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
// 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
|
|
||||||
// TODO: now we need to also clear the devices to restart everything,
|
|
||||||
// if the stream stopped we have to restart the devices and everything
|
|
||||||
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:
|
// TODO: add some way of retriggering this. Currently it should:
|
||||||
// start monitoring on startup -> find provider -> stop monitoring (when game closes for example)
|
// start monitoring on startup -> find provider -> stop monitoring (when game closes for example)
|
||||||
func (t *TelemetryService) FindProvider(ctx context.Context) {
|
func (t *TelemetryService) FindProvider(ctx context.Context) {
|
||||||
|
|||||||
@@ -0,0 +1,41 @@
|
|||||||
|
package services
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
|
||||||
|
telem "esdi/telemetry"
|
||||||
|
)
|
||||||
|
|
||||||
|
func (t *TelemetryService) onProviderHealthCheckFailed() {
|
||||||
|
// Just restart the whole lookup process
|
||||||
|
go t.FindProvider(t.CtxMonitor)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t *TelemetryService) onFindProvider(prov telem.TelemetryProvider) {
|
||||||
|
// Attach to the provider
|
||||||
|
t.logger.Info("found provider for " + prov.Name())
|
||||||
|
t.Messages <- fmt.Sprintf("Found provider \"%s\"\n", prov.Name())
|
||||||
|
err := t.SwitchProvider(prov)
|
||||||
|
if err != nil {
|
||||||
|
t.Messages <- fmt.Sprintf("Failed to switch to provider: %+v\n", err.Error())
|
||||||
|
t.logger.Error("failed to switch to provider onFindProvider", "err", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// Start the healthcheck on our provider so we can drop it if it stops
|
||||||
|
t.CtxHealthcheck, t.healthCheckCancel = context.WithCancel(context.Background())
|
||||||
|
go t.ProviderMonitor(t.CtxHealthcheck)
|
||||||
|
|
||||||
|
// Tell the devices service we got a provider
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t *TelemetryService) onProviderStopsMidStream() {
|
||||||
|
// clear the current provider
|
||||||
|
// TODO: now we need to also clear the devices to restart everything,
|
||||||
|
// if the stream stopped we have to restart the devices and everything
|
||||||
|
t.Messages <- "Telemetry provider stopped mid stream\n"
|
||||||
|
t.logger.Info("cleaning dropped provider and restarting lookup service")
|
||||||
|
t.dropActiveProvider()
|
||||||
|
go t.FindProvider(t.CtxMonitor)
|
||||||
|
}
|
||||||
@@ -16,19 +16,18 @@ type DeviceController struct {
|
|||||||
DeviceAPIView *views.DeviceAPIView
|
DeviceAPIView *views.DeviceAPIView
|
||||||
LayoutCtrl *LayoutController
|
LayoutCtrl *LayoutController
|
||||||
StreamCtrl *StreamingCtrl
|
StreamCtrl *StreamingCtrl
|
||||||
DevService *serv.DeviceService
|
Orchestrator *serv.Orchestrator
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewDeviceController(
|
func NewDeviceController(
|
||||||
base *Controller,
|
base *Controller,
|
||||||
devService *serv.DeviceService,
|
orchestrator *serv.Orchestrator,
|
||||||
telemService *serv.TelemetryService,
|
|
||||||
) *DeviceController {
|
) *DeviceController {
|
||||||
mc := &DeviceController{
|
mc := &DeviceController{
|
||||||
Controller: base,
|
Controller: base,
|
||||||
LayoutCtrl: NewLayoutController(base, devService),
|
LayoutCtrl: NewLayoutController(base, orchestrator.DeviceService),
|
||||||
DevService: devService,
|
Orchestrator: orchestrator,
|
||||||
StreamCtrl: NewStreamingCtrl(base, devService, telemService),
|
StreamCtrl: NewStreamingCtrl(base, orchestrator.DeviceService, orchestrator.TelemetryService),
|
||||||
}
|
}
|
||||||
|
|
||||||
return mc
|
return mc
|
||||||
@@ -57,7 +56,7 @@ func (mc *DeviceController) setDeviceAPIViewEvents() {
|
|||||||
SetInputCapture(func(ev *tcell.EventKey) *tcell.EventKey {
|
SetInputCapture(func(ev *tcell.EventKey) *tcell.EventKey {
|
||||||
switch ev.Rune() {
|
switch ev.Rune() {
|
||||||
case 'r':
|
case 'r':
|
||||||
go mc.DevService.FindDevices()
|
go mc.Orchestrator.DeviceService.FindDevices()
|
||||||
}
|
}
|
||||||
return ev
|
return ev
|
||||||
})
|
})
|
||||||
@@ -67,8 +66,8 @@ func (mc *DeviceController) AddDeviceAPIListItems() {
|
|||||||
mc.DeviceAPIView.DevAPIList.
|
mc.DeviceAPIView.DevAPIList.
|
||||||
AddItem("layout", "build a layout for CDashDisplay", func() {
|
AddItem("layout", "build a layout for CDashDisplay", func() {
|
||||||
// This CDashDisplay specific, only load if we have a CDashDisplay
|
// This CDashDisplay specific, only load if we have a CDashDisplay
|
||||||
if !mc.DevService.PeripheralExists(cdashdisplay.NAME) {
|
if !mc.Orchestrator.DeviceService.PeripheralExists(cdashdisplay.NAME) {
|
||||||
mc.DevService.Messages <- "CDashDisplay it not loaded yet\n"
|
mc.Orchestrator.DeviceService.Messages <- "CDashDisplay it not loaded yet\n"
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -116,7 +115,7 @@ func (mc *DeviceController) injectControllerCallbacks() {
|
|||||||
|
|
||||||
func (mc *DeviceController) injectChannels() {
|
func (mc *DeviceController) injectChannels() {
|
||||||
go func() {
|
go func() {
|
||||||
for msg := range mc.DevService.Messages {
|
for msg := range mc.Orchestrator.Messages {
|
||||||
mc.PrintToOutputWindow(msg)
|
mc.PrintToOutputWindow(msg)
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|||||||
@@ -59,11 +59,6 @@ func NewStreamingCtrl(
|
|||||||
return ctrl
|
return ctrl
|
||||||
}
|
}
|
||||||
|
|
||||||
// func (sc *StreamingCtrl) subscribeListeners() {
|
|
||||||
// // Here I will set a UIDevice
|
|
||||||
// sc.TelemetryCh = sc.TelemServ.SubscribeListener("UI", 1)
|
|
||||||
// }
|
|
||||||
|
|
||||||
func (sc *StreamingCtrl) registerHooks() {
|
func (sc *StreamingCtrl) registerHooks() {
|
||||||
sc.StreamView.Options.Form.SetInputCapture(func(ev *tcell.EventKey) *tcell.EventKey {
|
sc.StreamView.Options.Form.SetInputCapture(func(ev *tcell.EventKey) *tcell.EventKey {
|
||||||
switch ev.Key() {
|
switch ev.Key() {
|
||||||
|
|||||||
+5
-9
@@ -23,19 +23,15 @@ func NewControlPanel(logger *slog.Logger) *ControlPanel {
|
|||||||
App: tview.NewApplication(),
|
App: tview.NewApplication(),
|
||||||
}
|
}
|
||||||
|
|
||||||
// NOTE: create our device service here
|
orchestrator, err := services.NewOrchestrator(logger)
|
||||||
devService := services.NewDeviceService(logger.With("service", "DeviceService"))
|
if err != nil {
|
||||||
|
// TODO: no panic here
|
||||||
telemService := services.NewTelemetryService(logger, devService)
|
panic("failed to create services orchestrator")
|
||||||
if telemService == nil {
|
|
||||||
panic("failed to create the telemetry service")
|
|
||||||
}
|
}
|
||||||
|
|
||||||
go telemService.FindProvider(telemService.CtxMonitor)
|
|
||||||
|
|
||||||
return &ControlPanel{
|
return &ControlPanel{
|
||||||
Controller: baseController,
|
Controller: baseController,
|
||||||
DeviceController: controllers.NewDeviceController(baseController, devService, telemService),
|
DeviceController: controllers.NewDeviceController(baseController, orchestrator),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user