data_sender_go/data_sender.go

package data_sender_go

import (
	"context"
	"encoding/json"
	"google.golang.org/grpc"
	"io/ioutil"
	"loafle.com/overflow/agent_api/observer"
	pb "loafle.com/overflow/crawler_go/grpc"
	"loafle.com/overflow/cron_go"
	q "loafle.com/overflow/queue_go"
	"log"
	"os"
	"sync"
)

const (
	CENTRAL_ADDR     = "127.0.0.1:50052"
	FILE_PATH        = "/overflow/tmp/data.tmp"
	SENDER_ID        = "OVERFLOW_DATA_SENDER"
	DEFAULT_INTERVAL = uint64(5)
)

type Data struct {
	AgentId    string
	SensorId   string
	Data       map[string]string
	StartedAt  uint64
	FinishedAt uint64
}

type DataSender struct {
	once    sync.Once
	runStat chan bool
	queue   *q.LoafleQueue
}

func (ds *DataSender) Start() {
	ds.once.Do(func() {
		ds.init()
	})
}
func (ds *DataSender) Stop() {
	ds.runStat <- false
}

func (ds *DataSender) init() {
	ds.queue = q.NewQueue(observer.DATA_QUEUE, 3)

	cr := &cron.Cron{}
	ds.runStat = cr.Start()
	cr.AddTask(SENDER_ID, DEFAULT_INTERVAL).Invoke(ds.check)
}

func (ds *DataSender) check() {
	items := ds.queue.GetItems()
	len := len(items)
	if len <= 0 {
		return
	}
	result := make([]*Data, 0)
	for _, item := range items {
		collectedData := item.Value.(*Data)
		collectedData.AgentId = agentIdentifier()
		result = append(result, collectedData)
	}

	ds.send(result)
}

func (ds *DataSender) send(data []*Data) {

	bytes := ds.getFailedData()
	if bytes != nil {
		failed := Data{}
		err := json.Unmarshal(bytes, &failed)
		if err != nil {
			log.Println(err)
		}
		data = append([]*Data{&failed}, data...) //prepend
		ds.removeFailed()
	}

	conn, err := grpc.Dial(CENTRAL_ADDR, grpc.WithInsecure())
	if err != nil {
		ds.saveFailedData(data)
		return
	}
	defer conn.Close()

	//temporary
	client := pb.NewStatusClient(conn)
	out, err := client.Status(context.Background(), &pb.Empty{})
	if err != nil {
		ds.saveFailedData(data)
		return
	}
	log.Print(out)
}

func (ds *DataSender) getFailedData() []byte {
	b, err := ioutil.ReadFile(FILE_PATH)
	if err != nil {
		log.Println(err)
		return nil
	}
	return b
}

func (ds *DataSender) removeFailed() {
	err := os.Remove(FILE_PATH)
	if err != nil {
		log.Println(err)
	}
}

func (ds *DataSender) saveFailedData(datas []*Data) {

	file, err := tempFile()
	if err != nil {
		log.Println(err)
	}
	defer func() {
		if err := file.Close(); err != nil {
			log.Println(err)
			return
		}
	}()

	for _, data := range datas {
		bytes, err := json.Marshal(&data)
		if err != nil {
			log.Println(err)
		}
		if bytes != nil {
			log.Println("write : ", string(bytes))
			_, err = file.Write(bytes)
			if err != nil {
				log.Println(err)
			}
		}
	}
}

func tempFile() (*os.File, error) {

	var file *os.File
	var fileInfo os.FileInfo
	var err error
	if fileInfo, err = os.Stat(FILE_PATH); err != nil {
		if os.IsNotExist(err) {
			file, err = os.Create(FILE_PATH)
			if err != nil {
				return nil, err
			}
		}
	} else {
		if fileInfo != nil {
			file, err = os.OpenFile(FILE_PATH, os.O_RDWR|os.O_APPEND, 0660)
			if err != nil {
				return nil, err
			}
		}
	}

	return file, nil
}

func agentIdentifier() string {
	return "agentID_000000001"
}