del comment

main.go

@@ -282,11 +282,7 @@ func kafkaConsumerBytes(done chan interface{}, req, resTopic string) (chan pbUps
 
 func batchToQuestDb(done chan interface{}, reqStream chan pbUpstream.TaskContent, resStream chan pbUpstream.TaskReceipt, reqTableName string, resTableName, questAddr string) {
 
-	//tableName := "tasks"
-
 	ctx := context.TODO()
-	// Connect to QuestDB running on 127.0.0.1:9009
-	//addrOpt := qdb.WithAddress("192.168.1.220:9009")
 	addrOpt := qdb.WithAddress(questAddr)
 
 	var sender *qdb.LineSender
@@ -327,7 +323,6 @@ func batchToQuestDb(done chan interface{}, reqStream chan pbUpstream.TaskContent
 					Int64Column("workload", int64(task.TaskWorkload)).
 					Int64Column("out_len", int64(task.TaskOutLen)).
 					Int64Column("task_duration", int64(task.TaskDuration)).
-					//task_execute_duration
 					Int64Column("exec_duration", int64(task.TaskExecuteDuration)).
 					AtNow(ctx)
 
@@ -343,17 +338,6 @@ func batchToQuestDb(done chan interface{}, reqStream chan pbUpstream.TaskContent
 
 		case task := <-reqStream:
 
-			// sender.
-			// 	Table(reqTableName).
-			// 	Symbol("type", fmt.Sprintf("%d", randomType())).
-			// 	Symbol("uid", fmt.Sprintf("%d", randomUID())).
-			// 	StringColumn("id", uuid.New().String()).
-			// 	TimestampColumn("time", time.Now()).
-			// 	Int64Column("fee", int64(randomFeeAndWorkload())).
-			// 	Int64Column("in_len", int64(randomInput())).
-			// 	AtNow(ctx)
-
-			_ = task
 			taskFeeAsInt, err := strconv.Atoi(task.TaskFee)
 			if err != nil {
 				slog.Error("task.TaskFee string to int", "error", err)
@@ -365,10 +349,6 @@ func batchToQuestDb(done chan interface{}, reqStream chan pbUpstream.TaskContent
 			nanoseconds := int64(task.TaskTimestamp)
 			seconds := nanoseconds / 1e9
 
-			// 使用Unix函数转换为time.Time
-			// t := time.Unix(seconds, nanoseconds%1e9)
-			// time.Unix()
-
 			for {
 
 				sender.
@@ -408,20 +388,15 @@ type ResponseJson struct {
 func newCache(redisAddr, redisPass, mysqlIP, dbName, user, passwd string, port int) *cachedata.CacheData {
 
 	_cache := cachedata.NewCacheData(context.Background(), cachedata.RedisConnParam{
-		//Addr:     "192.168.1.10:6379",
 		Addr:     redisAddr,
 		Password: redisPass,
 		DbIndex:  0,
 	}, model.DbConfig{
-		//Host:   "192.168.1.211",
 		Host:   mysqlIP,
 		Port:   port,
 		DbName: dbName,
 		Passwd: passwd,
 		User:   user,
-		// DbName: "liuxuzhong",
-		// Passwd: "12345678",
-		// User:   "root",
 	})
 	return _cache
 }
@@ -550,7 +525,7 @@ func main() {
 	slog.Warn("start param", "!!!!!!!!!!!!!!!!!!!!**********With Bill DB**********!!!!!!!!!!!!!!!!!!!!", withBillDb)
 	slog.Warn("start param", "!!!!!!!!!!!!!!!!!!!!**********With Bill DB**********!!!!!!!!!!!!!!!!!!!!", withBillDb)
 
-	go Registry(redisAddr, redisPassWd)
+	//go Registry(redisAddr, redisPassWd)
 
 	kafkaBrokers = []string{kafkaBroker}
 	callbackAddr = callbackAddrP
@@ -564,10 +539,6 @@ func main() {
 	createTable(questAddr[:idx], createTaskTableSql)
 	createTable(questAddr[:idx], createBillsTableSql)
 
-	// DbName: "liuxuzhong",
-	// Passwd: "12345678",
-	// User:   "root",
-
 	cache = newCache(redisAddr, redisPassWd, mysqlAddr, mysqlDbName, mysqlUser, mysqlPassWd, mysqlPort)
 
 	// Create a WaitGroup to synchronize goroutines.
@@ -588,15 +559,6 @@ func main() {
 	// Create a new instance of the Fiber web framework.
 	app := fiber.New()
 
-	// cfg := swagger.Config{
-	// 	BasePath: "/",
-	// 	FilePath: "./docs/swagger.json",
-	// 	Path:     "swagger",
-	// 	Title:    "Swagger API Docs",
-	// }
-
-	// app.Use(swagger.New(cfg))
-
 	apiGroup := app.Group("/api")
 	jwtGroup := app.Group("/jwt")
 	callbackGroup := app.Group("/callback")
@@ -643,24 +605,10 @@ func main() {
 
 var replanceQueryTask *model.TaskType
 
-//var m = make([]byte, 1024*1024+1024*512)
-//var m = make([]byte, 1024*1024+1024*512)
-
-var mypool = &sync.Pool{
-	New: func() interface{} {
-		mem := make([]byte, 1024*1024+1024*512)
-		return &mem
-	},
-}
-
 func ApiAndJWT(c *fiber.Ctx) error {
 
 	slog.Debug("ApiAndJWT", "path", c.Route().Path)
 
-	//m := mypool.Get().(*[]byte)
-	//c, err := w.Write(*m)
-	//return c.Send(*m)
-
 	reqHeaders := c.GetReqHeaders()
 	//return c.SendString(reqHeaders["Task-Id"][0])
 
@@ -698,9 +646,6 @@ func ApiAndJWT(c *fiber.Ctx) error {
 
 	}
 
-	//var cmd string
-	//cmd = task.Cmd
-
 	pbMsg := pbUpstream.TaskContent{
 		TaskId:          reqHeaders["Task-Id"][0],
 		TaskType:        uint64(task.ID),
@@ -741,8 +686,6 @@ func ApiAndJWT(c *fiber.Ctx) error {
 		Bytes: pbBytes,
 	}
 
-	//fmt.Println("pbMsg.TaskUid--------------", pbMsg.TaskId)
-
 	asyncMode := false
 
 	// for k, v := range reqHeaders {
@@ -750,7 +693,6 @@ func ApiAndJWT(c *fiber.Ctx) error {
 	// }
 
 	if headerSync, ok := reqHeaders["Async"]; ok {
-		//fmt.Println("sync-----------------sync", headerSync)
 		for _, syncAsString := range headerSync {
 			if syncAsString == "true" {
 				asyncMode = true
@@ -761,8 +703,6 @@ func ApiAndJWT(c *fiber.Ctx) error {
 
 	slog.Info("task info", "taskid", pbMsg.TaskId, "asyncMode", asyncMode)
 
-	//fmt.Println("asyncMode-----------", asyncMode)
-
 	if asyncMode {
 		//time.Sleep(10 * time.Second)
 		//m := make([]byte, 1024*1024+1024*512)
@@ -795,8 +735,6 @@ func syncModeF(c *fiber.Ctx, taskid string) error {
 
 	slog.Debug("resAsPb.TaskResultHeader", "resAsPb.TaskResultHeader", resAsPb.TaskResultHeader)
 
-	//fmt.Println("resAsPb.TaskResultHeader", string(resAsPb.TaskResultHeader), len(resAsPb.TaskResultHeader))
-
 	if resAsPb.TaskResultHeader != nil {
 		if len(resAsPb.TaskResultHeader) != 0 {
 

service-register.go

 package main
 
-import (
-	"encoding/json"
-	"time"
-
-	"github.com/odysseus/service-registry/common"
-	"github.com/odysseus/service-registry/query"
-	registry "github.com/odysseus/service-registry/registry"
-)
-
-type demoService struct {
-}
-
-func (d demoService) ServiceType() common.ServiceType {
-	return common.SERVICE_API_GATEWAY
-}
-
-func (d demoService) Endpoint() string {
-	return "http://service-endpoint_ai-api-magr:6001"
-}
-
-func (d demoService) DetailInfo() (json.RawMessage, error) {
-
-	//return nil, nil
-	detail := struct {
-		IP          string `json:"ip"`
-		WorkerCount int    `json:"worker_count"`
-	}{}
-	detail.IP = ""
-	detail.WorkerCount = 10
-	return json.Marshal(detail)
-}
-
-func Registry(redisAddr, passWd string) {
-
-	_ = query.GatewayInfo{
-		Endpoint: "http://query-endpoint_ai-api-magr:6001",
-	}
-
-	r := registry.NewRegistry(registry.RedisConnParam{
-		Addr:     redisAddr,
-		Password: passWd,
-		DbIndex:  0,
-	}, demoService{})
-	r.Start()
-	time.Sleep(time.Second * 5)
-	r.Stop()
-}
+// import (
+// 	"encoding/json"
+// 	"time"
+
+// 	"github.com/odysseus/service-registry/common"
+// 	registry "github.com/odysseus/service-registry/registry"
+// )
+
+// type demoService struct {
+// }
+
+// func (d demoService) ServiceType() common.ServiceType {
+// 	return common.SERVICE_API_GATEWAY
+// }
+
+// func (d demoService) Endpoint() string {
+// 	return "http://service-endpoint_ai-api-magr:6001"
+// }
+
+// func (d demoService) DetailInfo() (json.RawMessage, error) {
+
+// 	//return nil, nil
+// 	detail := struct {
+// 		IP          string `json:"ip"`
+// 		WorkerCount int    `json:"worker_count"`
+// 	}{}
+// 	detail.IP = ""
+// 	detail.WorkerCount = 10
+// 	return json.Marshal(detail)
+// }
+
+// func Registry(redisAddr, passWd string) {
+
+// 	// _ = query.GatewayInfo{
+// 	// 	Endpoint: "http://query-endpoint_ai-api-magr:6001",
+// 	// }
+
+// 	r := registry.NewRegistry(registry.RedisConnParam{
+// 		Addr:     redisAddr,
+// 		Password: passWd,
+// 		DbIndex:  0,
+// 	}, demoService{})
+// 	r.Start()
+// 	time.Sleep(time.Second * 5)
+// 	r.Stop()
+// }