package data_sender_go import ( "google.golang.org/grpc" "log" "context" "encoding/json" "fmt" pb "loafle.com/overflow/crawler_go/grpc" "loafle.com/overflow/cron_go" q "loafle.com/overflow/queue_go" "os" "sync" ) const ( CENTRAL_ADDR = "127.0.0.1:50052" FILE_PATH = "/home/insanity/data/temp" 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(q.EVENT_TYPE, 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, len) for _, item := range items { collectedData := item.Value.(*Data) collectedData.AgentId = agentIdentifier() result = append(result, collectedData) } ds.send(result) } func (ds *DataSender) send(data []*Data) { conn, err := grpc.Dial(CENTRAL_ADDR, grpc.WithInsecure()) if err != nil { saveFailedData(data) return } defer conn.Close() //temporary client := pb.NewStatusClient(conn) out, err := client.Status(context.Background(), &pb.Empty{}) if err != nil { saveFailedData(data) return } log.Print(out) } func 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 { fmt.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" }