From 1873461012c99801f364bd07fae2c218d245048e Mon Sep 17 00:00:00 2001
From: zhangzengfei <zhangzengfei@smartai.com>
Date: 星期一, 11 九月 2023 19:42:39 +0800
Subject: [PATCH] 修复bom第一启动时的全量上报
---
kingdee/query.go | 102 ++++++++++++++++++++++++++++++++-------------------
1 files changed, 64 insertions(+), 38 deletions(-)
diff --git a/kingdee/query.go b/kingdee/query.go
index cfbaeb8..6bef988 100644
--- a/kingdee/query.go
+++ b/kingdee/query.go
@@ -2,6 +2,7 @@
import (
"encoding/json"
+ "errors"
"strings"
"kingdee-dbapi/config"
@@ -9,69 +10,94 @@
"kingdee-dbapi/nsqclient"
)
-func QueryMsgHandle(data []byte) error {
- var result []interface{}
+// 閫氱敤sql鏌ヨ鎺ュ彛
- var sql = string(data)
+type SqlQueryMsg struct {
+ Key string // 璇锋眰
+ Command string
+ Success bool
+ Message string
+ Result []byte
+}
- logger.Debug("鎺ユ敹鍒版煡璇㈣姹�,%s", sql)
+func SqlQueryHandle(msg []byte) error {
+ var query SqlQueryMsg
- if !sqlCheck(sql) {
- logger.Warn("璇嗗埆鍒板嵄闄╃殑sql璇彞, 鎷掔粷鎵ц. %s", sql)
-
- return nil
- }
-
- if db == nil {
- logger.Debug("鏁版嵁搴撴湭杩炴帴")
-
- return nil
- }
-
- rows, err := db.Raw(`select * from users`).Rows()
- if err != nil {
- result = append(result, err.Error())
+ if err := json.Unmarshal(msg, &query); err != nil {
+ logger.Warn("瑙ｆ瀽璇锋眰澶辫触, err:%s, msg:%s", err.Error(), string(msg))
return err
}
- var colums []string
+ var sql = query.Command
+ logger.Debug("鎺ユ敹鍒版煡璇㈣姹�,%s", sql)
+
+ if !sqlCheck(sql) {
+ query.Message = "鍗遍櫓鐨剆ql璇彞, 鎷掔粷鎵ц"
+ logger.Warn(query.Message)
+ } else {
+ result, err := execSqlCommand(sql)
+ if err != nil {
+ query.Message = err.Error()
+ logger.Warn("sql鎵ц澶辫触:%s", query.Message)
+ } else {
+ query.Result = result
+ query.Success = true
+ logger.Warn("sql鎵ц瀹屾垚.")
+ }
+ }
+
+ replyData, _ := json.Marshal(query)
+ ok := nsqclient.Produce(config.Options.SqlReplyTopic, replyData)
+ logger.Warn("搴旂瓟鏌ヨ璇锋眰缁撴灉:%t, key:%s", ok, query.Key)
+
+ return nil
+}
+
+func execSqlCommand(sql string) ([]byte, error) {
+ var result []interface{}
+
+ if db == nil {
+ return nil, errors.New("鏁版嵁搴撴湭杩炴帴")
+ }
+
+ rows, err := db.Raw(sql).Rows()
+ if err != nil {
+ return nil, err
+ }
+
+ var cols []string
for rows.Next() {
//鍏堣幏鍙栨墍鏈夌殑column
- if colums == nil {
- colums, _ = rows.Columns()
+ if cols == nil {
+ cols, _ = rows.Columns()
}
//寤虹珛淇╀釜interface鏁扮粍锛宑olumnPointers涓瓨鍦╟olumns鐨勫湴鍧�
- columns := make([]interface{}, len(colums))
- columnPointers := make([]interface{}, len(colums))
- for i, _ := range columns {
+ columns := make([]interface{}, len(cols))
+ columnPointers := make([]interface{}, len(cols))
+ for i := 0; i < len(columns); i++ {
//璧嬪�煎湴鍧�
columnPointers[i] = &columns[i]
}
//鎵弿缁撴灉
- rows.Scan(columnPointers...)
+ err = rows.Scan(columnPointers...)
+ if err != nil {
+ return nil, err
+ }
+
m := make(map[string]interface{})
- for i, colName := range colums {
+ for i, colName := range cols {
val := columnPointers[i].(*interface{})
m[colName] = *val
}
result = append(result, m)
- //result = append(result, row)
}
- logger.Debug("鏁版嵁搴撹繑鍥炴暟鎹�%+v", result)
- b, _ := json.Marshal(result)
+ rb, _ := json.Marshal(result)
- ok := nsqclient.Produce(config.Options.ReplyTopic, b)
- if !ok {
- logger.Warn("搴旂瓟鏌ヨ璇锋眰澶辫触.")
- } else {
- logger.Debug("搴旂瓟鏌ヨ璇锋眰鎴愬姛. 鏁版嵁:%s", string(b))
- }
-
- return nil
+ return rb, nil
}
// 绠�鍗曡繃婊や笅sql璇彞,鎷掔粷澧炲垹鏀规搷浣�
--
Gitblit v1.8.0