From 98a52a75596423e54feb5d3f2ef2ef82aadd2e63 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=82=B9=E5=AE=97=E6=A5=A0?= Date: Mon, 29 May 2023 15:21:20 +0800 Subject: [PATCH] 1 --- business/jxstore/event/event_tcp.go | 187 ++++++---------------------- 1 file changed, 39 insertions(+), 148 deletions(-) diff --git a/business/jxstore/event/event_tcp.go b/business/jxstore/event/event_tcp.go index 09f894f00..dbe0dbcf4 100644 --- a/business/jxstore/event/event_tcp.go +++ b/business/jxstore/event/event_tcp.go @@ -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) - // } - //} } }() }