liulei/jx-callback
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1

Browse Source
This commit is contained in:
邹宗楠
2023-05-29 15:21:20 +08:00
parent 49334fdb4c
commit 98a52a7559
1 changed files with 39 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files

187
business/jxstore/event/event_tcp.go
View File

@@ -5,7 +5,10 @@ import (
"encoding/json"
"errors"
"fmt"
"git.rosy.net.cn/baseapi"
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/jx-callback/business/jxutils/jxcontext"
"git.rosy.net.cn/jx-callback/business/jxutils/tasksch"
"git.rosy.net.cn/jx-callback/business/model"
"git.rosy.net.cn/jx-callback/business/model/dao"
"git.rosy.net.cn/jx-callback/globals"
@@ -24,22 +27,43 @@ func ListenTcp() {
return
}
for {
c, err := l.Accept()
if err != nil {
fmt.Println("accept error:", err)
break
}
go func() {
t := NewTcpClient()
if err := t.handleConn(c); err != nil {
t = nil
return
//c, err := l.Accept()
//if err != nil {
// fmt.Println("accept error:", err)
// break
//}
//go func() {
// t := NewTcpClient()
// if err := t.handleConn(c); err != nil {
// t = nil
// return
// }
//}()
taskFunc := func(task *tasksch.ParallelTask, batchItemList []interface{}, params ...interface{}) (retVal interface{}, err error) {
c, err := l.Accept()
if err != nil {
fmt.Println("accept error:", err)
return nil, err
}
}()
t := NewTcpClient()
if err := handleConn(c, t); err != nil {
t = nil
return nil, err
}
return nil, nil
}
task := tasksch.NewParallelTask("定时任务接口监听", tasksch.NewParallelConfig().SetParallelCount(50), jxcontext.AdminCtx, taskFunc, nil)
tasksch.HandleTask(task, nil, true).Run()
if _, err = task.GetResult(0); err != nil {
baseapi.SugarLogger.Debugf("LoadingStoreOrderSettleAmount :%v", err)
continue
}
}
}
func (t *TcpClient) handleConn(c net.Conn) error {
func handleConn(c net.Conn, t *TcpClient) error {
if c == nil {
globals.SugarLogger.Debugf("conn is nil")
return errors.New("conn is nil")
@@ -74,7 +98,6 @@ func (t *TcpClient) handleConn(c net.Conn) error {
c.Write([]byte("ok"))
return err
}
//data = "1e001a020001513132303232303931353030313935390000008a"
//证明是心跳
if strings.Contains(data, heartText) || strings.Contains(data, heartTextNew) {
printNoData, _ := hex.DecodeString(data[len(heartText) : len(data)-8])
@@ -143,9 +166,7 @@ func (t *TcpClient) changePrintMsg(data string, orderNo int64, printNo string) (
//1、先找出打印机编号和订单序列号，这两个确定唯一一条消息?
//通过参数传进来
//2、打印成功改变打印表的状态
globals.SugarLogger.Debugf("====callback Status %s", utils.Format4Output(data, false))
if strings.Contains(data, printSuccessText) || strings.Contains(data, printSuccessTextNew) {
//1e001802000150323032313036313530303030313000013c 成功消息例子
status = printMsgSuccess
comment = "回调成功,修改打印状态"
} else {
@@ -177,9 +198,10 @@ func (t *TcpClient) HandleTcpMessages(printNo string) {
if !t.isExistMsg(printNo) {
return
}
go func() {
for {
if t.TimeoutMap[printNo] == true {
if t.TimeoutMap[printNo] == true && time.Now().Unix()%3 == 0 {
timeNow := time.Now()
timeStart := time.Date(timeNow.Year(), timeNow.Month(), timeNow.Day(), 0, 0, 0, 0, timeNow.Location())
timeEnd := time.Date(timeNow.Year(), timeNow.Month(), timeNow.Day(), 23, 59, 59, 0, timeNow.Location())
@@ -204,7 +226,6 @@ func (t *TcpClient) HandleTcpMessages(printNo string) {
func (t *TcpClient) readTimeoutMap(key string) bool {
t.Lock()
defer t.Unlock()
//return <-t.TimeoutMap[key]
return t.TimeoutMap[key]
}
@@ -247,11 +268,9 @@ func (t *TcpClient) doPrint(key string) (err error) {
}
}
if err != nil {
//t.delConn(printMsg.PrintNo)
printMsg.Status = printMsgErr
printMsg.Comment = err.Error()
dao.UpdateEntity(db, printMsg, "Status", "Comment")
//close(t.TimeoutMap[key])
delete(t.TimeoutMap, key)
} else {
if c != nil {
@@ -308,95 +327,7 @@ func (t *TcpClient) doPrint(key string) (err error) {
globals.SugarLogger.Debugf("doPrint timeout")
return
}
//select {
//case <-t.TimeoutMap[key]:
// globals.SugarLogger.Debugf("doPrint timeout")
// return
//default:
// select {
// case printMsg, ok := <-t.MsgMap[key]:
// if !ok {
// globals.SugarLogger.Debugf("doPrint err !ok ...")
// return
// }
// var (
// data []byte
// c net.Conn
// )
// if printMsg != nil {
// if err = checkPrintMsg(db, printMsg); err == nil {
// status := t.getPrintStatus(printMsg.PrintNo)
// switch status {
// //只有在线才打印内容
// case printerStatusOnline:
// if c = t.getPrintConn(printMsg.PrintNo); c != nil {
// data, err = buildMsg(printMsg)
// }
// case printerStatusOffline:
// err = fmt.Errorf("打印机离线！")
// case printerStatusOnlineWithoutPaper:
// err = fmt.Errorf("打印机缺纸！")
// default:
// err = fmt.Errorf("打印机状态未知！")
// }
// }
// if err != nil {
// //t.delConn(printMsg.PrintNo)
// printMsg.Status = printMsgErr
// printMsg.Comment = err.Error()
// dao.UpdateEntity(db, printMsg, "Status", "Comment")
// close(t.TimeoutMap[key])
// } else {
// if c != nil {
// if _, err = c.Write(data); err != nil {
// globals.SugarLogger.Debugf("handleTcpMessages err [%v]", err)
// close(t.TimeoutMap[key])
// } else {
// //等待回调
// dataStr := <-t.CallBackMap[key]
// if dataStr != "" {
// a, b := getCallbackMsgInfo(dataStr)
// t.changePrintMsg(dataStr, a, b)
// // 查询打印机是否扣费,未扣费就扣费,已经扣费不做处理
// have, err := dao.QueryOrderDeductionRecord(db, b, utils.Int64ToStr(a))
// if err != nil && !have {
// // 扣除打印机账号金额
// if err = dao.DeductionPrintBalance(db, b); err != nil {
// globals.SugarLogger.Debugf("扣除用户打印机金额错误 %s", err)
// } else {
// // 添加打印记录(支出记录)
// if err = dao.AddPrintRecord(db, &model.PrintBillRecord{
// CreatedAt: time.Now(),
// UpdatedAt: time.Now(),
// PrintNo: b,
// PayType: 2,
// PayMoney: 1, // 固定支出一分钱
// OrderId: utils.Int64ToStr(a),
// UserId: "",
// }); err != nil {
// globals.SugarLogger.Debugf("添加打印机订单支付记录错误 %s", err)
// }
// }
// } else {
// globals.SugarLogger.Debugf("查询打印机扣费记录错误 %s", err)
// }
// //判断音频暂停？
// //收到打印成功回调后，如果消息中有音频，需要等待一下,等上一个音频播完
// //暂停时间就暂时取的sound标签内内容长度/2
// if sounds := regexpSoundSpan.FindStringSubmatch(printMsg.Content); len(sounds) > 0 {
// sound := sounds[1]
// lenTime := time.Duration(utf8.RuneCountInString(sound)) * time.Second
// time.Sleep(lenTime / 2)
// }
// }
// }
// }
// }
// } else {
// globals.SugarLogger.Debugf("msgMap is nil")
// }
// }
//}
}
}()
return err
@@ -412,8 +343,6 @@ func (t *TcpClient) HandleCheckTcpHeart(key string) {
//1分钟内没心跳判断打印机掉线了
if time.Now().Sub(statusTime) > time.Minute+time.Second*10 {
changePrinterStatus(key, printerStatusOffline)
//t.clear(key)
//close(t.TimeoutMap[key])
delete(t.TimeoutMap, key)
}
}
@@ -422,46 +351,8 @@ func (t *TcpClient) HandleCheckTcpHeart(key string) {
close(t.MsgMap[key])
close(t.CallBackMap[key])
t.delConn(key)
//delete(t.Clients, key)
return
}
//select {
//case <-t.TimeoutMap[key]:
// t.getClients(key).C.Close()
// close(t.MsgMap[key])
// close(t.CallBackMap[key])
// t.delConn(key)
// //delete(t.Clients, key)
// return
//default:
// statusTime := t.getPrintStatusTime(key)
// if !utils.IsTimeZero(statusTime) {
// //1分钟内没心跳判断打印机掉线了
// if time.Now().Sub(statusTime) > time.Minute+time.Second*10 {
// changePrinterStatus(key, printerStatusOffline)
// //t.clear(key)
// //close(t.TimeoutMap[key])
// delete(t.TimeoutMap,key)
// }
// }
//}
//keys := []string{}
//t.RLock()
//for k, v := range t.Clients {
// if time.Now().Sub(v.StatusTime) > time.Minute+time.Second {
// v.Status = printerStatusOffline
// keys = append(keys, k)
// }
//}
//t.RUnlock()
//globals.SugarLogger.Debugf("HandleCheckTcpHeart del keys: %v", keys)
//for _, v := range keys {
// changePrinterStatus(v, printerStatusOffline)
// if t.isExist(v) {
// t.clear(v)
// }
//}
}
}()
}