working downstream

main
KeeganForelight 2 years ago
parent efd91b1c90
commit 42ce886114

@ -1,13 +1,55 @@
### Planning ### Planning
**Monitoring Changes** **Monitoring Changes**
I want to refactor the reactor stuff to be less method oriented as far as data collection. For example, the monitoring stuff is all about events that happen pretty infrequently. It makes sense to then use a channel on the device side to just feed relevant status updates back to the reactor. I think that this makes the most sense because this will synchronize updates and leverage the rarity of events to cut down on errant calls. I want to refactor the reactor stuff to be less method oriented as far as data collection. For example, the monitoring stuff is all about events that happen pretty infrequently. It makes sense to then use a channel on the device side to just feed relevant status updates back to the reactor. I think that this makes the most sense because this will synchronize updates and leverage the rarity of events to cut down on errant calls.
- pros - pros
- less repitive method calls needed - less repitive method calls needed
- less device locking
- localize the information to different packages - localize the information to different packages
- cons - cons
- extra memory for channels and duplicate storage info - extra memory for channels and duplicate storage info
- could just remove status from dm? - could just remove status from dm?
**New Idea**
I can leverage wireguard to do server-> reactor connections even beyond the testing phase
Changes:
1) move device coordinator into device package
2) expose relevant methods to reactor interface
3) clarify individual package responsibilities
4) add stuff server side to create/destroy grpc connections as the information is rendered client side
- this might be scuffed but oh well
### Package Separation
**Reactor**
- coordinator
- creates initial link to the server
- creates database client
- creates and starts a device coordinator
**Device**
- coordinator
- searches i2c bus for connected devices
- spins up managers to control the connected devices
- relays information back up to the reactor coordinator
- manager
- control over singular device
- has the core information that will be needed across any type of device (name, status, address etc)
- sub-manager
- fine grained struct with methods specific to the device
**Server**
Going to ignore for now because I am lazy
- central coordinator starts up database connection config etc
- reactor coordinator
### TODO ### TODO
**Monitoring Changes** **Monitoring Changes**

@ -23,7 +23,7 @@ type I2CClient struct {
sync.Mutex sync.Mutex
} }
func NewI2CClient(config *viper.Viper) (*I2CClient, error) { func NewClient(config *viper.Viper) (*I2CClient, error) {
var err error var err error
var bus int var bus int
client := &I2CClient{} client := &I2CClient{}

@ -10,6 +10,7 @@ import (
type Manager interface { type Manager interface {
Start() error Start() error
Exit() error Exit() error
IsActive() int
} }
func NewManager(max int) Manager { func NewManager(max int) Manager {

@ -3,7 +3,10 @@ package controller
// do sensor and methods // do sensor and methods
import ( import (
"fmt"
"sync" "sync"
"github.com/spf13/viper"
) )
type PWMManager struct { type PWMManager struct {
@ -26,6 +29,10 @@ func (m *PWMManager) GetFrequency() (int, error) {
return m.Frequency, nil return m.Frequency, nil
} }
func (m *PWMManager) LoadConfig(config *viper.Viper, key string) {
fmt.Printf("config\n")
}
func (m *PWMManager) GetDefaultName() string { func (m *PWMManager) GetDefaultName() string {
return "pwm controller" return "pwm controller"
} }

@ -1,25 +1,147 @@
package device package device
// serves as a server side device coordinator to sync import (
"FRMS/internal/pkg/I2C"
pb "FRMS/internal/pkg/grpc"
"FRMS/internal/pkg/manager"
"fmt"
"sync"
"time"
// assume has a server connection from DM "github.com/spf13/viper"
// STEPS )
// 1) client loads web page
// 2) DC pushes what it has to the client
// 3) requests DM for what it doesnt
// 4) DM responds
// 5) DC forwards responses to client
// What can happen // Created by rlc to manage devices
// - Client can push something to client
// - Client requests info
// - If this happens, we can push it to client when we get it. Just need to know it was requests
// basic manager to embed
type Manager interface {
Start() error
Exit() error
// create a heartbeat to send to chan at intervals
HeartBeat(chan struct{}, int, int, time.Duration)
}
func NewManager() Manager {
// dont need timeout functionality
return manager.New(0)
}
// I2C client for locking
type I2CClient interface {
// i2c client w/ locking
GetConnected() (map[int]bool, error) // gets connected addresses
SendCmd(int, string) (string, error)
}
func NewI2CClient(config *viper.Viper) (I2CClient, error) {
return I2C.NewClient(config)
}
// device coordinator itself
type DeviceCoordinator struct { type DeviceCoordinator struct {
NameChan chan string I2C I2CClient
StatusChan chan string Manager
Config *viper.Viper
managersMu sync.RWMutex
Managers map[int]*DeviceManager
}
func NewCoordinator(config *viper.Viper) *DeviceCoordinator {
dm := make(map[int]*DeviceManager)
m := NewManager()
c := &DeviceCoordinator{
Manager: m,
Managers: dm,
Config: config,
}
return c
}
func (c *DeviceCoordinator) Start() error {
var err error
if err = c.Manager.Start(); err != nil {
return err
}
if c.I2C, err = NewI2CClient(c.Config); err != nil {
return err
}
go c.Monitor()
return err
} }
func ClientSetName() { func (c *DeviceCoordinator) Monitor() {
// pass client set names to DM // monitor I2C for new devices
ch := make(chan struct{})
go c.HeartBeat(ch, 10, 0, time.Second)
for range ch {
// on notification (10s)
devs, err := c.I2C.GetConnected()
if err != nil {
panic(err)
}
// update list
go c.UpdateManagers(devs)
}
}
func (c *DeviceCoordinator) UpdateManagers(active map[int]bool) {
// updates managers
c.managersMu.Lock()
defer c.managersMu.Unlock()
for addr, dm := range c.Managers {
_, ok := active[addr]
if ok && dm.IsActive() == 0 {
// active and dm not
if err := dm.Start(); err != nil {
panic(err)
}
} else if dm.IsActive() == 1 {
// not active and dm is
if err := dm.Exit(); err != nil {
panic(err)
}
}
// remove from map
delete(active, addr)
}
for addr, _ := range active {
// no manager, create one
fmt.Printf("New device %d!\n", addr)
dm, err := NewDeviceManager(addr, c.Config, "", c.I2C)
if err != nil {
panic(err)
}
if err := dm.Start(); err != nil {
panic(err)
}
c.Managers[addr] = dm
}
}
func (c *DeviceCoordinator) GetDeviceInfo() ([]*pb.Device, error) {
// gets device info for monitoring
c.managersMu.RLock()
defer c.managersMu.RUnlock()
var devices []*pb.Device
for addr, dm := range c.Managers {
// looping over devices
devices = append(devices, &pb.Device{
Addr: int32(addr),
Status: pb.Status(dm.IsActive()),
})
}
return devices, nil
} }

@ -10,7 +10,9 @@ import (
type SubManager interface { type SubManager interface {
Start(int) error Start(int) error
Exit() error Exit() error
IsActive() int
String() string // printing info about the sub manager String() string // printing info about the sub manager
LoadConfig(*viper.Viper, string)
// for config bs // for config bs
GetDefaultName() string GetDefaultName() string
@ -23,23 +25,22 @@ type NameChan struct {
// base device manager // base device manager
type DeviceManager struct { type DeviceManager struct {
SubManager `mapstructure:",squash"` SubManager
// across controllers/sensors // across controllers/sensors
Address int `mapstructure:"address"` Address int `mapstructure:"address"`
Name string `mapstructure:"name"`
infoMu sync.RWMutex infoMu sync.RWMutex
Name string `mapstructure:"name"`
Status string // easier to remember
Config *viper.Viper Config *viper.Viper
ConfigPrefix string ConfigPrefix string
} }
func NewDeviceManager(addr int, config *viper.Viper, configPrefix string) (*DeviceManager, error) { func NewDeviceManager(addr int, config *viper.Viper, configPrefix string, i2c I2CClient) (*DeviceManager, error) {
// validate prefix // validate prefix
s, err := NewSubManager(addr) s, err := NewSubManager(addr, i2c)
dm := &DeviceManager{ dm := &DeviceManager{
SubManager: s, SubManager: s,
@ -61,26 +62,25 @@ func (m *DeviceManager) LoadConfig() error {
} }
m.Config.UnmarshalKey(mainKey, m) m.Config.UnmarshalKey(mainKey, m)
m.SubManager.LoadConfig(m.Config, mainKey)
return nil return nil
} }
func (m *DeviceManager) Start() error { func (m *DeviceManager) Start() error {
return m.SubManager.Start(m.Address) // load config and then start
} var err error
func (m *DeviceManager) UpdateStatus(status string) error { // load config
// updates device status if err = m.LoadConfig(); err != nil {
m.infoMu.Lock() return err
defer m.infoMu.Unlock() }
m.Status = status
return nil // start
} if err = m.SubManager.Start(m.Address); err != nil {
return err
}
func (m *DeviceManager) GetStatus() string { return err
// updates device status
m.infoMu.RLock()
defer m.infoMu.RUnlock()
return m.Status
} }
// dev info grpc handlers // dev info grpc handlers

@ -11,20 +11,20 @@ import (
Returns the correct manager for sensor/controller Returns the correct manager for sensor/controller
*/ */
func NewSubManager(addr int) (SubManager, error) { func NewSubManager(addr int, i2c I2CClient) (SubManager, error) {
// returns correct device manager by ID // returns correct device manager by ID
var m SubManager var m SubManager
var err error var err error
switch addr { switch addr {
case 97: case 97:
// DO // DO
m = sensor.NewDOManager() m = sensor.NewDOManager(i2c)
case 99: case 99:
// pH // pH
m = sensor.NewPHManager() m = sensor.NewPHManager(i2c)
case 102: case 102:
// RTD // RTD
m = sensor.NewRTDManager() m = sensor.NewRTDManager(i2c)
case 256: case 256:
// PWM // PWM
m = controller.NewPWMManager() m = controller.NewPWMManager()

@ -50,6 +50,10 @@ func (m *Manager) Exit() error {
return errors.New("Manager not active!") return errors.New("Manager not active!")
} }
func (m *Manager) IsActive() int {
return int(atomic.LoadInt32(&m.Active))
}
// Heartbeat tracker // Heartbeat tracker
func (m *Manager) HeartBeat(ping chan struct{}, hb, interval int, scale time.Duration) { func (m *Manager) HeartBeat(ping chan struct{}, hb, interval int, scale time.Duration) {

@ -3,7 +3,7 @@ package reactor
// file describes reactor level coordinator and associated implementation // file describes reactor level coordinator and associated implementation
import ( import (
"FRMS/internal/pkg/I2C" "FRMS/internal/pkg/device"
pb "FRMS/internal/pkg/grpc" pb "FRMS/internal/pkg/grpc"
"FRMS/internal/pkg/influxdb" "FRMS/internal/pkg/influxdb"
"FRMS/internal/pkg/logging" "FRMS/internal/pkg/logging"
@ -31,7 +31,7 @@ func NewManager(max int) Manager {
return manager.New(max) return manager.New(max)
} }
// db client interface // db client
type DBClient interface { type DBClient interface {
// //
Start() error Start() error
@ -41,14 +41,15 @@ func NewDBClient(config *viper.Viper) (DBClient, error) {
return influxdb.NewDBClient(config) return influxdb.NewDBClient(config)
} }
type I2CClient interface { // device coordinator
// simple client to push responsibilites to sensor type DeviceCoordinator interface {
GetConnected() (map[int]bool, error) // gets all connected addr Start() error
SendCmd(int, string) (string, error) // send cmd, string is return // in grpc format
GetDeviceInfo() ([]*pb.Device, error)
} }
func NewI2CClient(config *viper.Viper) (I2CClient, error) { func NewDeviceCoordinator(config *viper.Viper) DeviceCoordinator {
return I2C.NewI2CClient(config) return device.NewCoordinator(config)
} }
type Server struct { type Server struct {
@ -71,11 +72,10 @@ type ReactorCoordinator struct {
Info `mapstructure:",squash"` Info `mapstructure:",squash"`
Database DBClient Database DBClient
I2C I2CClient
MonitoringClient pb.MonitoringClient // grpc pb.MonitoringClient // grpc embedding
*DeviceCoordinator // struct for locking DeviceCoordinator // struct for locking
Err chan error Err chan error
} }
@ -111,12 +111,12 @@ func (c *ReactorCoordinator) Start() {
c.Err <- err c.Err <- err
} }
// loading clients if err = c.DeviceCoordinator.Start(); err != nil {
if c.Database, err = NewDBClient(c.Config); err != nil {
c.Err <- err c.Err <- err
} }
if c.I2C, err = NewI2CClient(c.Config); err != nil { // loading clients
if c.Database, err = NewDBClient(c.Config); err != nil {
c.Err <- err c.Err <- err
} }
@ -168,12 +168,7 @@ func (c *ReactorCoordinator) Monitor() {
for range ch { for range ch {
// check devs and ping // check devs and ping
logging.Debug(logging.DClient, "RLC Pinging server") logging.Debug(logging.DClient, "RLC Pinging server")
// this can probably be offloaded // ping central server with status
active, err := c.I2C.GetConnected()
if err != nil {
c.Err <- err
}
go c.DeviceCoordinator.UpdateDevices(c.Config, c.I2C, active)
go c.Ping() go c.Ping()
} }
} }
@ -230,3 +225,26 @@ func (c *ReactorCoordinator) Connect(ip string, port int) (*grpc.ClientConn, err
} }
return conn, nil return conn, nil
} }
func (c *ReactorCoordinator) Ping() {
// send device info to central coordinator
fmt.Printf("Pinging server\n")
var devices []*pb.Device
var err error
if devices, err = c.GetDeviceInfo(); err != nil {
c.Err <- err
}
// create request
req := &pb.ReactorStatusPing{
Id: int32(c.ID),
Devices: devices,
}
// ping server
if _, err = c.ReactorStatusHandler(context.Background(), req); err != nil {
c.Err <- err
}
}

@ -1,100 +0,0 @@
package reactor
import (
"FRMS/internal/pkg/device"
pb "FRMS/internal/pkg/grpc"
"sync"
"github.com/spf13/viper"
)
type DeviceManager interface {
Start() error
Exit() error
GetStatus() string
LoadConfig() error
}
func NewDeviceManager(addr int, config *viper.Viper, prefix string) (DeviceManager, error) {
return device.NewDeviceManager(addr, config, prefix)
}
type DeviceCoordinator struct {
Config *viper.Viper
Managers map[int]DeviceManager
sync.RWMutex
}
func NewDeviceCoordinator(config *viper.Viper) *DeviceCoordinator {
dm := &DeviceCoordinator{Config: config}
dm.Managers = make(map[int]DeviceManager)
return dm
}
func (c *DeviceCoordinator) UpdateDevices(config *viper.Viper, i2c I2CClient, active map[int]bool) error {
// update device list
c.Lock()
defer c.Unlock()
for addr, _ := range active {
// loop over devs
if _, ok := c.Managers[addr]; !ok {
// no device, creating one
dm, err := NewDeviceManager(addr, c.Config, "")
if err != nil {
return err
}
// starting
if err = dm.Start(); err != nil {
return err
}
// loading config
if err = dm.LoadConfig(); err != nil {
return err
}
// update entry
c.Managers[addr] = dm
}
}
// all devs accounted for
// I can rework this to rely on individual devices to keep track of status and only need above
// for addr, dm := range c.Managers {
// if active[addr] {
// // active
// if dm.IsActive() != 1 {
// err = dm.Start()
// }
// } else {
// if dm.IsActive() != 0 {
// err = dm.Exit()
// }
// }
// if err != nil {
// return err
// }
// }
return nil
}
func (c *DeviceCoordinator) GetDevices() ([]*pb.Device, error) {
c.RLock()
defer c.RUnlock()
var err error
var devices []*pb.Device
for addr, dm := range c.Managers {
status := pb.Status(pb.Status_value[dm.GetStatus()])
devices = append(devices, &pb.Device{
Addr: int32(addr),
Status: status,
})
}
return devices, err
}

@ -1,27 +0,0 @@
package reactor
import (
"context"
"fmt"
//"FRMS/internal/pkg/logging"
//"google.golang.org/grpc"
pb "FRMS/internal/pkg/grpc"
)
// implements grpc handler and device data aggregater handler
// grpc status update handler
func (c *ReactorCoordinator) Ping() {
// sends all device status to central coordinator
fmt.Printf("Pinging coordinator\n")
// get devices
devices, err := c.GetDevices()
if err != nil {
c.Err <- err
}
req := &pb.ReactorStatusPing{Id: int32(c.ID), Devices: devices}
if _, err := c.MonitoringClient.ReactorStatusHandler(context.Background(), req); err != nil {
c.Err <- err
}
}

@ -2,6 +2,7 @@ package sensor
import ( import (
"errors" "errors"
"strconv"
"time" "time"
) )
@ -15,8 +16,8 @@ type Atlas struct {
// helper struct to embedd // helper struct to embedd
I2C I2CClient I2C I2CClient
// delays unmarshalled // delays unmarshalled
CalDelay int `mapstructure:"cal"` CalDelay int `mapstructure:"cal_delay"`
ReadDelay int `mapstructure:"read"` ReadDelay int `mapstructure:"read_delay"`
} }
func (a *Atlas) Calibrate(addr int, cal string) error { func (a *Atlas) Calibrate(addr int, cal string) error {
@ -42,10 +43,18 @@ func (a *Atlas) Read(addr int) (float32, error) {
//fmt.Printf("Error writing! %s", err) //fmt.Printf("Error writing! %s", err)
return 0, err return 0, err
} }
if a.ReadDelay == 0 {
return 0, errors.New("Read Delay unset, please check config")
}
sleep := time.Duration(a.ReadDelay) * time.Millisecond sleep := time.Duration(a.ReadDelay) * time.Millisecond
time.Sleep(sleep) // sleep between reads time.Sleep(sleep) // sleep between reads
//data, err := a.I2C.SendCmd(addr, "") data, err := a.I2C.SendCmd(addr, "")
return 0, nil if err != nil {
return 0, err
}
f, err := strconv.ParseFloat(data, 32)
return float32(f), err
} }
// for config // for config

@ -3,19 +3,22 @@ package sensor
// do sensor and methods // do sensor and methods
import ( import (
"fmt"
"sync" "sync"
"github.com/spf13/viper"
) )
type DOManager struct { type DOManager struct {
// do sensor manager // do sensor manager
*Atlas // atlas helper *Atlas `mapstructure:",squash"`
*SensorManager `mapstructure:",squash"` *SensorManager `mapstructure:",squash"`
sync.RWMutex sync.RWMutex
} }
func NewDOManager() *DOManager { func NewDOManager(i2c I2CClient) *DOManager {
a := &Atlas{} a := &Atlas{I2C: i2c}
sm := NewSensorManager(a.Read) sm := NewSensorManager(a.Read)
m := &DOManager{ m := &DOManager{
@ -25,6 +28,11 @@ func NewDOManager() *DOManager {
return m return m
} }
func (m *DOManager) LoadConfig(config *viper.Viper, key string) {
config.UnmarshalKey(key, m)
fmt.Printf("DO: %v\n", m.Atlas)
}
func (m *DOManager) GetDefaultName() string { func (m *DOManager) GetDefaultName() string {
return "DO Sensor" return "DO Sensor"
} }

@ -2,6 +2,7 @@ package sensor
import ( import (
"FRMS/internal/pkg/manager" "FRMS/internal/pkg/manager"
"fmt"
"sync" "sync"
"time" "time"
) )
@ -9,6 +10,7 @@ import (
type Manager interface { type Manager interface {
Start() error Start() error
Exit() error Exit() error
IsActive() int
HeartBeat(chan struct{}, int, int, time.Duration) HeartBeat(chan struct{}, int, int, time.Duration)
} }
@ -43,12 +45,17 @@ func (s *SensorManager) Start(addr int) error {
if err := s.Manager.Start(); err != nil { if err := s.Manager.Start(); err != nil {
return err return err
} }
fmt.Printf("Sensor %d: %v\n", addr, s)
// starting monitoring
go s.Monitor(addr) go s.Monitor(addr)
return nil return nil
} }
func (s *SensorManager) Monitor(addr int) { func (s *SensorManager) Monitor(addr int) {
ch := make(chan struct{}) // hb chan ch := make(chan struct{}) // hb chan
if s.SampleRate == 0 {
s.SampleRate = 5000
}
go s.HeartBeat(ch, s.SampleRate, 1000, time.Millisecond) go s.HeartBeat(ch, s.SampleRate, 1000, time.Millisecond)
for range ch { for range ch {
@ -61,6 +68,7 @@ func (r *Reading) TakeReading(addr int) {
if err != nil { if err != nil {
panic(err) panic(err)
} }
fmt.Printf("got sample: %v\n", sample)
r.Lock() r.Lock()
defer r.Unlock() defer r.Unlock()
r.Latest = sample r.Latest = sample

@ -3,19 +3,22 @@ package sensor
// do sensor and methods // do sensor and methods
import ( import (
"fmt"
"sync" "sync"
"github.com/spf13/viper"
) )
type PHManager struct { type PHManager struct {
// do sensor manager // do sensor manager
*Atlas *Atlas `mapstructure:",squash"`
*SensorManager `mapstructure:",squash"` *SensorManager `mapstructure:",squash"`
sync.RWMutex sync.RWMutex
} }
func NewPHManager() *PHManager { func NewPHManager(i2c I2CClient) *PHManager {
a := &Atlas{} a := &Atlas{I2C: i2c}
sm := NewSensorManager(a.Read) sm := NewSensorManager(a.Read)
m := &PHManager{ m := &PHManager{
SensorManager: sm, SensorManager: sm,
@ -24,6 +27,11 @@ func NewPHManager() *PHManager {
return m return m
} }
func (s *PHManager) LoadConfig(config *viper.Viper, key string) {
config.UnmarshalKey(key, s)
fmt.Printf("PH: %v\n", s.Atlas)
}
func (s *PHManager) GetDefaultName() string { func (s *PHManager) GetDefaultName() string {
return "pH Sensor" return "pH Sensor"
} }

@ -3,18 +3,21 @@ package sensor
// do sensor and methods // do sensor and methods
import ( import (
"fmt"
"sync" "sync"
"github.com/spf13/viper"
) )
type RTDManager struct { type RTDManager struct {
// do sensor manager // do sensor manager
*Atlas *Atlas `mapstructure:",squash"`
*SensorManager `mapstructure:",squash"` *SensorManager `mapstructure:",squash"`
sync.RWMutex sync.RWMutex
} }
func NewRTDManager() *RTDManager { func NewRTDManager(i2c I2CClient) *RTDManager {
a := &Atlas{} a := &Atlas{I2C: i2c}
sm := NewSensorManager(a.Read) sm := NewSensorManager(a.Read)
m := &RTDManager{ m := &RTDManager{
SensorManager: sm, SensorManager: sm,
@ -23,6 +26,11 @@ func NewRTDManager() *RTDManager {
return m return m
} }
func (s *RTDManager) LoadConfig(config *viper.Viper, key string) {
config.UnmarshalKey(key, s)
fmt.Printf("RTD: %v\n", s.Atlas)
}
func (s *RTDManager) GetDefaultName() string { func (s *RTDManager) GetDefaultName() string {
return "RTD Sensor" return "RTD Sensor"
} }

@ -1,7 +1,6 @@
package server package server
import ( import (
"FRMS/internal/pkg/device"
pb "FRMS/internal/pkg/grpc" pb "FRMS/internal/pkg/grpc"
"FRMS/internal/pkg/logging" "FRMS/internal/pkg/logging"
"FRMS/internal/pkg/manager" "FRMS/internal/pkg/manager"
@ -11,7 +10,6 @@ import (
"context" "context"
"fmt" "fmt"
_ "log" _ "log"
"sync"
"github.com/spf13/viper" "github.com/spf13/viper"
) )
@ -34,30 +32,30 @@ type ReactorManager struct {
// *ClientManager // client manager (OUTDATED) // *ClientManager // client manager (OUTDATED)
*Client // access to ID etc *Client // access to ID etc
// StatusMon *StatusMonitor putting on pause // StatusMon *StatusMonitor putting on pause
*ReactorDevices // *ReactorDevices
Config *viper.Viper // config to update Config *viper.Viper // config to update
Err chan error Err chan error
} }
type ReactorDevices struct { // type ReactorDevices struct {
// device struct // // device struct
Devices map[int]DeviceManager // Devices map[int]DeviceManager
sync.RWMutex // sync.RWMutex
} // }
func NewReactorManager(cl *Client, config *viper.Viper, errCh chan error) *ReactorManager { func NewReactorManager(cl *Client, config *viper.Viper, errCh chan error) *ReactorManager {
// making managers // making managers
m := NewManager(6) m := NewManager(6)
dm := make(map[int]DeviceManager) //dm := make(map[int]DeviceManager)
rd := &ReactorDevices{Devices: dm} //rd := &ReactorDevices{Devices: dm}
//cm := NewClientManager(cl) //cm := NewClientManager(cl)
r := &ReactorManager{ r := &ReactorManager{
//ClientManager: cm, //ClientManager: cm,
Manager: m, Manager: m,
Client: cl, Client: cl,
ReactorDevices: rd, //ReactorDevices: rd,
Config: config, Config: config,
Err: errCh, Err: errCh,
} }
return r return r
} }
@ -88,65 +86,69 @@ func (r *ReactorManager) ReactorStatusHandler(ctx context.Context, req *pb.React
//go r.PingReset() //go r.PingReset()
fmt.Printf("Recieved ping from %d!\n", req.GetId()) fmt.Printf("Recieved ping from %d!\n", req.GetId())
// update devices/sensors // update devices/sensors
go r.UpdateDevices(req.GetDevices()) for _, dev := range req.GetDevices() {
fmt.Printf("Device %d is %s ", dev.GetAddr(), dev.GetStatus().String())
}
fmt.Printf("\n")
// go r.UpdateDevices(req.GetDevices())
return &pb.ReactorStatusResponse{Id: int32(r.Id)}, nil return &pb.ReactorStatusResponse{Id: int32(r.Id)}, nil
} }
// device stuff // // device stuff
type DeviceManager interface { // type DeviceManager interface {
LoadConfig() error // LoadConfig() error
UpdateStatus(string) error // UpdateStatus(string) error
String() string // printing // String() string // printing
} // }
func NewDeviceManager(addr int, config *viper.Viper, prefix string) (DeviceManager, error) { // func NewDeviceManager(addr int, config *viper.Viper, prefix string) (DeviceManager, error) {
// returns a manager struct // // returns a manager struct
return device.NewDeviceManager(addr, config, prefix) // return device.NewDeviceManager(addr, config, prefix)
} // }
func (r *ReactorManager) UpdateDevices(devs []*pb.Device) { //func (r *ReactorManager) UpdateDevices(devs []*pb.Device) {
// pass updates to correct manager // // pass updates to correct manager
r.ReactorDevices.RLock() // read lock only // r.ReactorDevices.RLock() // read lock only
defer r.ReactorDevices.RUnlock() // defer r.ReactorDevices.RUnlock()
for _, dev := range devs { // for _, dev := range devs {
// looping over devs // // looping over devs
if dm, ok := r.ReactorDevices.Devices[int(dev.GetAddr())]; ok { // if dm, ok := r.ReactorDevices.Devices[int(dev.GetAddr())]; ok {
// device manager found // // device manager found
go dm.UpdateStatus(dev.GetStatus().String()) // go dm.UpdateStatus(dev.GetStatus().String())
//fmt.Println(dm) // //fmt.Println(dm)
} else { // } else {
// not found // // not found
go r.AddDevice(dev, r.Id, r.Config, r.Err) // go r.AddDevice(dev, r.Id, r.Config, r.Err)
} // }
} // }
} //}
func (r *ReactorDevices) AddDevice(dev *pb.Device, id int, config *viper.Viper, errCh chan error) { // func (r *ReactorDevices) AddDevice(dev *pb.Device, id int, config *viper.Viper, errCh chan error) {
// setting vars // // setting vars
prefix := fmt.Sprintf("reactors.%d.", id) // prefix := fmt.Sprintf("reactors.%d.", id)
addr := int(dev.GetAddr()) // addr := int(dev.GetAddr())
var dm DeviceManager // var dm DeviceManager
var err error // var err error
// write locking // // write locking
r.Lock() // r.Lock()
defer r.Unlock() // defer r.Unlock()
if dm, err = NewDeviceManager(addr, config, prefix); err != nil { // if dm, err = NewDeviceManager(addr, config, prefix); err != nil {
errCh <- err // errCh <- err
} // }
// setting status // // setting status
if err = dm.UpdateStatus(dev.GetStatus().String()); err != nil { // if err = dm.UpdateStatus(dev.GetStatus().String()); err != nil {
errCh <- err // errCh <- err
} // }
// loading config // // loading config
if err = dm.LoadConfig(); err != nil { // if err = dm.LoadConfig(); err != nil {
errCh <- err // errCh <- err
} // }
r.Devices[int(addr)] = dm // r.Devices[int(addr)] = dm
} // }

Loading…
Cancel
Save