DMAC/internal/pkg/server/system.go

package server

import (
    "sync"
    _ "fmt"
    "FRMS/internal/pkg/logging"
)

// package will create and maintain a concurrent system structure
// allows for multiple readers/writers
type DeviceInfo struct {
    Id uint32
    Type string
    Status string
    Data string
    Index uint32
    TransactionId uint32
}

func NewStatusMonitor(ch chan *DeviceInfo) *StatusMonitor {
    tid := make(chan uint32)
    sm := &StatusMonitor{StatusChan:ch,TransactionId:tid}
    return sm
}

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
}

type listeners struct {
    sync.RWMutex
    Listeners map[uint32]*lischan
}

type lischan struct {
    sync.WaitGroup
    *StatusMonitor
}

type syslayout struct {
    Devs map[uint32]*DeviceInfo
    uint32 //index
    sync.RWMutex
}

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
}

func (s *InfoStream) Start() {
    // consistency
    go s.Listener()
}
// goal is to hook every new manager into the reactor status chan
func (s *InfoStream) AddSender() *StatusMonitor {
    sm := NewStatusMonitor(s.Stream) // give the channel we monitor for incoming updates
    return sm
}

func (s *InfoStream) Listener() {
    for {
        deviceInfo := <-s.Stream
        go s.Update(deviceInfo)
        go s.Echo(deviceInfo)
    }
}

func (s *InfoStream) Update(d *DeviceInfo) {
    s.Layout.Lock()
    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 *listener) 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, *StatusMonitor) {
    s.listener.Lock()
    s.Layout.Lock()
    defer s.listner.Unlock()
    defer s.Layout.Unlock()
    if sm, ok l.listener.Listeners[id]; ok {
        // listener already exists return nil
        // going to delete and create a new one here just because f it
        
        return s.Layout.Devs, sm
    } else {
        ch := make(chan *DeviceInfo)
        sm := NewStatusMonitor(ch)
        s.listener.Listeners[id] = sm
        return s.Layout.Devs, sm
    }
}

func (l *listener) RemoveListener(id uint32) error {
    l.Lock()
    defer l.Unlock()
    if lis, ok := l.Listeners[id]; !ok {
        return errors.New("Listener doesn't exists!")
    } else {
        lis.Wait()
        lis.StatusChan.Close()
        delete(l.Listeners,id)
    }
    return nil
}
//func (s *InfoStream) GetLayout() map[uint32]*DeviceInfo {
    //s.Layout.RLock()
    //defer s.Layout.RUnlock()
    //return s.Layout.Devs
//}

// status buffer maintaince
type StatusMonitor struct {
    // serve as base to embed in managers to send/receive device info
    TransactionId chan uint32 // monotonically increases to track outdated reqs
    StatusChan chan *DeviceInfo // sending reactor info in same fmt
    Buffer *buf
}

type buf struct {
    Buffer map[uint32]*DeviceInfo //directory of changes since last req
    sync.Mutex
}

func (s *StatusMonitor) Start() {
    go s.GenerateIds()
}

func (s *StatusMonitor) GenerateIds() {
    var id uint32
    id = 0
    for {
        s.TransactionId <-id
        id += 1
    }
}

func (s *StatusMonitor) Send(d *DeviceInfo) {
    d.TransactionId = <-s.TransactionId
    logging.Debug(logging.DClient,"SYS 01 Updating %s Info (%s, %s)", d.Type, d.Status, d.Data)
    s.StatusChan <-d
}

// now to tie it all together

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
}

type ds struct {
    Reactors map[uint32]*InfoStream //map from reactor id to its device info stream
    Clients map[uint32]uint32 // maps tui ids to checked out reactor dev ids
    sync.Mutex
}

func NewSystemViewer() *SystemViewer {
    rs := NewInfoStream()
    s := &SystemViewer{ReactorStream:rs}
    m := make(map[uint32]*InfoStream)
    c := make(map[uint32]uint32)
    s.DeviceStream =  &ds{Reactors:m,Clients:c}
    return s
}

func (s *SystemViewer) Start() {
    go s.ReactorStream.Start()
}

func (s *SystemViewer) AddReactorSender() *StatusMonitor {
    return s.ReactorStream.AddSender()
}

func (s *SystemViewer) AddDeviceSender(reactorId uint32) *StatusMonitor {
    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) AddReactorListener(tid uint32) (map[uint32]*DeviceInfo, *StatusMonitor) {
    // adds status monitor as a listener and returns any reactors loaded before our channel is active
    // id serves as client id and limits them to one of each
    rstatus, ch := s.ReactorStream.AddListener(id)
    if sm

func (s *SystemViewer) AddDeviceListener(rid, tid uint32) (map[uint32]*DeviceInfo, *StatusMonitor) {
    // adds status monitor as a listener and returns any reactors loaded before our channel is active
    // id serves as client id and limits them to one of each
    rstatus, sm := s.ReactorStream.AddListener(id)
    if sm


/*
func (s *SystemViewer) GetReactorStatus() map[uint32]DeviceInfo {
    devs := s.ReactorStream.GetLayout()
    ret := make(map[uint32]DeviceInfo)
    for id, dev := range devs {
        ret[id] = *dev
    }
    return ret //value instead of ptr
}

func (s *SystemViewer) GetDeviceStatus(reactorId uint32) map[uint32]DeviceInfo {
    s.DeviceStream.Lock()
    ds := s.DeviceStream.Reactors[reactorId]
    s.DeviceStream.Unlock()
    devs := ds.GetLayout()
    ret := make(map[uint32]DeviceInfo)
    for id, dev := range devs {
        ret[id] = *dev
    }
    return ret
}
*/