collector

collector.go

@@ -1,9 +1,13 @@
 package collector_go
 
 import (
+	"context"
 	confMng "loafle.com/overflow/agent_api/config_manager"
-	s "loafle.com/overflow/collector_go/scheduler"
+	"loafle.com/overflow/crawler_go/grpc"
+	crm "loafle.com/overflow/crawler_manager_go"
+	s "loafle.com/overflow/scheduler_go"
 	"log"
+	"strconv"
 	"sync"
 	"time"
 )
@@ -14,7 +18,7 @@ var (
 )
 
 func init() {
-	AddObservers()
+	handleConfigLoaded()
 }
 
 func GetInstance() *Collector {
@@ -24,26 +28,27 @@ func GetInstance() *Collector {
 	return instance
 }
 
-func AddObservers() {
-	go handleConfigLoaded()
-}
-
 type Collector struct {
-	scheduler   s.Scheduler
+	scheduler *s.Scheduler
 	cm        confMng.ConfigManager
+
+	startSensorCh     chan interface{}
+	stopSensorCh      chan interface{}
 	addSensorCh       chan interface{}
 	remSensorCh       chan interface{}
+	updateSensorCh    chan interface{}
+	crmSensorUpdateCh chan interface{}
+	crmUpdateCh       chan interface{}
+	crmUpdateDoneCh   chan interface{}
 }
 
 func (c *Collector) start(conf confMng.ConfigManager) {
 	go func() {
 		c.cm = conf
-		c.scheduler = s.Scheduler{}
+		c.scheduler = &s.Scheduler{}
-		c.scheduler.Init()
+		c.scheduler.Start()
-		c.addSensorCh = make(chan interface{})
+		c.addObservers()
-		c.remSensorCh = make(chan interface{})
+		c.notifyCollectorReady()
-		handleSensorAdded(c.addSensorCh)
-		handleSensorRemoved(c.remSensorCh)
 
 		for _, conf := range c.cm.GetSensors() {
 			if err := c.addSensor(conf.Id); err != nil {
@@ -53,29 +58,50 @@ func (c *Collector) start(conf confMng.ConfigManager) {
 	}()
 }
 
+func (c *Collector) addObservers() {
+
+	c.startSensorCh = make(chan interface{})
+	c.stopSensorCh = make(chan interface{})
+	c.addSensorCh = make(chan interface{})
+	c.remSensorCh = make(chan interface{})
+	c.updateSensorCh = make(chan interface{})
+	c.crmSensorUpdateCh = make(chan interface{})
+	c.crmUpdateCh = make(chan interface{})
+	c.crmUpdateDoneCh = make(chan interface{})
+
+	go c.handleAgentWillStop()
+	go c.handleSensorStart()
+	go c.handleSensorStop()
+	go c.handleSensorAdd()
+	go c.handleSensorRemove()
+	go c.handleSensorUpdate()
+	go c.handleCrmSensorUpdateDone()
+	go c.handleCrawlerUpdate()
+	go c.handleCrawlerUpdateDone()
+}
+
 func (c *Collector) Stop() {
-	cleanObserver(c.addSensorCh, c.remSensorCh)
+	c.cleanObserver(c.addSensorCh, c.remSensorCh)
 	c.scheduler.RemoveAllSchedule()
 	c.scheduler.Stop()
+	c.notifyCollectorStopped()
 }
 
 func (c *Collector) collect(id string) {
 
 	conf := c.cm.GetSensorById(id)
 	log.Printf("COLLECT %s - [ID: %s] [Crawler : %s]", time.Now(), conf.Id, conf.Crawler.Name)
-
-	/*
 	conn, err := crm.GetInstance().GetClient(conf.Crawler.Container)
 	if err != nil {
 		log.Println(err)
 	}
 	defer conn.Close()
 
-		dc := g.NewDataClient(conn)
+	dc := grpc.NewDataClient(conn)
-		in := &g.Input{}
+	in := &grpc.Input{}
 
 	in.Id = id
-		in.Name = g.Crawlers(g.Crawlers_value[conf.Crawler.Name])
+	in.Name = grpc.Crawlers(grpc.Crawlers_value[conf.Crawler.Name])
 
 	out, err := dc.Get(context.Background(), in)
 
@@ -83,17 +109,25 @@ func (c *Collector) collect(id string) {
 		log.Println(err)
 	}
 	log.Println(out)
-	*/
+
+	c.notifyData(out)
 }
 
 func (c *Collector) addSensor(sensorId string) error {
 	sensor := c.cm.GetSensorById(sensorId)
-	return c.scheduler.NewSchedule(sensorId, sensor.Schedule.Interval, c.collect)
+	interval, err := strconv.Atoi(sensor.Schedule.Interval)
+	if err != nil {
+		return err
+	}
+	return c.scheduler.NewSchedule(sensorId, uint64(interval), c.collect)
 }
 
 func (c *Collector) removeSensor(id string) {
 	if err := c.scheduler.RemoveSchedule(id); err != nil {
 		log.Println(err)
-		return
 	}
 }
+
+func (c *Collector) updateSensor(id string) {
+	//update
+}

collector_event.go

@@ -8,40 +8,103 @@ import (
 
 func handleConfigLoaded() {
 	ch := make(chan interface{}, 0)
-	observer.Add(messages.CONFIGMANAGER_LOADED, ch)
+	observer.Add(messages.CRM_READY, ch)
 
 	data := <-ch
 	confMng := data.(config_manager.ConfigManager)
 
 	coll := GetInstance()
 	coll.start(confMng)
-
+	observer.Remove(messages.CRM_READY, ch)
-	observer.Remove(messages.CONFIGMANAGER_LOADED, ch)
 }
 
-func handleSensorAdded(ch chan interface{}) {
+func (c *Collector) notifyCollectorReady() {
-	coll := GetInstance()
+	observer.Notify(messages.CLT_READY, nil)
-	observer.Add(messages.ADD_SENSOR_2_END, ch)
+}
 
-	for {
+func (c *Collector) notifyCollectorStopped() {
-		configId := <-ch
+	observer.Notify(messages.CLT_STOPPED, nil)
-		coll.addSensor(configId.(string))
+}
+
+func (c *Collector) handleAgentWillStop() {
+	ch := make(chan interface{}, 0)
+	observer.Add(messages.AGT_WILL_STOPPED, ch)
+	data := <-ch
+	if data {
+		c.Stop()
+	}
+	observer.Remove(messages.AGT_WILL_STOPPED, ch)
+}
+
+func (c *Collector) handleSensorStart() {
+	observer.Add(messages.TASK_SENSOR_START, c.startSensorCh)
+	configId := <-c.startSensorCh
+	err := c.addSensor(configId.(string))
+	if err == nil {
+		//todo task done
 	}
 }
 
-func handleSensorRemoved(ch chan interface{}) {
+func (c *Collector) handleSensorStop() {
-	coll := GetInstance()
+	observer.Add(messages.TASK_SENSOR_STOP, c.stopSensorCh)
-	observer.Add(messages.REMOVE_SENSOR_1, ch)
+	configId := <-c.stopSensorCh
+	c.removeSensor(configId.(string))
+	//todo task done
+}
 
-	for {
+func (c *Collector) handleSensorAdd() {
-		configId := <-ch
+	observer.Add(messages.CRM_SENSOR_ADD_DONE, c.addSensorCh)
-		coll.removeSensor(configId.(string))
+	configId := <-c.addSensorCh
-		observer.Notify(messages.REMOVE_SENSOR_2_END, configId)
+	err := c.addSensor(configId.(string))
+	if err == nil {
+		//todo task done
 	}
 }
 
-func cleanObserver(ach chan interface{}, rch chan interface{}) {
+func (c *Collector) handleSensorRemove() {
-	observer.Remove(messages.ADD_SENSOR_2_END, ach)
+	observer.Add(messages.TASK_SENSOR_REMOVE, c.remSensorCh)
-	observer.Remove(messages.REMOVE_SENSOR_1, rch)
+	configId := <-c.remSensorCh
-	observer.Notify(messages.COLLECTOR_STOPPED, true)
+	c.removeSensor(configId.(string))
+	observer.Notify(messages.CLT_SENSOR_REMOVE_DONE, configId)
+}
+
+func (c *Collector) handleSensorUpdate() {
+	observer.Add(messages.TASK_SENSOR_UPDATE, c.updateSensorCh)
+
+	configId := <-c.updateSensorCh
+	//todo update logic
+	observer.Notify(messages.CLT_SENSOR_UPDATE_DONE, configId)
+}
+
+func (c *Collector) handleCrmSensorUpdateDone() {
+	observer.Add(messages.CRM_SENSOR_UPDATE_DONE, c.crmSensorUpdateCh)
+
+	//configId := <-c.crmSensorUpdateCh
+	//todo update logic
+	//todo task done
+}
+
+func (c *Collector) handleCrawlerUpdate() {
+	observer.Add(messages.TASK_CRAWLER_UPDATE, c.crmUpdateCh)
+	//todo. do what collector has to do
+}
+
+func (c *Collector) handleCrawlerUpdateDone() {
+	observer.Add(messages.CRM_UPDATE_DONE, c.crmUpdateDoneCh)
+	//todo. task done
+}
+
+func (c *Collector) cleanObserver(ach chan interface{}, rch chan interface{}) {
+	observer.Remove(messages.TASK_SENSOR_START, c.startSensorCh)
+	observer.Remove(messages.TASK_SENSOR_STOP, c.stopSensorCh)
+	observer.Remove(messages.CRM_SENSOR_ADD_DONE, c.addSensorCh)
+	observer.Remove(messages.TASK_SENSOR_REMOVE, c.remSensorCh)
+	observer.Remove(messages.TASK_SENSOR_UPDATE, c.updateSensorCh)
+	observer.Remove(messages.TASK_CRAWLER_UPDATE, c.crmUpdateCh)
+	observer.Remove(messages.CRM_SENSOR_UPDATE_DONE, c.crmSensorUpdateCh)
+	observer.Remove(messages.CRM_UPDATE_DONE, c.crmUpdateDoneCh)
+}
+
+func (c *Collector) notifyData(data interface{}) {
+	observer.Notify(messages.CLT_DATA_SEND, data)
 }

collector_test.go

@@ -1,28 +1,7 @@
 package collector_go
 
-import (
+import "testing"
-	s "loafle.com/overflow/collector_go/scheduler"
-	"log"
-	"testing"
-	"time"
-)
 
-func TestCollector(t *testing.T) {
+func TestCollect(t *testing.T) {
-	sc := s.Scheduler{}
-	sc.Init()
 
-	for i := 0; i < 9999; i++ {
-		sc.NewSchedule(string(i), "5", test)
-	}
-	sc.NewSchedule("#######################", "5", test)
-	time.Sleep(time.Second * 10)
-
-	sc.NewSchedule("#######################", "1", test)
-	time.Sleep(time.Second * 100)
-}
-
-func test(id string) {
-	if id == "#######################" {
-		log.Println(id)
-	}
 }

scheduler/scheduler.go

@@ -1,56 +0,0 @@
-package scheduler
-
-import (
-	c "loafle.com/overflow/crawler_go"
-	"loafle.com/overflow/cron_go"
-	"strconv"
-	"sync"
-)
-
-const DEFAULT_INTERVAL = 5
-
-type Scheduler struct {
-	crawler  *c.CrawlerImpl
-	once     sync.Once
-	cronChan chan bool
-}
-
-func (s *Scheduler) Init() {
-	s.once.Do(func() {
-		s.cronChan = cron.Start()
-		s.crawler = &c.CrawlerImpl{}
-	})
-}
-
-func (s *Scheduler) Stop() {
-	s.cronChan <- false
-}
-
-func (s *Scheduler) NewSchedule(id, interval string, fn interface{}) error {
-
-	return s.newSchedule(id, interval, fn)
-}
-
-func (s *Scheduler) RemoveSchedule(id string) error {
-	return cron.Remove(id)
-}
-
-func (s *Scheduler) RemoveAllSchedule() error {
-	return cron.RemoveAll()
-}
-
-func (s *Scheduler) UpdateSchedule(id string, interval string) {
-	i, err := strconv.Atoi(interval)
-	if err != nil {
-		i = DEFAULT_INTERVAL
-	}
-	cron.UpdateTask(id, uint64(i))
-}
-
-func (s *Scheduler) newSchedule(id string, interval string, fn interface{}) error {
-	i, err := strconv.Atoi(interval)
-	if err != nil {
-		i = DEFAULT_INTERVAL
-	}
-	return cron.AddTask(id, uint64(i)).Invoke(fn, id)
-}