DMAC/internal/pkg/server/coordinator.go

package server

import (
	pb "FRMS/internal/pkg/grpc"
	"FRMS/internal/pkg/influxdb"
	_ "FRMS/internal/pkg/influxdb"
	"FRMS/internal/pkg/logging"
	"context"
	"errors"
	"fmt"
	"net"
	"sync"

	"github.com/spf13/viper"
	"google.golang.org/grpc"
)

// this package creates the central coordiantor and sub coordiantors to route clients

// db client interface
type Database interface {
	// getters (all create if doesnt exist)
	GetReactorClient(int) (string, string, string, string, error) // returns (url, org, bucket, token, err)
}

func NewDatabaseAdmin(config *viper.Viper) (Database, error) {
	return influxdb.NewDBAdmin(config)
}

type CentralCoordinator struct {
	// main coordinator
	ClientConnections *ClientPacket
	*ReactorCoordinator
	Database
	Config *viper.Viper
	// from config
	Ports map[string]int `mapstructure:"ports"`
	Err   chan error
}

func NewCentralCoordinator(config *viper.Viper, ch chan error) *CentralCoordinator {
	// create a central coordinator to manage requests
	db, err := NewDatabaseAdmin(config)
	if err != nil {
		ch <- err
	}

	rc, err := NewReactorCoordinator(config, ch)
	if err != nil {
		ch <- err
	}
	config.UnmarshalKey("server.ports", rc) // get reactor port
	c := &CentralCoordinator{
		Err:                ch,
		Config:             config,
		Database:           db,
		ReactorCoordinator: rc,
	}
	// grab config settings
	if err = config.UnmarshalKey("server", c); err != nil {
		ch <- err
	}

	return c
}

func (c *CentralCoordinator) Start() {
	// starts up associated funcs
	clientChan := make(chan *ClientPacket)
	l := NewListener(clientChan, c.Err)
	// grabs lis port
	c.Config.UnmarshalKey("server.ports", l)

	// starting reactor coordinator
	if err := c.ReactorCoordinator.Start(); err != nil {
		c.Err <- err
	}
	// starting listener
	if err := l.Start(); err != nil {
		c.Err <- err
	}
	// lastly start client listener
	go c.ClientListener(clientChan)
}

func (c *CentralCoordinator) ClientListener(ch chan *ClientPacket) {
	for client := range ch {
		// basically loops until channel is closed
		client.Response <- c.ClientHandler(client.Client) // respond with cred
	}
}

func (c *CentralCoordinator) ClientHandler(cl *Client) *ClientResponse {
	// returns reactor db info
	var err error
	cr := &ClientResponse{Port: c.Ports[cl.Type]}

	if cl.Type == "reactor" {
		// get reactor info
		go c.ReactorCoordinator.ClientHandler(cl)
		// db info
		cr.URL, cr.Org, cr.Token, cr.Bucket, err = c.Database.GetReactorClient(cl.Id)
	} else {
		// throw error
		err = errors.New(fmt.Sprintf("Client type %s not recognized!"))
	}
	// returns based on cl type
	if err != nil {
		c.Err <- err
	}
	return cr
}

type ReactorCoordinator struct {
	Port int `mapstructure:"reactor"`
	*ReactorManagers
	Err chan error
	pb.UnimplementedMonitoringServer
}

type ReactorManagers struct {
	Config    *viper.Viper
	Directory map[int]*ReactorManager
	sync.RWMutex
}

func NewReactorCoordinator(config *viper.Viper, errCh chan error) (*ReactorCoordinator, error) {
	rmap := make(map[int]*ReactorManager)
	rm := &ReactorManagers{Directory: rmap, Config: config}
	c := &ReactorCoordinator{Err: errCh, ReactorManagers: rm}
	return c, nil
}

func (c *ReactorCoordinator) Start() error {
	logging.Debug(logging.DStart, "RCO 01 Starting!")
	// register grpc service
	return c.Register()
}

func (c *ReactorCoordinator) ClientHandler(cl *Client) {
	// updates clients if nessecary
	if err := c.UpdateReactorManager(cl, c.Err); err != nil {
		c.Err <- err
	}
}

func (m *ReactorManagers) GetReactorManager(id int) (*ReactorManager, error) {
	m.RLock()
	defer m.RUnlock()

	rm, exists := m.Directory[id]
	if !exists {
		return &ReactorManager{}, errors.New(fmt.Sprintf("No manager for reactor %d!", id))
	}
	return rm, nil
}

func (m *ReactorManagers) UpdateReactorManager(cl *Client, errCh chan error) error {
	// locking
	m.RLock()
	defer m.RUnlock()

	var err error

	rm, exists := m.Directory[cl.Id]
	if !exists {
		logging.Debug(logging.DClient, "RCO creating manager for reactor client %v", cl.Id)
		// creating
		rm = NewReactorManager(cl, m.Config, errCh)
		// starting
		if err = rm.Start(); err != nil {
			return err
		}
		m.Directory[cl.Id] = rm
	}
	return rm.UpdateClient(cl)
}

func (r *ReactorCoordinator) Register() error {
	lis, err := net.Listen("tcp", fmt.Sprintf(":%v", r.Port))
	if err != nil {
		return err
	}
	grpcServer := grpc.NewServer()
	pb.RegisterMonitoringServer(grpcServer, r)
	go grpcServer.Serve(lis)
	logging.Debug(logging.DClient, "RCO ready for client requests")
	return nil
}

func (r *ReactorCoordinator) ReactorStatusHandler(ctx context.Context, req *pb.ReactorStatusPing) (*pb.ReactorStatusResponse, error) {
	rm, err := r.GetReactorManager(int(req.GetId()))
	// error checking
	if err != nil {
		return &pb.ReactorStatusResponse{}, err
	}
	return rm.ReactorStatusHandler(ctx, req)
}