DMAC/internal/pkg/server/reactormanager.go

package server

import (
	"FRMS/internal/pkg/device"
	pb "FRMS/internal/pkg/grpc"
	"FRMS/internal/pkg/logging"
	"FRMS/internal/pkg/manager"
	"time"

	//"FRMS/internal/pkg/device"
	"context"
	"fmt"
	_ "log"
	"sync"

	"github.com/spf13/viper"
)

// this package will implement a reactor manager and associated go routines

type Manager interface {
	Start() error // status checks
	Exit() error
	Timeout() (time.Duration, error) // TO Generator
}

func NewManager(max int) Manager {
	// takes a heartbeat and max connection attempts
	return manager.New(max)
}

type ReactorManager struct {
	Manager        // base manager interface
	*ClientManager // client manager (OUTDATED)
	//	StatusMon *StatusMonitor putting on pause
	*ReactorDevices
	Config *viper.Viper // config to update
	Err    chan error
}

type ReactorDevices struct {
	// device struct
	Devices map[int]DeviceManager
	sync.RWMutex
}

func NewReactorManager(cl *Client, config *viper.Viper, errCh chan error) *ReactorManager {
	// making managers
	m := NewManager(6)
	dm := make(map[int]DeviceManager)
	rd := &ReactorDevices{Devices: dm}
	cm := NewClientManager(cl)
	r := &ReactorManager{
		Manager:        m,
		ClientManager:  cm,
		ReactorDevices: rd,
		Config:         config,
		Err:            errCh,
	}
	return r
}

func (r *ReactorManager) Start() error {
	// allows for extra stuff
	logging.Debug(logging.DStart, "RMA %v starting", r.Id)
	return r.Manager.Start()
	//go r.StatusMon.Send(&DeviceInfo{Id: r.Id, Type: "Reactor", Status: "[green]ONLINE[white]"}, "Reactor")
}

func (r *ReactorManager) Exit() error {
	// allows for extra stuff
	logging.Debug(logging.DExit, "RMA %v exiting", r.Id)
	return r.Manager.Exit()
	//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")
}

func (r *ReactorManager) ReactorStatusHandler(ctx context.Context, req *pb.ReactorStatusPing) (*pb.ReactorStatusResponse, error) {
	// function client will call to update reactor information
	//go r.PingReset()
	fmt.Printf("Recieved ping from %d!\n", req.GetId())
	// update devices/sensors
	go r.UpdateDevices(req.GetDevices())

	return &pb.ReactorStatusResponse{Id: int32(r.Id)}, nil
}

// device stuff

type DeviceManager interface {
	LoadConfig(string) error
	UpdateDevice(*pb.Device) error
	String() string // printing
}

func NewDeviceManager(dev *pb.Device, config *viper.Viper) (DeviceManager, error) {
	// returns a manager struct
	return device.NewDeviceManager(dev, config)
}

func (r *ReactorManager) UpdateDevices(devs []*pb.Device) {
	// pass updates to correct manager
	r.ReactorDevices.RLock() // read lock only
	defer r.ReactorDevices.RUnlock()

	for _, dev := range devs {
		// looping over devs
		if dm, ok := r.ReactorDevices.Devices[int(dev.GetAddr())]; ok {
			// device manager found
			go dm.UpdateDevice(dev)
			//fmt.Println(dm)
		} else {
			// not found
			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) {
	r.Lock() // write lock
	defer r.Unlock()
	dm, err := NewDeviceManager(dev, config)
	if err != nil {
		errCh <- err
	}
	if err := dm.LoadConfig(fmt.Sprintf("reactors.%d", id)); err != nil {
		errCh <- err
	}
	r.Devices[int(dev.GetAddr())] = dm
}