moxitech
/
go_rabbitmq
1
1
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
go_rabbitmq/app/lib/task_executor.go

246 lines
7.8 KiB
Go
Raw Blame History

package lib
//
//import (
// "encoding/json"
// "fmt"
// "log"
// "math"
// "strconv"
// "sync"
// "time"
//)
//
//type TasksExecutor struct {
// RequestController *RequestController
// RabbitMQ *RabbitMQ
// DeviceRegistry *DeviceRegistry
//}
//
//func NewTasksExecutor(requestController *RequestController, rabbitMQ *RabbitMQ, deviceRegistry *DeviceRegistry) *TasksExecutor {
// return &TasksExecutor{
// RequestController: requestController,
// RabbitMQ: rabbitMQ,
// DeviceRegistry: deviceRegistry,
// }
//}
//
//func (te *TasksExecutor) Start() {
// // Добавить ожидание коннекта к RabbitMQ
// wg := sync.WaitGroup{}
// wg.Add(1)
// go func() {
// defer wg.Done()
// for {
// time.Sleep(2 * time.Second)
// te.Check()
// }
// }()
// wg.Wait()
//}
//
//func (te *TasksExecutor) Check() {
// result, err := te.RequestController.MySQL.Query("SELECT * FROM `queue_task` WHERE (`status` = 0 OR `status` = 1 OR `status` = 2) AND `type`=1")
// if err != nil {
// log.Printf("Error querying tasks: %v", err)
// return
// }
//
// if len(result) == 0 {
// return
// }
//
// for _, task := range result {
// if task.TsScheduled > time.Now().Unix() {
// continue
// }
//
// if time.Now().Unix() > (task.Ts+(task.Timeout)) && task.Timeout != 0 {
// te.DeleteTask("Task timeout", task.Id)
// continue
// }
//
// if task.Status == 0 {
// go te.RunTask(task)
// }
// }
//}
//
//func (te *TasksExecutor) RunTask(task Task) {
// _, err := te.RequestController.MysqlQuery(fmt.Sprintf("UPDATE `queue_task` SET `status`=1, `ts_start`='%d' WHERE `id`=%d", time.Now().Unix(), task.Id))
// if err != nil {
// log.Printf("Error updating task status: %v", err)
// return
// }
//
// switch task.NumCom {
// case 3, 8, 9, 16, 17, 18, 19, 20, 21:
// go te.SendTask(task)
// default:
// go te.DeleteTask("Unknown id_task", task.Id)
// }
//}
//
//func (te *TasksExecutor) DeleteTask(reason string, id int) {
// log.Printf("[TASK EXECUTOR] %s (id=%d)", reason, id)
// _, err := te.RequestController.MysqlQuery(fmt.Sprintf("UPDATE `queue_task` SET `status`=3, `ts_end`=%d WHERE `id`=%d", time.Now().Unix(), id))
// if err != nil {
// log.Printf("Error deleting task: %v", err)
// }
//}
//
//func (te *TasksExecutor) SendTask(task Task) {
// channel, err := te.RabbitMQ.CreateQueue(fmt.Sprintf("rpc_%d", task.Mid))
// if err != nil {
// log.Printf("Error creating queue: %v", err)
// return
// }
//
// options := map[string]interface{}{
// "replyTo": "rpc_queue",
// "deliveryMode": 2,
// }
//
// task.Params = json.RawMessage(task.ParamsStr)
// if task.Timeout != 0 {
// options["expiration"] = task.Timeout * 1000
// }
//
// err = channel.SendToQueue(fmt.Sprintf("rpc_%d", task.Mid), []byte(fmt.Sprintf(`{"task_id":%d,"num_com":%d,"params":%s}`, task.Id, task.NumCom, string(task.Params))), options)
// if err != nil {
// log.Printf("Error sending task to queue: %v", err)
// return
// }
//
// if task.NumCom == 17 {
// findDevice, err := te.DeviceRegistry.FindDevice(task.Mid)
// if err != nil {
// log.Printf("Error finding device: %v", err)
// return
// }
// if findDevice != nil {
// findDevice.Blocked = task.Params.Command
// _, err := te.RequestController.MysqlQuery(fmt.Sprintf("UPDATE `device` SET `blocked`= '%s' WHERE id = %d", task.Params.Command, task.DevId))
// if err != nil {
// log.Printf("Error updating device: %v", err)
// }
// }
// }
//}
//
//func (te *TasksExecutor) TaskParse(packet Packet) {
// task, err := te.RequestController.MysqlQuery(fmt.Sprintf("SELECT * FROM `queue_task` WHERE `id`=%d", packet.TaskId))
// if err != nil {
// log.Printf("Error querying task: %v", err)
// return
// }
//
// if packet.Status != nil {
// switch task[0].NumCom {
// case 16:
// te.RequestController.STOMP.SendMessage("selftest", task[0].Id, fmt.Sprintf("%v", packet))
// }
// if *packet.Status >= 4 && *packet.Status <= 6 {
// _, err := te.RequestController.MysqlQuery(fmt.Sprintf("UPDATE `queue_task` SET `status` = %d, `ts_end` = %d WHERE `id`=%d", *packet.Status, time.Now().Unix(), packet.TaskId))
// if err != nil {
// log.Printf("Error updating task status: %v", err)
// }
// } else {
// _, err := te.RequestController.MysqlQuery(fmt.Sprintf("UPDATE `queue_task` SET `status` = %d WHERE `id`=%d", *packet.Status, packet.TaskId))
// if err != nil {
// log.Printf("Error updating task status: %v", err)
// }
// }
// } else {
// if len(task) == 0 {
// return
// }
// switch task[0].NumCom {
// case 3:
// case 8:
// case 9:
// case 16:
// te.RequestController.STOMP.SendMessage("selftest", task[0].Id, fmt.Sprintf("%v", packet))
// if packet.Data != nil && packet.Data.Done {
// te.SelftestResult(packet.Data, packet.TaskId, task[0].DevId)
// }
// case 17:
// case 18:
// go te.RPCResult(packet)
// case 19:
// case 20:
// log.Printf("%d - %v", task[0].Id, packet)
// case 21:
// default:
// log.Printf("[TASK EXECUTOR] Incorrect num_com for task %d - %v", task[0].Id, packet)
// }
// }
//}
//
//func (te *TasksExecutor) RPCResult(data Packet) {
// data.Mid = nil
// _, err := te.RequestController.MysqlQuery("SET sql_mode = NO_BACKSLASH_ESCAPES;")
// if err != nil {
// log.Printf("Error setting SQL mode: %v", err)
// }
// te.RequestController.STOMP.SendMessage("rpc_result", data.TaskId, fmt.Sprintf("%v", data))
// _, err = te.RequestController.MysqlQuery(fmt.Sprintf("UPDATE `queue_task` SET `log` = '%v', `ts_end`=%d WHERE `id`=%d", data, time.Now().Unix(), data.TaskId))
// if err != nil {
// log.Printf("Error updating task status: %v", err)
// }
//}
//
//func (te *TasksExecutor) SelftestResult(data map[string]interface{}, taskId, deviceId int) {
// testingModes := []string{}
// for testingMode := range data {
// if testingMode != "done" {
// testingModes = append(testingModes, testingMode)
// }
// }
//
// result := map[string]map[string][]string{}
// for _, iface := range testingModes {
// result[iface] = map[string][]string{}
// for key, value := range data[iface].(map[string]interface{}) {
// result[iface][key] = []string{}
// if iface == "ping" {
// result[iface][key] = append(result[iface][key], fmt.Sprintf("%f ms", value.(float64)))
// } else {
// result[iface][key] = append(result[iface][key], fmt.Sprintf("%v", value))
// }
// }
// }
//
// // Перепаршиваем данные в читаемый для фронта вид
// dataForStomp := map[string]map[string][]string{}
// for iface, modes := range result {
// dataForStomp[iface] = map[string][]string{}
// for mode, values := range modes {
// dataForStomp[iface][mode] = []string{}
// if mode != "latency" && mode != "packet_loss" {
// speedLimits := []int{0, 1024, 1024 * 1024, 1024 * 1024 * 1024, 1024 * 1024 * 1024 * 1024, 1024 * 1024 * 1024 * 1024 * 1024}
// speedNames := []string{"", "b/s", "Kb/s", "Mb/s", "Gb/s", "Tb/s"}
// for i := 1; i <= len(speedNames); i++ {
// if floatValue, err := strconv.ParseFloat(values[0], 64); err == nil {
// if floatValue >= float64(speedLimits[i-1]) && floatValue < float64(speedLimits[i]) {
// n := float64(speedLimits[i-1])
// dataForStomp[iface][mode] = append(dataForStomp[iface][mode], fmt.Sprintf("%.2f %s", math.Floor((floatValue/n)*100)/100, speedNames[i]))
// break
// }
// }
// }
// } else if mode == "latency" {
// dataForStomp[iface][mode] = append(dataForStomp[iface][mode], fmt.Sprintf("%v ms", values[0]))
// } else {
// dataForStomp[iface][mode] = append(dataForStomp[iface][mode], values...)
// }
// }
// }
//
// te.RequestController.STOMP.SendMessage("selftest", taskId, fmt.Sprintf(`{"type":"result", "data": %v}`, dataForStomp))
// _, err := te.RequestController.MysqlQuery(fmt.Sprintf("UPDATE queue_task SET log = '%v', ts_end = %d WHERE id = %d", result, time.Now().Unix(), taskId))
// if err != nil {
// log.Printf("Error updating task status: %v", err)
// }
//}
Reference in New Issue View Git Blame Copy Permalink