173 lines
3.0 KiB
Go
173 lines
3.0 KiB
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 = "/home/insanity/data/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()
|
|
}
|
|
|
|
return
|
|
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"
|
|
}
|