stripping all status monitor and TUI stuff out for now, focusing on core reactor impl
KeeganForelight
2 years ago
parent
ef7bf9d665
commit
62f37b5f80
@ -1,149 +1,153 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
_ "log"
|
||||
"context"
|
||||
"sync"
|
||||
"FRMS/internal/pkg/logging"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/status"
|
||||
"google.golang.org/grpc/credentials/insecure"
|
||||
pb "FRMS/internal/pkg/grpc"
|
||||
pb "FRMS/internal/pkg/grpc"
|
||||
"FRMS/internal/pkg/logging"
|
||||
"context"
|
||||
"fmt"
|
||||
_ "log"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/credentials/insecure"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
// this package will implement a reactor coordinator and associated go routines
|
||||
// this package will implement a reactor manager and associated go routines
|
||||
|
||||
type ReactorManager struct {
|
||||
*Manager
|
||||
StatusMon *StatusMonitor
|
||||
*devstatus
|
||||
*Manager
|
||||
// StatusMon *StatusMonitor putting on pause
|
||||
*devstatus
|
||||
}
|
||||
|
||||
type devstatus struct {
|
||||
sync.Mutex
|
||||
Devs map[uint32]*DeviceInfo
|
||||
// keeping this around but not using it to create status for status mon
|
||||
sync.Mutex
|
||||
Devs map[int]*DeviceInfo
|
||||
}
|
||||
|
||||
func NewReactorManager(c *Client,sys *SystemViewer,err chan error) GeneralManager {
|
||||
r := &ReactorManager{}
|
||||
di := make(map[uint32]*DeviceInfo)
|
||||
r.devstatus = &devstatus{Devs:di}
|
||||
r.Manager = NewManager(err)
|
||||
r.StatusMon = NewStatusMonitor("Reactor",c.Id,sys)
|
||||
return r
|
||||
func NewReactorManager(c *Client, err chan error) GeneralManager {
|
||||
r := &ReactorManager{}
|
||||
di := make(map[int]*DeviceInfo)
|
||||
r.devstatus = &devstatus{Devs: di}
|
||||
r.Manager = NewManager(err)
|
||||
//r.StatusMon = NewStatusMonitor("Reactor", c.Id, sys)
|
||||
return r
|
||||
}
|
||||
|
||||
func (r *ReactorManager) Start() {
|
||||
r.Manager.Start()
|
||||
logging.Debug(logging.DStart,"RMA %v starting", r.Id)
|
||||
go r.StatusMon.Send(&DeviceInfo{Id:r.Id,Type:"Reactor",Status:"[green]ONLINE[white]"},"Reactor")
|
||||
//conn := r.Connect()
|
||||
//empty := &grpc.ClientConn{}
|
||||
//if conn != empty {
|
||||
//}
|
||||
r.Manager.Start()
|
||||
logging.Debug(logging.DStart, "RMA %v starting", r.Id)
|
||||
//go r.StatusMon.Send(&DeviceInfo{Id: r.Id, Type: "Reactor", Status: "[green]ONLINE[white]"}, "Reactor")
|
||||
//conn := r.Connect()
|
||||
//empty := &grpc.ClientConn{}
|
||||
//if conn != empty {
|
||||
//}
|
||||
}
|
||||
|
||||
func (r *ReactorManager) Exit() {
|
||||
r.Manager.Exit()
|
||||
logging.Debug(logging.DExit, "RMA %v exiting", r.Id)
|
||||
go r.StatusMon.Send(&DeviceInfo{Id:r.Id,Type:"Reactor",Status:"[red]OFFLINE[white]",Data:fmt.Sprintf("Last Seen %v",time.Now().Format("Mon at 03:04:05pm MST"))},"Reactor")
|
||||
r.devstatus.Lock()
|
||||
defer r.devstatus.Unlock()
|
||||
for _, d := range r.Devs {
|
||||
newd := d
|
||||
newd.Status = "[yellow]UNKOWN[white]"
|
||||
r.Devs[newd.Id] = newd
|
||||
go r.StatusMon.Send(newd,"Device")
|
||||
}
|
||||
r.Manager.Exit()
|
||||
logging.Debug(logging.DExit, "RMA %v exiting", r.Id)
|
||||
//go r.StatusMon.Send(&DeviceInfo{Id: r.Id, Type: "Reactor", Status: "[red]OFFLINE[white]", Data: fmt.Sprintf("Last Seen %v", time.Now().Format("Mon at 03:04:05pm MST"))}, "Reactor")
|
||||
r.devstatus.Lock()
|
||||
defer r.devstatus.Unlock()
|
||||
// keeping this because it **COULD** be useful, maybe
|
||||
for _, d := range r.Devs {
|
||||
newd := d
|
||||
newd.Status = "UNKOWN"
|
||||
r.Devs[newd.Id] = newd
|
||||
//go r.StatusMon.Send(newd, "Device")
|
||||
}
|
||||
}
|
||||
|
||||
func (r *ReactorManager) Connect() *grpc.ClientConn {
|
||||
// establish gRPC conection with reactor
|
||||
var opts []grpc.DialOption
|
||||
var conn *grpc.ClientConn
|
||||
opts = append(opts,grpc.WithTransportCredentials(insecure.NewCredentials()))
|
||||
// establish gRPC conection with reactor
|
||||
// this seems pretty stupid, seems like reactor should communicate up the chain to avoid unnessecary comms.
|
||||
var opts []grpc.DialOption
|
||||
var conn *grpc.ClientConn
|
||||
opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
|
||||
|
||||
for {
|
||||
if !r.IsActive() {
|
||||
logging.Debug(logging.DClient,"RMA %v No longer active, aborting connection attempt",r.Id)
|
||||
return &grpc.ClientConn{}
|
||||
}
|
||||
var err error
|
||||
conn, err = grpc.Dial(fmt.Sprintf("%v:%v",r.Ip,r.Port),opts...)
|
||||
// error handling
|
||||
code := status.Code(err)
|
||||
if code != 0 { // != OK
|
||||
if code == (5 | 14) { // unavailable or not found
|
||||
to := r.Timeout()
|
||||
if to == 0 {
|
||||
logging.Debug(logging.DClient,"RMA %v Client not responding",r.Id)
|
||||
return &grpc.ClientConn{}
|
||||
}
|
||||
logging.Debug(logging.DClient,"RMA %v Client currently down, retrying in %v ms",r.Id, to)
|
||||
time.Sleep(time.Duration(to) * time.Millisecond)
|
||||
for {
|
||||
if !r.IsActive() {
|
||||
logging.Debug(logging.DClient, "RMA %v No longer active, aborting connection attempt", r.Id)
|
||||
return &grpc.ClientConn{}
|
||||
}
|
||||
var err error
|
||||
conn, err = grpc.Dial(fmt.Sprintf("%v:%v", r.Ip, r.Port), opts...)
|
||||
// error handling
|
||||
code := status.Code(err)
|
||||
if code != 0 { // != OK
|
||||
if code == (5 | 14) { // unavailable or not found
|
||||
to := r.Timeout()
|
||||
if to == 0 {
|
||||
logging.Debug(logging.DClient, "RMA %v Client not responding", r.Id)
|
||||
return &grpc.ClientConn{}
|
||||
}
|
||||
logging.Debug(logging.DClient, "RMA %v Client currently down, retrying in %v ms", r.Id, to)
|
||||
time.Sleep(time.Duration(to) * time.Millisecond)
|
||||
|
||||
} else {
|
||||
logging.Debug(logging.DError,"RMA %v GRPC ERROR: %v",r.Id, code)
|
||||
r.Err <- err
|
||||
}
|
||||
}
|
||||
break;
|
||||
}
|
||||
return conn
|
||||
} else {
|
||||
logging.Debug(logging.DError, "RMA %v GRPC ERROR: %v", r.Id, code)
|
||||
r.Err <- err
|
||||
}
|
||||
}
|
||||
break
|
||||
}
|
||||
return conn
|
||||
}
|
||||
|
||||
func (r *ReactorManager) ReactorStatusHandler(ctx context.Context, req *pb.ReactorStatusPing) (*pb.ReactorStatusResponse, error) {
|
||||
// function client will call to update reactor information
|
||||
//go r.PingReset()
|
||||
for _, dev := range req.GetDevices() {
|
||||
d := &DeviceInfo{Id:uint32(dev.GetAddr()),Type:dev.GetType(),Status:dev.GetStatus(),Data:dev.GetData()}
|
||||
go r.UpdateDevice(d)
|
||||
}
|
||||
return &pb.ReactorStatusResponse{Id:r.Id}, nil
|
||||
// function client will call to update reactor information
|
||||
//go r.PingReset()
|
||||
for _, dev := range req.GetDevices() {
|
||||
d := &DeviceInfo{Id: int(dev.GetAddr()), Type: dev.GetType(), Status: dev.GetStatus(), Data: dev.GetData()}
|
||||
go r.UpdateDevice(d)
|
||||
}
|
||||
return &pb.ReactorStatusResponse{Id: uint32(r.Id)}, nil
|
||||
}
|
||||
|
||||
|
||||
/*
|
||||
func (r *ReactorManager) Monitor(conn *grpc.ClientConn) {
|
||||
defer conn.Close()
|
||||
client := pb.NewMonitoringClient(conn)
|
||||
for r.IsActive() {
|
||||
req := &pb.ReactorStatusRequest{Id:r.Id}
|
||||
resp, err := client.GetReactorStatus(context.Background(),req)
|
||||
code := status.Code(err)
|
||||
if code != 0 { // if != OK
|
||||
logging.Debug(logging.DClient,"RMA %v Reactor not responding! Code: %v\n", r.Id,code)
|
||||
r.devstatus.Lock()
|
||||
for _, d := range r.Devs {
|
||||
newd := d
|
||||
newd.Status = "[yellow]UNKOWN[white]"
|
||||
r.Devs[newd.Id] = newd
|
||||
go r.StatusMon.Send(newd,"Device")
|
||||
}
|
||||
r.devstatus.Unlock()
|
||||
r.Exit()
|
||||
break;
|
||||
}
|
||||
for _,v := range resp.GetDevices() {
|
||||
d := &DeviceInfo{Id:uint32(v.GetAddr()),Type:v.GetType(),Status:v.GetStatus(),Data:v.GetData()}
|
||||
go r.UpdateDevice(d)
|
||||
}
|
||||
time.Sleep(r.Hb) // time between sensor pings
|
||||
}
|
||||
}
|
||||
func (r *ReactorManager) Monitor(conn *grpc.ClientConn) {
|
||||
defer conn.Close()
|
||||
client := pb.NewMonitoringClient(conn)
|
||||
for r.IsActive() {
|
||||
req := &pb.ReactorStatusRequest{Id:r.Id}
|
||||
resp, err := client.GetReactorStatus(context.Background(),req)
|
||||
code := status.Code(err)
|
||||
if code != 0 { // if != OK
|
||||
logging.Debug(logging.DClient,"RMA %v Reactor not responding! Code: %v\n", r.Id,code)
|
||||
r.devstatus.Lock()
|
||||
for _, d := range r.Devs {
|
||||
newd := d
|
||||
newd.Status = "[yellow]UNKOWN[white]"
|
||||
r.Devs[newd.Id] = newd
|
||||
go r.StatusMon.Send(newd,"Device")
|
||||
}
|
||||
r.devstatus.Unlock()
|
||||
r.Exit()
|
||||
break;
|
||||
}
|
||||
for _,v := range resp.GetDevices() {
|
||||
d := &DeviceInfo{Id:uint32(v.GetAddr()),Type:v.GetType(),Status:v.GetStatus(),Data:v.GetData()}
|
||||
go r.UpdateDevice(d)
|
||||
}
|
||||
time.Sleep(r.Hb) // time between sensor pings
|
||||
}
|
||||
}
|
||||
*/
|
||||
|
||||
func (r *ReactorManager) UpdateDevice(d *DeviceInfo) {
|
||||
r.devstatus.Lock()
|
||||
defer r.devstatus.Unlock()
|
||||
if olddev, ok := r.Devs[d.Id]; !ok {
|
||||
// new device
|
||||
r.Devs[d.Id] = d
|
||||
go r.StatusMon.Send(d,"Device")
|
||||
} else if olddev.Status != d.Status || olddev.Data != d.Data {
|
||||
// dev status or data has changed
|
||||
r.Devs[d.Id] = d
|
||||
go r.StatusMon.Send(d,"Device")
|
||||
}
|
||||
r.devstatus.Lock()
|
||||
defer r.devstatus.Unlock()
|
||||
if olddev, ok := r.Devs[d.Id]; !ok {
|
||||
// new device
|
||||
r.Devs[d.Id] = d
|
||||
//go r.StatusMon.Send(d, "Device")
|
||||
} else if olddev.Status != d.Status || olddev.Data != d.Data {
|
||||
// dev status or data has changed
|
||||
r.Devs[d.Id] = d
|
||||
//go r.StatusMon.Send(d, "Device")
|
||||
}
|
||||
}
|
||||
|
||||
@ -1,315 +1,318 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"sync"
|
||||
_ "fmt"
|
||||
"FRMS/internal/pkg/logging"
|
||||
_ "fmt"
|
||||
)
|
||||
|
||||
// allows for multiple readers/writers
|
||||
type DeviceInfo struct {
|
||||
Id uint32
|
||||
Type string
|
||||
Status string
|
||||
Data string
|
||||
Index uint32
|
||||
TransactionId uint32
|
||||
Id int
|
||||
Type string
|
||||
Status string
|
||||
Data string
|
||||
Index int
|
||||
TransactionId uint32
|
||||
}
|
||||
|
||||
/*
|
||||
|
||||
type StatusMonitor struct {
|
||||
// allows for embedding into managers
|
||||
TransactionId chan uint32 // monotonically increases to track outdated reqs
|
||||
DevChan chan *DeviceInfo // channel for device status
|
||||
ReactorChan chan *DeviceInfo // channel for reactor status
|
||||
*SystemViewer
|
||||
DevBuf *devbuf
|
||||
sync.Mutex
|
||||
// allows for embedding into managers
|
||||
TransactionId chan uint32 // monotonically increases to track outdated reqs
|
||||
DevChan chan *DeviceInfo // channel for device status
|
||||
ReactorChan chan *DeviceInfo // channel for reactor status
|
||||
*SystemViewer
|
||||
DevBuf *devbuf
|
||||
sync.Mutex
|
||||
}
|
||||
|
||||
type devbuf struct {
|
||||
ReactorId uint32 // reactor we are looking at, if any
|
||||
Buf map[string]map[uint32]*DeviceInfo // convienent way to store/seperate device data
|
||||
sync.Mutex
|
||||
}
|
||||
|
||||
func NewBuffer() map[string]map[uint32]*DeviceInfo {
|
||||
rbuf := make(map[uint32]*DeviceInfo)
|
||||
dbuf := make(map[uint32]*DeviceInfo)
|
||||
sbuf := make(map[string]map[uint32]*DeviceInfo)
|
||||
sbuf["Reactor"] = rbuf
|
||||
sbuf["Device"] = dbuf
|
||||
return sbuf
|
||||
}
|
||||
|
||||
func NewStatusMonitor(t string, id uint32, sys *SystemViewer) *StatusMonitor {
|
||||
tid := make(chan uint32)
|
||||
sm := &StatusMonitor{TransactionId:tid}
|
||||
sm.SystemViewer = sys
|
||||
logging.Debug(logging.DClient,"SYS Creating new status monitor")
|
||||
if t == "Reactor" {
|
||||
// reactor status monitor
|
||||
sm.ReactorChan = sys.AddReactorSender()
|
||||
sm.DevChan = sys.AddDeviceSender(id)
|
||||
go sm.GenerateIds()
|
||||
} else {
|
||||
// tui status monitor
|
||||
sbuf := NewBuffer()
|
||||
//sm.ReactorChan, sbuf["Reactor"] = sys.AddListener(id,0)
|
||||
sm.DevBuf = &devbuf{Buf:sbuf} // makes it easier to work with
|
||||
go sm.UpdateListener(id,0)
|
||||
}
|
||||
return sm
|
||||
ReactorId int // reactor we are looking at, if any
|
||||
Buf map[string]map[int]*DeviceInfo // convienent way to store/seperate device data
|
||||
sync.Mutex
|
||||
}
|
||||
|
||||
func NewBuffer() map[string]map[int]*DeviceInfo {
|
||||
rbuf := make(map[int]*DeviceInfo)
|
||||
dbuf := make(map[int]*DeviceInfo)
|
||||
sbuf := make(map[string]map[int]*DeviceInfo)
|
||||
sbuf["Reactor"] = rbuf
|
||||
sbuf["Device"] = dbuf
|
||||
return sbuf
|
||||
}
|
||||
|
||||
func NewStatusMonitor(t string, id int, sys *SystemViewer) *StatusMonitor {
|
||||
tid := make(chan uint32)
|
||||
sm := &StatusMonitor{TransactionId: tid}
|
||||
sm.SystemViewer = sys
|
||||
logging.Debug(logging.DClient, "SYS Creating new status monitor")
|
||||
if t == "Reactor" {
|
||||
// reactor status monitor
|
||||
sm.ReactorChan = sys.AddReactorSender()
|
||||
sm.DevChan = sys.AddDeviceSender(id)
|
||||
go sm.GenerateIds()
|
||||
} else {
|
||||
// tui status monitor
|
||||
sbuf := NewBuffer()
|
||||
//sm.ReactorChan, sbuf["Reactor"] = sys.AddListener(id,0)
|
||||
sm.DevBuf = &devbuf{Buf: sbuf} // makes it easier to work with
|
||||
go sm.UpdateListener(id, 0)
|
||||
}
|
||||
return sm
|
||||
}
|
||||
|
||||
func (s *StatusMonitor) GenerateIds() {
|
||||
var id uint32
|
||||
id = 0
|
||||
for {
|
||||
s.TransactionId <-id
|
||||
id += 1
|
||||
}
|
||||
var id uint32
|
||||
id = 0
|
||||
for {
|
||||
s.TransactionId <- id
|
||||
id += 1
|
||||
}
|
||||
}
|
||||
|
||||
func (s *StatusMonitor) Send(d *DeviceInfo, dtype string) {
|
||||
d.TransactionId = <-s.TransactionId
|
||||
logging.Debug(logging.DClient,"SYS 01 Sending update for: %s (%s)", d.Type, d.Status)
|
||||
if dtype == "Reactor" {
|
||||
s.ReactorChan <-d
|
||||
} else {
|
||||
s.DevChan <-d
|
||||
}
|
||||
d.TransactionId = <-s.TransactionId
|
||||
logging.Debug(logging.DClient, "SYS 01 Sending update for: %s (%s)", d.Type, d.Status)
|
||||
if dtype == "Reactor" {
|
||||
s.ReactorChan <- d
|
||||
} else {
|
||||
s.DevChan <- d
|
||||
}
|
||||
}
|
||||
|
||||
func (s *StatusMonitor) GetBuffer() []*DeviceInfo {
|
||||
// also clears buffer
|
||||
s.DevBuf.Lock()
|
||||
defer s.DevBuf.Unlock()
|
||||
res := []*DeviceInfo{}
|
||||
if len(s.DevBuf.Buf["Reactor"]) != 0 || len(s.DevBuf.Buf["Device"]) != 0 {
|
||||
logging.Debug(logging.DClient,"Clearing buff %v", s.DevBuf.Buf)
|
||||
}
|
||||
for _, devs := range s.DevBuf.Buf {
|
||||
for _, dev := range devs {
|
||||
// loops over reactors then devices
|
||||
res = append(res,dev)
|
||||
}
|
||||
}
|
||||
s.DevBuf.Buf = NewBuffer() // clearing old buffer
|
||||
return res
|
||||
// also clears buffer
|
||||
s.DevBuf.Lock()
|
||||
defer s.DevBuf.Unlock()
|
||||
res := []*DeviceInfo{}
|
||||
if len(s.DevBuf.Buf["Reactor"]) != 0 || len(s.DevBuf.Buf["Device"]) != 0 {
|
||||
logging.Debug(logging.DClient, "Clearing buff %v", s.DevBuf.Buf)
|
||||
}
|
||||
for _, devs := range s.DevBuf.Buf {
|
||||
for _, dev := range devs {
|
||||
// loops over reactors then devices
|
||||
res = append(res, dev)
|
||||
}
|
||||
}
|
||||
s.DevBuf.Buf = NewBuffer() // clearing old buffer
|
||||
return res
|
||||
}
|
||||
|
||||
func (s *StatusMonitor) UpdateListener(tid, reactorId uint32) {
|
||||
s.DevBuf.Lock()
|
||||
defer s.DevBuf.Unlock()
|
||||
// clearing proper buffer
|
||||
if reactorId == 0 {
|
||||
logging.Debug(logging.DClient,"SYS 01 Adding %v as reactor listener", tid)
|
||||
s.DevBuf.Buf["Reactor"] = make(map[uint32]*DeviceInfo)
|
||||
s.ReactorChan, s.DevBuf.Buf["Reactor"] = s.SystemViewer.AddListener(tid, reactorId)
|
||||
go s.Listen(s.ReactorChan)
|
||||
} else {
|
||||
logging.Debug(logging.DClient,"SYS 01 Adding %v as reactor %v listener", tid, reactorId)
|
||||
s.DevBuf.Buf["Device"] = make(map[uint32]*DeviceInfo) // clearing old devices
|
||||
if s.DevBuf.ReactorId != reactorId && s.DevBuf.ReactorId != 0{
|
||||
go s.SystemViewer.RemoveListener(s.DevBuf.ReactorId, tid)
|
||||
}
|
||||
s.DevBuf.ReactorId = reactorId
|
||||
s.DevChan, s.DevBuf.Buf["Device"] = s.SystemViewer.AddListener(tid, reactorId)
|
||||
go s.Listen(s.DevChan)
|
||||
}
|
||||
s.DevBuf.Lock()
|
||||
defer s.DevBuf.Unlock()
|
||||
// clearing proper buffer
|
||||
if reactorId == 0 {
|
||||
logging.Debug(logging.DClient, "SYS 01 Adding %v as reactor listener", tid)
|
||||
s.DevBuf.Buf["Reactor"] = make(map[uint32]*DeviceInfo)
|
||||
s.ReactorChan, s.DevBuf.Buf["Reactor"] = s.SystemViewer.AddListener(tid, reactorId)
|
||||
go s.Listen(s.ReactorChan)
|
||||
} else {
|
||||
logging.Debug(logging.DClient, "SYS 01 Adding %v as reactor %v listener", tid, reactorId)
|
||||
s.DevBuf.Buf["Device"] = make(map[uint32]*DeviceInfo) // clearing old devices
|
||||
if s.DevBuf.ReactorId != reactorId && s.DevBuf.ReactorId != 0 {
|
||||
go s.SystemViewer.RemoveListener(s.DevBuf.ReactorId, tid)
|
||||
}
|
||||
s.DevBuf.ReactorId = reactorId
|
||||
s.DevChan, s.DevBuf.Buf["Device"] = s.SystemViewer.AddListener(tid, reactorId)
|
||||
go s.Listen(s.DevChan)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *StatusMonitor) UpdateBuffer(d *DeviceInfo, dtype string, ch chan *DeviceInfo) {
|
||||
s.DevBuf.Lock()
|
||||
defer s.DevBuf.Unlock()
|
||||
logging.Debug(logging.DClient,"SYS 01 Dev %v update requested", d)
|
||||
if dev, exists := s.DevBuf.Buf[dtype][d.Id]; exists {
|
||||
// already a device in the buffer
|
||||
if dev.TransactionId > d.TransactionId {
|
||||
logging.Debug(logging.DClient,"SYS 01 Update Processed. Old: %v, New: %v \n", dev, d)
|
||||
d = dev // not sure if i can do this lol
|
||||
}
|
||||
}
|
||||
if ch == s.ReactorChan || ch == s.DevChan {
|
||||
// hacky way to check if the device came from a listener of a current channel
|
||||
s.DevBuf.Buf[dtype][d.Id] = d
|
||||
} else {
|
||||
logging.Debug(logging.DClient,"Dev out of date!")
|
||||
}
|
||||
s.DevBuf.Lock()
|
||||
defer s.DevBuf.Unlock()
|
||||
logging.Debug(logging.DClient, "SYS 01 Dev %v update requested", d)
|
||||
if dev, exists := s.DevBuf.Buf[dtype][d.Id]; exists {
|
||||
// already a device in the buffer
|
||||
if dev.TransactionId > d.TransactionId {
|
||||
logging.Debug(logging.DClient, "SYS 01 Update Processed. Old: %v, New: %v \n", dev, d)
|
||||
d = dev // not sure if i can do this lol
|
||||
}
|
||||
}
|
||||
if ch == s.ReactorChan || ch == s.DevChan {
|
||||
// hacky way to check if the device came from a listener of a current channel
|
||||
s.DevBuf.Buf[dtype][d.Id] = d
|
||||
} else {
|
||||
logging.Debug(logging.DClient, "Dev out of date!")
|
||||
}
|
||||
}
|
||||
|
||||
func (s *StatusMonitor) Listen(ch chan *DeviceInfo) {
|
||||
for dev := range ch {
|
||||
if dev.Type == "Reactor" {
|
||||
go s.UpdateBuffer(dev,"Reactor", ch)
|
||||
} else {
|
||||
go s.UpdateBuffer(dev, "Device", ch)
|
||||
}
|
||||
}
|
||||
for dev := range ch {
|
||||
if dev.Type == "Reactor" {
|
||||
go s.UpdateBuffer(dev, "Reactor", ch)
|
||||
} else {
|
||||
go s.UpdateBuffer(dev, "Device", ch)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type InfoStream struct {
|
||||
// base stream for any reactor/device
|
||||
// NewSender embedds the channel to send updates on
|
||||
// NewListener will add the statusmon to the list of devs to echo to
|
||||
Stream chan *DeviceInfo
|
||||
Layout *syslayout
|
||||
*listeners
|
||||
// base stream for any reactor/device
|
||||
// NewSender embedds the channel to send updates on
|
||||
// NewListener will add the statusmon to the list of devs to echo to
|
||||
Stream chan *DeviceInfo
|
||||
Layout *syslayout
|
||||
*listeners
|
||||
}
|
||||
|
||||
type listeners struct {
|
||||
sync.RWMutex
|
||||
Listeners map[uint32]*lischan
|
||||
sync.RWMutex
|
||||
Listeners map[uint32]*lischan
|
||||
}
|
||||
|
||||
type lischan struct {
|
||||
sync.WaitGroup
|
||||
StatusChan chan *DeviceInfo
|
||||
sync.WaitGroup
|
||||
StatusChan chan *DeviceInfo
|
||||
}
|
||||
|
||||
type syslayout struct {
|
||||
Devs map[uint32]*DeviceInfo
|
||||
uint32 //index
|
||||
sync.RWMutex
|
||||
Devs map[uint32]*DeviceInfo
|
||||
uint32 //index
|
||||
sync.RWMutex
|
||||
}
|
||||
|
||||
func NewLisChan(ch chan *DeviceInfo) *lischan {
|
||||
l := &lischan{StatusChan:ch}
|
||||
return l
|
||||
l := &lischan{StatusChan: ch}
|
||||
return l
|
||||
}
|
||||
|
||||
func NewInfoStream() *InfoStream {
|
||||
dch := make(chan *DeviceInfo)
|
||||
s := &InfoStream{Stream:dch}
|
||||
m := make(map[uint32]*DeviceInfo)
|
||||
s.Layout = &syslayout{Devs:m}
|
||||
s.listeners = &listeners{Listeners:make(map[uint32]*lischan)}
|
||||
return s
|
||||
dch := make(chan *DeviceInfo)
|
||||
s := &InfoStream{Stream: dch}
|
||||
m := make(map[uint32]*DeviceInfo)
|
||||
s.Layout = &syslayout{Devs: m}
|
||||
s.listeners = &listeners{Listeners: make(map[uint32]*lischan)}
|
||||
return s
|
||||
}
|
||||
|
||||
func (s *InfoStream) Start() {
|
||||
// consistency
|
||||
go s.Listen()
|
||||
// consistency
|
||||
go s.Listen()
|
||||
}
|
||||
|
||||
// goal is to hook every new manager into the reactor status chan
|
||||
func (s *InfoStream) AddSender() chan *DeviceInfo {
|
||||
return s.Stream
|
||||
return s.Stream
|
||||
}
|
||||
|
||||
func (s *InfoStream) Listen() {
|
||||
for deviceInfo := range s.Stream {
|
||||
go s.Update(deviceInfo)
|
||||
}
|
||||
for deviceInfo := range s.Stream {
|
||||
go s.Update(deviceInfo)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *InfoStream) Update(d *DeviceInfo) {
|
||||
s.Layout.Lock()
|
||||
defer s.Layout.Unlock()
|
||||
if dev, ok := s.Layout.Devs[d.Id]; !ok {
|
||||
d.Index = s.Layout.uint32
|
||||
s.Layout.uint32 += 1
|
||||
} else {
|
||||
d.Index = dev.Index
|
||||
}
|
||||
s.Layout.Devs[d.Id] = d
|
||||
go s.Echo(d)
|
||||
s.Layout.Lock()
|
||||
defer s.Layout.Unlock()
|
||||
if dev, ok := s.Layout.Devs[d.Id]; !ok {
|
||||
d.Index = s.Layout.uint32
|
||||
s.Layout.uint32 += 1
|
||||
} else {
|
||||
d.Index = dev.Index
|
||||
}
|
||||
s.Layout.Devs[d.Id] = d
|
||||
go s.Echo(d)
|
||||
}
|
||||
|
||||
func (l *listeners) Echo(d *DeviceInfo) {
|
||||
l.RLock()
|
||||
defer l.RUnlock()
|
||||
// read only lock
|
||||
for _, lis := range l.Listeners {
|
||||
lis.Add(1)
|
||||
go func(listener *lischan, dev *DeviceInfo){
|
||||
defer listener.Done()
|
||||
listener.StatusChan <-dev
|
||||
}(lis,d)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *InfoStream) AddListener(id uint32, ch chan *DeviceInfo) map[uint32]*DeviceInfo {
|
||||
// if i get a memory leak ill eat my shoe
|
||||
s.listeners.Lock()
|
||||
defer s.listeners.Unlock()
|
||||
if _, ok := s.listeners.Listeners[id]; ok {
|
||||
// exists
|
||||
go s.RemoveListener(id)
|
||||
}
|
||||
s.listeners.Listeners[id] = NewLisChan(ch)
|
||||
logging.Debug(logging.DClient,"SYS 01 Getting Devices")
|
||||
//s.Layout.RLock()
|
||||
//defer s.Layout.RUnlock()
|
||||
logging.Debug(logging.DClient,"SYS 01 Returning Devices %v", s.Layout.Devs)
|
||||
return s.Layout.Devs
|
||||
}
|
||||
|
||||
func (l *listeners) RemoveListener(id uint32) {
|
||||
l.Lock()
|
||||
defer l.Unlock()
|
||||
if lis, ok := l.Listeners[id]; ok {
|
||||
delete(l.Listeners,id)
|
||||
go func(ls *lischan){
|
||||
ls.Wait()
|
||||
close(ls.StatusChan)
|
||||
}(lis)
|
||||
}
|
||||
l.RLock()
|
||||
defer l.RUnlock()
|
||||
// read only lock
|
||||
for _, lis := range l.Listeners {
|
||||
lis.Add(1)
|
||||
go func(listener *lischan, dev *DeviceInfo) {
|
||||
defer listener.Done()
|
||||
listener.StatusChan <- dev
|
||||
}(lis, d)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *InfoStream) AddListener(id int, ch chan *DeviceInfo) map[uint32]*DeviceInfo {
|
||||
// if i get a memory leak ill eat my shoe
|
||||
s.listeners.Lock()
|
||||
defer s.listeners.Unlock()
|
||||
if _, ok := s.listeners.Listeners[id]; ok {
|
||||
// exists
|
||||
go s.RemoveListener(id)
|
||||
}
|
||||
s.listeners.Listeners[id] = NewLisChan(ch)
|
||||
logging.Debug(logging.DClient, "SYS 01 Getting Devices")
|
||||
//s.Layout.RLock()
|
||||
//defer s.Layout.RUnlock()
|
||||
logging.Debug(logging.DClient, "SYS 01 Returning Devices %v", s.Layout.Devs)
|
||||
return s.Layout.Devs
|
||||
}
|
||||
|
||||
func (l *listeners) RemoveListener(id int) {
|
||||
l.Lock()
|
||||
defer l.Unlock()
|
||||
if lis, ok := l.Listeners[id]; ok {
|
||||
delete(l.Listeners, id)
|
||||
go func(ls *lischan) {
|
||||
ls.Wait()
|
||||
close(ls.StatusChan)
|
||||
}(lis)
|
||||
}
|
||||
}
|
||||
|
||||
// status buffer maintaince
|
||||
|
||||
type SystemViewer struct {
|
||||
// stores system directory and provide methods to be embedded in managers
|
||||
ReactorStream *InfoStream // can add itself as a listener to provide methods
|
||||
DeviceStream *ds
|
||||
// stores system directory and provide methods to be embedded in managers
|
||||
ReactorStream *InfoStream // can add itself as a listener to provide methods
|
||||
DeviceStream *ds
|
||||
}
|
||||
|
||||
type ds struct {
|
||||
Reactors map[uint32]*InfoStream //map from reactor id to its device info stream
|
||||
sync.Mutex
|
||||
Reactors map[uint32]*InfoStream //map from reactor id to its device info stream
|
||||
sync.Mutex
|
||||
}
|
||||
|
||||
func NewSystemViewer() *SystemViewer {
|
||||
rs := NewInfoStream()
|
||||
s := &SystemViewer{ReactorStream:rs}
|
||||
m := make(map[uint32]*InfoStream)
|
||||
s.DeviceStream = &ds{Reactors:m}
|
||||
return s
|
||||
rs := NewInfoStream()
|
||||
s := &SystemViewer{ReactorStream: rs}
|
||||
m := make(map[uint32]*InfoStream)
|
||||
s.DeviceStream = &ds{Reactors: m}
|
||||
return s
|
||||
}
|
||||
|
||||
func (s *SystemViewer) Start() {
|
||||
go s.ReactorStream.Start()
|
||||
go s.ReactorStream.Start()
|
||||
}
|
||||
|
||||
func (s *SystemViewer) AddReactorSender() chan *DeviceInfo {
|
||||
return s.ReactorStream.AddSender()
|
||||
return s.ReactorStream.AddSender()
|
||||
}
|
||||
|
||||
func (s *SystemViewer) AddDeviceSender(reactorId uint32) chan *DeviceInfo {
|
||||
s.DeviceStream.Lock()
|
||||
defer s.DeviceStream.Unlock()
|
||||
var ds *InfoStream
|
||||
var ok bool
|
||||
if ds, ok = s.DeviceStream.Reactors[reactorId]; !ok {
|
||||
ds = NewInfoStream()
|
||||
s.DeviceStream.Reactors[reactorId] = ds
|
||||
go ds.Start()
|
||||
}
|
||||
return ds.AddSender()
|
||||
}
|
||||
|
||||
func (s *SystemViewer) AddListener(id, rid uint32) (chan *DeviceInfo, map[uint32]*DeviceInfo) {
|
||||
// returns a listener for that chan
|
||||
ch := make(chan *DeviceInfo)
|
||||
if rid != 0 {
|
||||
return ch, s.DeviceStream.Reactors[rid].AddListener(id, ch)
|
||||
} else {
|
||||
return ch, s.ReactorStream.AddListener(id, ch)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *SystemViewer) RemoveListener(rid, tid uint32) {
|
||||
// removes chan for specific tid and rid
|
||||
s.DeviceStream.Lock()
|
||||
defer s.DeviceStream.Unlock()
|
||||
go s.DeviceStream.Reactors[rid].RemoveListener(tid)
|
||||
}
|
||||
s.DeviceStream.Lock()
|
||||
defer s.DeviceStream.Unlock()
|
||||
var ds *InfoStream
|
||||
var ok bool
|
||||
if ds, ok = s.DeviceStream.Reactors[reactorId]; !ok {
|
||||
ds = NewInfoStream()
|
||||
s.DeviceStream.Reactors[reactorId] = ds
|
||||
go ds.Start()
|
||||
}
|
||||
return ds.AddSender()
|
||||
}
|
||||
|
||||
func (s *SystemViewer) AddListener(id, rid int) (chan *DeviceInfo, map[uint32]*DeviceInfo) {
|
||||
// returns a listener for that chan
|
||||
ch := make(chan *DeviceInfo)
|
||||
if rid != 0 {
|
||||
return ch, s.DeviceStream.Reactors[rid].AddListener(id, ch)
|
||||
} else {
|
||||
return ch, s.ReactorStream.AddListener(id, ch)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *SystemViewer) RemoveListener(rid, tid int) {
|
||||
// removes chan for specific tid and rid
|
||||
s.DeviceStream.Lock()
|
||||
defer s.DeviceStream.Unlock()
|
||||
go s.DeviceStream.Reactors[rid].RemoveListener(tid)
|
||||
}
|
||||
*/
|
||||
|
||||
Loading…
Reference in New Issue