long_poller_go/poller.go

package long_poller_go

import (
	"context"
	"google.golang.org/grpc"
	cm "loafle.com/overflow/agent_api/config_manager"
	pb "loafle.com/overflow/crawler_go/grpc" //temp
	s "loafle.com/overflow/scheduler_go"
	"log"
	"strconv"
	"sync"
)

const (
	POLLING_ID       = "OVERFLOW_LONG_POLLING"
	DEFAULT_INTERVAL = uint64(3)
)

var (
	instance *Poller
	once     sync.Once
)

type Poller struct {
	once       sync.Once
	runStat    chan bool
	gconf      *cm.GlobalConfig
	intervalCh chan interface{}
	scheduler  *s.Scheduler
}

func init() {
	go handleConfigLoaded()
}

func GetInstance() *Poller {
	once.Do(func() {
		instance = &Poller{}
	})
	return instance
}

func startPoller(ch chan interface{}) {
	p := GetInstance()
	go func() {
		data := <-ch
		p.gconf = data.(cm.ConfigManager).GetGlobalConfig()
		p.start()
		p.addObservers()
	}()
}

func (p *Poller) addObservers() {
	go p.handleAgentStop()
}

func (p *Poller) handleInterval(ch chan interface{}) {
	go func() {
		data := <-ch
		interval := data.(string)
		i, err := strconv.Atoi(interval)
		if err != nil {
			log.Println(err)
			return
		}
		p.scheduler.UpdateInterval(POLLING_ID, uint64(i))
	}()
}

func (p *Poller) start() {
	p.once.Do(func() {
		p.scheduler = &s.Scheduler{}
		p.scheduler.Start()
		p.scheduler.NewSchedule(POLLING_ID, DEFAULT_INTERVAL, p.polling)
	})
}

func (p *Poller) Stop() {
	p.scheduler.Stop()
	p.runStat <- false
}

func (p *Poller) polling(agentId string) {

	addr := p.gconf.Central.Address + ":" + string(p.gconf.Central.Port)
	conn, err := grpc.Dial(addr, grpc.WithInsecure())
	if err != nil {
		log.Println(err)
		return
	}
	defer conn.Close()

	//todo temporary
	client := pb.NewStatusClient(conn)

	//printStatusStream(client) //stream

	out, err := client.Status(context.Background(), &pb.Empty{})
	if err != nil {
		log.Println(err)
	}

	p.dispatchNotify(out)
}

func (p *Poller) dispatchNotify(result interface{}) {
	log.Printf("Polling result - %s ", result)
	/*
		TASK_SENSOR_START
		TASK_SENSOR_STOP
		TASK_SENSOR_ADD
		TASK_SENSOR_REMOVE
		TASK_SENSOR_UPDATE
		TASK_CRAWLER_UPDATE
		TASK_AGENT_UPDATE
		TASK_LOG_SEND
	*/
}

func (p *Poller) updateInterval(interval string) {
	i, err := strconv.Atoi(interval)
	if err != nil {
		log.Println(err)
		return
	}
	p.scheduler.UpdateInterval(POLLING_ID, uint64(i))
}

/*
func printStatusStream(client pb.StatusClient) {
	stream, err := client.Status(context.Background(), &pb.Empty{})
	if err != nil {
		grpclog.Fatalf("%v.List(_) = _, %v", client, err)
	}
	for {
		feature, err := stream.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			grpclog.Fatalf("%v.List(_) = _, %v", client, err)
		}
		grpclog.Println(feature)
	}
	grpclog.Println("--------------------------")
}
*/