DMAC/internal/pkg/reactor/rlcoordinator.go

package reactor

// file describes reactor level coordinator and associated implementation

import (
    "fmt"
    "sync"
    "net"
    "FRMS/internal/pkg/system"
    "FRMS/internal/pkg/I2C"
    "FRMS/internal/pkg/sensor"
    "errors"
    "context"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
    pb "FRMS/internal/pkg/grpc"
)

type Hardware struct {
    Id uint32
    Ip string
    Bus int
    Model string
    Port int
}

type HWinfo interface {
    Get() (*Hardware, error)
}

type Coordinator struct {
    Server string
    HW Hardware
    Connected <-chan int
    Disconnected <-chan int
    Sensors map[int]SensorManager
    Err chan error
    mu sync.Mutex
    pb.UnimplementedMonitoringServer
}

type SensorManager interface {
    Start()
    GetStatus() bool
    GetType() string
}

func NewSensorManager(addr int) SensorManager {
    return sensor.NewSensorManager(addr)
}

func NewCoordinator(s string,ch chan error) *Coordinator {
    c := make(Coordinator{Server:s,Err:ch}
    c.Connected = make(<-chan int)
    c.Disconnected = make(<-chan int)
    return c
}

func NewI2CMonitor(c, dc <-chan int, b int) error {
    return I2C.NewMonitor(b)
}

func GetHWInfo() Hwinfo {
    return system.NewHWMonitor()
}

func (c *Coordinator) Start() error {
    // should discover hwinfo and sensors on its own
    // now setting up sensor managers
    c.HW, err = GetHWInfo()
    if err != nil {
        return err
    }
    err = NewI2CMonitor(c.Connect,c.Disconnect,c.Bus)
    if err != nil {
        return err
    }
    fmt.Printf("Reactor at IP addr %v using I2C bus %v\n",c.Ip,c.Bus)
    response := c.EstablishConnection()
    for response != nil {
        fmt.Println("Connection failed!, Retrying")
        response = c.EstablishConnection()
    }
    c.Register()
    return nil
}

func (c *Coordinator) Monitor() {
    // function to automatically create and destroy sm
    for {
        select {
        case addr := <-c.Connect:
            //check if sm exists and add sm for addr
            c.Connected(addr)
        case addr := <-c.Disconnect:
            // kill sm for addr
            c.Disconnected(addr)
        }
    }
}

func (c *Coordinator) Connected(addr int) {
    c.mu.Lock()
    defer c.mu.Unlock()
    sm, exists := c.Sensors[addr]
    if !exists {
        sm := NewSensorManager(addr)
        c.Sensors[addr] = sm
        go sm.Start()
    } // ignoring case of existing sm
}

func (c *Coordinator) Disconnected(addr int) {
    c.mu.Lock()
    defer c.mu.Unlock()
    sm, exists := c.Sensors[addr]
    if exists {
        c.Sensors = delete(c.Sensors,addr)
        sm.Kill()
    } // not dealing with kill requests for not existing sm

func (c *Coordinator) EstablishConnection() error {
    // function connects to central server and passes hwinfo
    var opts []grpc.DialOption
    opts = append(opts,grpc.WithTransportCredentials(insecure.NewCredentials()))
    conn, err := grpc.Dial(c.Server,opts...)
    if err != nil {
        return err
    }
    defer conn.Close()
    client := pb.NewHandshakeClient(conn)
    req := &pb.ReactorDiscoveryRequest{Id:c.Id,Ip:c.Ip,Port:c.Port,Model:c.Model}
    resp, err := client.ReactorDiscoveryHandler(context.Background(), req)
    if err != nil {
        return err
    }
    if resp.GetSuccess() {
        fmt.Println("Central server reached")
        return nil
    } else {
        return errors.New("Failed to reach central server!")
    }
}

func (c *Coordinator) Register() error {
    fmt.Printf("Listing for pings on %v:%v\n",c.Ip,c.Port)
    lis, err := net.Listen("tcp", fmt.Sprintf("%v:%v",c.Ip(),c.Port))
    if err != nil {
        return err
    }
    grpcServer := grpc.NewServer()
    pb.RegisterMonitoringServer(grpcServer, c)
    go grpcServer.Serve(lis)
    return nil
}

func (c *Coordinator) SensorStatusHandler(ctx context.Context, ping *pb.SensorStatusRequest) (*pb.SensorStatusResponse,error) {
    c.mu.Lock()
    defer c.mu.Unlock()
    var sensors []*pb.SensorStatus
    resp := &pb.SensorStatusResponse{Id:c.Id,Sensors:sensors}
    for a, s := range c.Sensors {
        resp.Sensors = append(resp.Sensors,&pb.SensorStatus{Addr:int32(a), Type:s.GetType(), Status:s.GetStatus()})
    }
    return resp, nil
}