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:
邹宗楠
2026-05-12 17:51:23 +08:00
parent eaa27cbaa7
commit de1c4d00de
3 changed files with 398 additions and 400 deletions
Download Patch File Download Diff File Expand all files Collapse all files

192
business/jxstore/event/event_print.go
View File

@@ -1,99 +1,97 @@
package event package event
//package event import (
// "fmt"
//import ( "sync"
// "fmt" )
// "sync"
//) var (
// PrintObject *PrintObjectStruct // 缓存的打印机对象
//var ( PrintAddrAndIp *PrintAddrAndIpStruct // 缓存打印机地址:[ip:printNo] event 文件包,connect只能获取到addr
// PrintObject *PrintObjectStruct // 缓存的打印机对象 PrintIpAndAddr *PrintIpAndAddrStruct // 缓存打印机地址:[printNo:ip] api_controller 只能获取到printNo
// PrintAddrAndIp *PrintAddrAndIpStruct // 缓存打印机地址:[ip:printNo] event 文件包,connect只能获取到addr )
// PrintIpAndAddr *PrintIpAndAddrStruct // 缓存打印机地址:[printNo:ip] api_controller 只能获取到printNo
//) func init() {
// fmt.Println("初始化打印机对象")
//func init() { PrintObject = &PrintObjectStruct{
// fmt.Println("初始化打印机对象") PrintObject: make(map[string]*TcpClient),
// PrintObject = &PrintObjectStruct{ RWMutex: new(sync.RWMutex),
// PrintObject: make(map[string]*TcpClient), }
// RWMutex: new(sync.RWMutex), PrintAddrAndIp = &PrintAddrAndIpStruct{
// } PrintObject: make(map[string]string),
// PrintAddrAndIp = &PrintAddrAndIpStruct{ RWMutex: new(sync.RWMutex),
// PrintObject: make(map[string]string), }
// RWMutex: new(sync.RWMutex), PrintIpAndAddr = &PrintIpAndAddrStruct{
// } PrintObject: make(map[string]string),
// PrintIpAndAddr = &PrintIpAndAddrStruct{ RWMutex: new(sync.RWMutex),
// PrintObject: make(map[string]string), }
// RWMutex: new(sync.RWMutex), }
// }
//} type PrintObjectStruct struct {
// PrintObject map[string]*TcpClient
//type PrintObjectStruct struct { *sync.RWMutex
// PrintObject map[string]*TcpClient }
// *sync.RWMutex
//} func (p *PrintObjectStruct) GetPrintObj(printNo string) (*TcpClient, bool) {
// p.RLock()
//func (p *PrintObjectStruct) GetPrintObj(printNo string) (*TcpClient, bool) { defer p.RUnlock()
// p.RLock() tcpObj, ok := PrintObject.PrintObject[printNo]
// defer p.RUnlock() return tcpObj, ok
// tcpObj, ok := PrintObject.PrintObject[printNo] }
// return tcpObj, ok func (p *PrintObjectStruct) SetPrintObj(printNo string, tcpObj *TcpClient) {
//} p.RLock()
//func (p *PrintObjectStruct) SetPrintObj(printNo string, tcpObj *TcpClient) { defer p.RUnlock()
// p.RLock() PrintObject.PrintObject[printNo] = tcpObj
// defer p.RUnlock() }
// PrintObject.PrintObject[printNo] = tcpObj
//} func (p *PrintObjectStruct) DelPrintObj(printNo string) {
// p.RLock()
//func (p *PrintObjectStruct) DelPrintObj(printNo string) { defer p.RUnlock()
// p.RLock() delete(PrintObject.PrintObject, printNo)
// defer p.RUnlock() }
// delete(PrintObject.PrintObject, printNo)
//} type PrintAddrAndIpStruct struct {
// PrintObject map[string]string
//type PrintAddrAndIpStruct struct { *sync.RWMutex
// PrintObject map[string]string }
// *sync.RWMutex
//} func (p *PrintAddrAndIpStruct) GetPrintAddrAndIp(ip string) (string, bool) {
// p.RLock()
//func (p *PrintAddrAndIpStruct) GetPrintAddrAndIp(ip string) (string, bool) { defer p.RUnlock()
// p.RLock() printNo, ok := PrintAddrAndIp.PrintObject[ip]
// defer p.RUnlock() return printNo, ok
// printNo, ok := PrintAddrAndIp.PrintObject[ip] }
// return printNo, ok func (p *PrintAddrAndIpStruct) SetPrintAddrAndIp(ip string, printNo string) {
//} p.RLock()
//func (p *PrintAddrAndIpStruct) SetPrintAddrAndIp(ip string, printNo string) { defer p.RUnlock()
// p.RLock() PrintAddrAndIp.PrintObject[ip] = printNo
// defer p.RUnlock() }
// PrintAddrAndIp.PrintObject[ip] = printNo
//} func (p *PrintAddrAndIpStruct) DelPrintAddrAndIp(ip string) {
// p.RLock()
//func (p *PrintAddrAndIpStruct) DelPrintAddrAndIp(ip string) { defer p.RUnlock()
// p.RLock() delete(PrintAddrAndIp.PrintObject, ip)
// defer p.RUnlock() }
// delete(PrintAddrAndIp.PrintObject, ip)
//} type PrintIpAndAddrStruct struct {
// PrintObject map[string]string
//type PrintIpAndAddrStruct struct { *sync.RWMutex
// PrintObject map[string]string }
// *sync.RWMutex
//} func (p *PrintIpAndAddrStruct) GetPrintIpAndAddr(printNo string) (string, bool) {
// p.RLock()
//func (p *PrintIpAndAddrStruct) GetPrintIpAndAddr(printNo string) (string, bool) { defer p.RUnlock()
// p.RLock() tcpObj, ok := PrintIpAndAddr.PrintObject[printNo]
// defer p.RUnlock() return tcpObj, ok
// tcpObj, ok := PrintIpAndAddr.PrintObject[printNo] }
// return tcpObj, ok func (p *PrintIpAndAddrStruct) SetPrintIpAndAddr(printNo string, ip string) {
//} p.RLock()
//func (p *PrintIpAndAddrStruct) SetPrintIpAndAddr(printNo string, ip string) { defer p.RUnlock()
// p.RLock() PrintIpAndAddr.PrintObject[printNo] = ip
// defer p.RUnlock() }
// PrintIpAndAddr.PrintObject[printNo] = ip
//} func (p *PrintIpAndAddrStruct) DelPrintIpAndAddr(printNo string) {
// p.RLock()
//func (p *PrintIpAndAddrStruct) DelPrintIpAndAddr(printNo string) { defer p.RUnlock()
// p.RLock() delete(PrintIpAndAddr.PrintObject, printNo)
// defer p.RUnlock() }
// delete(PrintIpAndAddr.PrintObject, printNo)
//}

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

@@ -4,12 +4,15 @@ import (
"encoding/hex" "encoding/hex"
"errors" "errors"
"fmt" "fmt"
"git.rosy.net.cn/baseapi/utils"
"git.rosy.net.cn/jx-callback/business/model"
"git.rosy.net.cn/jx-callback/business/model/dao" "git.rosy.net.cn/jx-callback/business/model/dao"
"git.rosy.net.cn/jx-print/globals" "git.rosy.net.cn/jx-callback/globals"
"io" "io"
"net" "net"
"strings" "strings"
"time" "time"
"unicode/utf8"
) )
// ConnRead 获取链接数据 // ConnRead 获取链接数据
@@ -43,7 +46,7 @@ func ListenTcp() {
} }
}() }()
if err = handleConn(c); err != nil { if err := handleConn(c); err != nil {
c.Close() c.Close()
Pool.Wait() Pool.Wait()
Pool.Stop() Pool.Stop()
@@ -60,20 +63,33 @@ func handleConn(c net.Conn) error {
} }
for { for {
buffer, n, err := ConnRead(c) buffer, n, err := ConnRead(c)
printRemoteAddr := c.RemoteAddr().String()
//remoteAddr := c.RemoteAddr().(*net.TCPAddr) printRemoteAddr = strings.Split(printRemoteAddr, ":")[0]
//remoteIP := remoteAddr.IP.String() // 打印机IP printNoByIP, _ := PrintAddrAndIp.GetPrintAddrAndIp(printRemoteAddr)
//remotePort := remoteAddr.Port // 打印机端口
//globals.SugarLogger.Debugf("remoteIP1: %s", remoteIP)
//globals.SugarLogger.Debugf("remotePort1: %d", remotePort)
// 重连
if err != nil { if err != nil {
if err == io.EOF { if err == io.EOF {
fmt.Println("connection close") fmt.Println("connection close")
} else { } else {
fmt.Println("ReadString err:", err) fmt.Println("ReadString err:", err)
} }
globals.SugarLogger.Debugf("--------printRemoteAddr := %s,printNo := %s", printRemoteAddr, printNoByIP)
if printNo, ok := PrintAddrAndIp.GetPrintAddrAndIp(printRemoteAddr); ok {
PrintAddrAndIp.DelPrintAddrAndIp(printRemoteAddr)
PrintObject.DelPrintObj(printNo)
PrintIpAndAddr.DelPrintIpAndAddr(printRemoteAddr)
dao.ExecuteSQL(dao.GetDB(), `UPDATE printer SET status = -1,is_online = -1 WHERE print_no = ? `, []interface{}{printNo}...)
} else {
printStatusOff := make(map[string]int, 0)
for ip, pn := range PrintAddrAndIp.PrintObject {
if ip == printRemoteAddr {
PrintAddrAndIp.DelPrintAddrAndIp(printRemoteAddr)
PrintIpAndAddr.DelPrintIpAndAddr(pn)
PrintObject.DelPrintObj(printNo)
} else if pn != "" {
printStatusOff[pn] = 1
}
}
}
return err return err
} }
@@ -83,7 +99,6 @@ func handleConn(c net.Conn) error {
printNo string = "" //打印机编号 printNo string = "" //打印机编号
heartbeat bool = false heartbeat bool = false
callback bool = false callback bool = false
t *TcpClient = nil
) )
if strings.Contains(data, heartText) || strings.Contains(data, heartTextNew) { if strings.Contains(data, heartText) || strings.Contains(data, heartTextNew) {
printNoData, _ := hex.DecodeString(data[len(heartText) : len(data)-8]) printNoData, _ := hex.DecodeString(data[len(heartText) : len(data)-8])
@@ -93,19 +108,15 @@ func handleConn(c net.Conn) error {
_, printNo = getCallbackMsgInfo(data) _, printNo = getCallbackMsgInfo(data)
callback = true callback = true
} }
globals.SugarLogger.Debugf("printNoData------: %s", printNo)
t, ok := PrintObject.GetPrintObj(printNo)
if !ok || t.Clients[printNo] == nil || time.Now().Sub(t.Clients[printNo].StatusTime).Seconds() >= 120 {
t = NewTcpClient() t = NewTcpClient()
TcpClientList.Delete(printNo) }
TcpClientList.Store(printNo, t)
//if value, have := TcpClientList.Load(printNo); !have {
// TcpClientList.Store(printNo, t)
//} else {
// t = value.(*TcpClient)
//}
if heartbeat { if heartbeat {
// 证明是心跳 // 证明是心跳
Heartbeat(c, t, data, printNo, "") Heartbeat(c, t, data, printNo, printRemoteAddr)
} else if callback { } else if callback {
// 打印回调 // 打印回调
Callback(c, t, data, printNo) Callback(c, t, data, printNo)
@@ -125,8 +136,238 @@ func (t *TcpClient) printFail() (err error) {
return err return err
} }
func (t *TcpClient) changePrintMsg(data string, orderNo int64, printNo string) (err error) {
var (
db = dao.GetDB()
comment string
status int
)
//1、先找出打印机编号和订单序列号，这两个确定唯一一条消息?
//通过参数传进来
//2、打印成功改变打印表的状态
if strings.Contains(data, printSuccessText) || strings.Contains(data, printSuccessTextNew) {
status = printMsgSuccess
comment = "回调成功,修改打印状态"
} else {
//打印失败也改变状态并更新失败原因
status = printMsgFail
comment = printErrMap[data[12:14]]
}
//这里序号重复会有问题
if printMsgs, err := dao.GetPrintMsgNoPage(db, printNo, orderNo); err != nil {
globals.SugarLogger.Debugf("changePrintMsg err :[%v]", err)
return err
} else if len(printMsgs) == 0 {
globals.SugarLogger.Debugf("changePrintMsg err ,not found printMsg printNo:[%v], orderNo :[%v]", printNo, orderNo)
} else if len(printMsgs) > 0 {
for _, v := range printMsgs {
v.Comment = comment
v.Status = status
dao.UpdateEntity(db, v, "Comment", "Status")
}
}
return err
}
func HandleTcpMessages(t *TcpClient, printNo string) {
var (
db = dao.GetDB()
offset, pageSize = 0, 10
)
if !t.isExistMsg(printNo) {
return
}
fn := func() {
//for {
// time.Sleep(2 * time.Second)
if t.TimeoutMap[printNo] == true {
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())
prints, _ := dao.GetPrintMsgs(db, printNo, []int{PrintMsgWait}, timeStart.AddDate(0, 0, -1), timeEnd, offset, pageSize)
for _, printMsg := range prints {
printMsg.Status = PrintMsgAlreadyLoad
//先避免重复读再插到channel？
if _, err := dao.UpdateEntity(db, printMsg, "Status"); err == nil {
if err = t.addMsgChan(printMsg); err != nil {
globals.SugarLogger.Debugf("HandleTcpMessages addMsgChan Err: %v", err)
}
}
}
} else {
globals.SugarLogger.Debugf("HandleTcpMessages timeout")
return
}
}
Pool.AddJob(fn)
}
func (t *TcpClient) readTimeoutMap(key string) bool { func (t *TcpClient) readTimeoutMap(key string) bool {
t.Lock() t.Lock()
defer t.Unlock() defer t.Unlock()
return t.TimeoutMap[key] return t.TimeoutMap[key]
} }
func doPrint(t *TcpClient, key string) (err error) {
var (
db = dao.GetDB()
)
if !t.isExistMsg(key) {
return err
}
fn := func() {
for {
if t.TimeoutMap[key] == true {
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 {
globals.SugarLogger.Debugf("print msg is nil")
continue
}
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 c == nil {
if printRemoteAddrIP, have := PrintIpAndAddr.GetPrintIpAndAddr(key); have {
PrintIpAndAddr.DelPrintIpAndAddr(key)
PrintAddrAndIp.DelPrintAddrAndIp(printRemoteAddrIP)
PrintObject.DelPrintObj(key)
}
return
}
if err != nil {
printMsg.Status = printMsgErr
printMsg.Comment = err.Error()
dao.UpdateEntity(db, printMsg, "Status", "Comment")
if printRemoteAddrIP, have := PrintIpAndAddr.GetPrintIpAndAddr(key); have {
PrintIpAndAddr.DelPrintIpAndAddr(key)
PrintAddrAndIp.DelPrintAddrAndIp(printRemoteAddrIP)
PrintObject.DelPrintObj(key)
}
return
}
if _, err = c.Write(data); err != nil {
globals.SugarLogger.Debugf("handleTcpMessages err [%v]", err)
if printRemoteAddrIP, have := PrintIpAndAddr.GetPrintIpAndAddr(key); have {
PrintIpAndAddr.DelPrintIpAndAddr(key)
PrintAddrAndIp.DelPrintAddrAndIp(printRemoteAddrIP)
PrintObject.DelPrintObj(key)
}
} else {
//等待回调
dataStr := <-t.CallBackMap[key]
if dataStr != "" {
a, b := getCallbackMsgInfo(dataStr)
t.changePrintMsg(dataStr, a, b)
// 查询打印机是否扣费,未扣费就扣费,已经扣费不做处理
have, err2 := dao.QueryOrderDeductionRecord(db, b, utils.Int64ToStr(a))
if err2 == 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("今天已经扣除过了! %v %d %s", err2, a, b)
}
// 回调重置打印机状态时间
t.Clients[b].StatusTime = time.Now()
//判断音频暂停？
//收到打印成功回调后，如果消息中有音频，需要等待一下,等上一个音频播完
//暂停时间就暂时取的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("doPrint timeout")
return
}
}
}
Pool.AddJob(fn)
return err
}
// HandleCheckTcpHeart 检测心跳
func HandleCheckTcpHeart(t *TcpClient, key string) {
if t.TimeoutMap[key] == true {
statusTime := t.getPrintStatusTime(key)
if !utils.IsTimeZero(statusTime) {
//1分钟内没心跳判断打印机掉线了
if time.Now().Sub(statusTime) > time.Second*75 {
globals.SugarLogger.Debugf("超过一分十秒没有心跳的打印机[%s],当前心跳时间: %s ,上一次心跳时间 : %s", key, utils.Time2TimeStr(time.Now()), utils.Time2TimeStr(statusTime))
changePrinterStatus(key, printerStatusOffline)
// 链接出错,彻底删除换成
if printRemoteAddrIP, have := PrintIpAndAddr.GetPrintIpAndAddr(key); have {
PrintIpAndAddr.DelPrintIpAndAddr(key)
PrintAddrAndIp.DelPrintAddrAndIp(printRemoteAddrIP)
PrintObject.DelPrintObj(key)
}
}
}
} else {
t.getClients(key).C.Close()
close(t.MsgMap[key])
close(t.CallBackMap[key])
//t.delConn(key)
t.clear(key)
// 链接出错,彻底删除换成
if printRemoteAddrIP, have := PrintIpAndAddr.GetPrintIpAndAddr(key); have {
PrintIpAndAddr.DelPrintIpAndAddr(key)
PrintAddrAndIp.DelPrintAddrAndIp(printRemoteAddrIP)
PrintObject.DelPrintObj(key)
}
return
}
}

317
business/jxstore/event/event_tcp_utils.go
View File

@@ -8,18 +8,14 @@ import (
"git.rosy.net.cn/jx-callback/business/model" "git.rosy.net.cn/jx-callback/business/model"
"git.rosy.net.cn/jx-callback/business/model/dao" "git.rosy.net.cn/jx-callback/business/model/dao"
"git.rosy.net.cn/jx-callback/globals" "git.rosy.net.cn/jx-callback/globals"
"log"
"net" "net"
"regexp" "regexp"
"strconv" "strconv"
"strings" "strings"
"sync" "sync"
"time" "time"
"unicode/utf8"
) )
var TcpClientList = new(sync.Map)
const ( const (
heartText = "1e000f02000151" // 老版心跳 heartText = "1e000f02000151" // 老版心跳
heartTextNew = "1e001a02000151" // 新版心跳 heartTextNew = "1e001a02000151" // 新版心跳
@@ -153,6 +149,7 @@ func (t *TcpClient) delConn(key string) {
t.Lock() t.Lock()
defer t.Unlock() defer t.Unlock()
if t.Clients[key].C != nil { if t.Clients[key].C != nil {
globals.SugarLogger.Debugf("-------close2 := %s", key)
t.Clients[key].C.Close() t.Clients[key].C.Close()
} }
delete(t.Clients, key) delete(t.Clients, key)
@@ -241,11 +238,6 @@ func (t *TcpClient) getPrintConn(key string) net.Conn {
return nil return nil
} }
} }
func (t *TcpClient) setPrintConn(key string, c net.Conn) {
t.RLock()
defer t.RUnlock()
t.Clients[key].C = c
}
func (t *TcpClient) getPrintStatusTime(key string) time.Time { func (t *TcpClient) getPrintStatusTime(key string) time.Time {
t.RLock() t.RLock()
@@ -402,6 +394,38 @@ func changePrinterStatus(printNo string, status int) {
} }
} }
//按打印机方提供的文档来的
func buildMsg(printMsg *model.PrintMsg) (data []byte, err error) {
var (
content = printMsg.Content
orderNo = printMsg.OrderNo
str = "1e"
const1 = "0200ff50"
printInit = "1b40" //打印机初始化
//voice = "1d6b401dfd001a01015b7631365d736f756e64622cc4fad3d0d0c2b6a9b5a5c0b1" //语音，中国
//qr = "1d58021b5a0001061600747470733a2f2f7777772e62616964752e636f6d2f1b000A0A0A1B40"
orderNoHexH, orderNoHexL, printData string
)
//写入数据
no, err := strconv.ParseInt(orderNo, 10, 64)
if err != nil {
globals.SugarLogger.Debug("order_msg Order_no 转换异常")
}
orderNoHexH, orderNoHexL = int2h8l8(no)
// 将数据与模板组装
if strings.Contains(content, "•") {
content = strings.ReplaceAll(content, "•", "-")
}
printDataGBK, _ := jxutils.Utf8ToGbk([]byte(utils.FilterEmoji(content)))
printData = hex.EncodeToString(printDataGBK)
printData = replaceContent(printData, printMsg)
lenData := int64(len(str) + len(const1) + len(orderNoHexH) + len(orderNoHexL) + len(printInit) + 2 + 4 + len(printData))
x1, x2 := int2h8l8(lenData / 2)
dataStr := str + x1 + x2 + const1 + orderNoHexH + orderNoHexL + printInit + printData
check := getCheckSum(dataStr)
return jxutils.Hextob(dataStr + check), err
}
func getCheckSum(str string) (check string) { func getCheckSum(str string) (check string) {
var sum int64 var sum int64
for i := 0; i < len(str); i = i + 2 { for i := 0; i < len(str); i = i + 2 {
@@ -623,7 +647,7 @@ func Heartbeat(c net.Conn, t *TcpClient, data string, printNo string, printRemot
//4、读打印channel并打印，并切等待回调channel中的消息 //4、读打印channel并打印，并切等待回调channel中的消息
//5、修改数据库中打印机状态（没在连接池中说明是重新连接的） //5、修改数据库中打印机状态（没在连接池中说明是重新连接的）
//6、监听心跳时间，超过1分多钟就clear掉 //6、监听心跳时间，超过1分多钟就clear掉
if t.getClients(printNo) == nil || t.getPrintStatusTime(printNo).IsZero() || time.Now().Sub(t.Clients[printNo].StatusTime).Seconds() >= 120 { if t.getClients(printNo) == nil || t == nil || t.getPrintStatusTime(printNo).IsZero() || time.Now().Sub(t.Clients[printNo].StatusTime).Seconds() >= 120 {
addConn(c, t, printNo, status) addConn(c, t, printNo, status)
buildAllMap(t, printNo) buildAllMap(t, printNo)
//t.TimeoutMap[printNo] <- true //t.TimeoutMap[printNo] <- true
@@ -639,6 +663,10 @@ func Heartbeat(c net.Conn, t *TcpClient, data string, printNo string, printRemot
if err := dao.NotExistsCreate(printNo); err != nil { if err := dao.NotExistsCreate(printNo); err != nil {
globals.SugarLogger.Debugf("监听打印机心跳,不存在则创建 :[%v],printNo[%s]", err, printNo) globals.SugarLogger.Debugf("监听打印机心跳,不存在则创建 :[%v],printNo[%s]", err, printNo)
} }
PrintObject.SetPrintObj(printNo, t)
PrintAddrAndIp.SetPrintAddrAndIp(printRemoteAddr, printNo)
PrintIpAndAddr.SetPrintIpAndAddr(printNo, printRemoteAddr)
} else { } else {
//在加到连接池中已经更新了时间，所以放在else里 //在加到连接池中已经更新了时间，所以放在else里
t.setPrintStatusTime(printNo) t.setPrintStatusTime(printNo)
@@ -657,272 +685,3 @@ func Callback(c net.Conn, t *TcpClient, data string, printNo string) {
t.setPrintStatusTime(printNo) t.setPrintStatusTime(printNo)
t.addCallbackChan(printNo, data) t.addCallbackChan(printNo, data)
} }
func HandleTcpMessages(t *TcpClient, printNo string) {
var (
db = dao.GetDB()
offset, pageSize = 0, 10
)
if !t.isExistMsg(printNo) {
return
}
fn := func() {
//for {
// time.Sleep(2 * time.Second)
if t.TimeoutMap[printNo] == true {
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())
prints, _ := dao.GetPrintMsgs(db, printNo, []int{PrintMsgWait}, timeStart.AddDate(0, 0, -1), timeEnd, offset, pageSize)
for _, printMsg := range prints {
printMsg.Status = PrintMsgAlreadyLoad
//先避免重复读再插到channel？
if _, err := dao.UpdateEntity(db, printMsg, "Status"); err == nil {
if err = t.addMsgChan(printMsg); err != nil {
globals.SugarLogger.Debugf("HandleTcpMessages addMsgChan Err: %v", err)
}
}
}
} else {
globals.SugarLogger.Debugf("HandleTcpMessages timeout")
return
}
}
Pool.AddJob(fn)
}
func doPrint(t *TcpClient, key string) (err error) {
var (
db = dao.GetDB()
)
if !t.isExistMsg(key) {
return err
}
fn := func() {
for {
if t.TimeoutMap[key] == true {
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 {
globals.SugarLogger.Debugf("print msg is nil")
continue
}
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 c == nil {
return
}
if err != nil {
printMsg.Status = printMsgErr
printMsg.Comment = err.Error()
dao.UpdateEntity(db, printMsg, "Status", "Comment")
return
}
if _, err = c.Write(data); err != nil {
globals.SugarLogger.Debugf("handleTcpMessages err [%v]", err)
} else {
//等待回调
dataStr := <-t.CallBackMap[key]
if dataStr != "" {
a, b := getCallbackMsgInfo(dataStr)
t.changePrintMsg(dataStr, a, b)
// 查询打印机是否扣费,未扣费就扣费,已经扣费不做处理
have, err2 := dao.QueryOrderDeductionRecord(db, b, utils.Int64ToStr(a))
if err2 == 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("今天已经扣除过了! %v %d %s", err2, a, b)
}
// 回调重置打印机状态时间
t.Clients[b].StatusTime = time.Now()
//判断音频暂停？
//收到打印成功回调后，如果消息中有音频，需要等待一下,等上一个音频播完
//暂停时间就暂时取的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("doPrint timeout")
return
}
}
}
Pool.AddJob(fn)
return err
}
//按打印机方提供的文档来的
func buildMsg(printMsg *model.PrintMsg) (data []byte, err error) {
var (
content = printMsg.Content
orderNo = printMsg.OrderNo
str = "1e"
const1 = "0200ff50"
printInit = "1b40" //打印机初始化
//voice = "1d6b401dfd001a01015b7631365d736f756e64622cc4fad3d0d0c2b6a9b5a5c0b1" //语音，中国
//qr = "1d58021b5a0001061600747470733a2f2f7777772e62616964752e636f6d2f1b000A0A0A1B40"
orderNoHexH, orderNoHexL, printData string
)
//写入数据
no, err := strconv.ParseInt(orderNo, 10, 64)
if err != nil {
globals.SugarLogger.Debug("order_msg Order_no 转换异常")
}
orderNoHexH, orderNoHexL = int2h8l8(no)
// 将数据与模板组装
if strings.Contains(content, "•") {
content = strings.ReplaceAll(content, "•", "-")
}
printDataGBK, _ := jxutils.Utf8ToGbk([]byte(utils.FilterEmoji(content)))
printData = hex.EncodeToString(printDataGBK)
printData = replaceContent(printData, printMsg)
lenData := int64(len(str) + len(const1) + len(orderNoHexH) + len(orderNoHexL) + len(printInit) + 2 + 4 + len(printData))
x1, x2 := int2h8l8(lenData / 2)
dataStr := str + x1 + x2 + const1 + orderNoHexH + orderNoHexL + printInit + printData
check := getCheckSum(dataStr)
return jxutils.Hextob(dataStr + check), err
}
// HandleCheckTcpHeart 检测心跳
func HandleCheckTcpHeart(t *TcpClient, key string) {
if t.TimeoutMap[key] == true {
statusTime := t.getPrintStatusTime(key)
if !utils.IsTimeZero(statusTime) {
//1分钟内没心跳判断打印机掉线了
if time.Now().Sub(statusTime) > time.Second*75 {
globals.SugarLogger.Debugf("超过一分十秒没有心跳的打印机[%s],当前心跳时间: %s ,上一次心跳时间 : %s", key, utils.Time2TimeStr(time.Now()), utils.Time2TimeStr(statusTime))
changePrinterStatus(key, printerStatusOffline)
}
}
} else {
t.getClients(key).C.Close()
close(t.MsgMap[key])
close(t.CallBackMap[key])
//t.delConn(key)
t.clear(key)
TcpClientList.Delete(key)
return
}
}
func (t *TcpClient) changePrintMsg(data string, orderNo int64, printNo string) (err error) {
var (
db = dao.GetDB()
comment string
status int
)
//1、先找出打印机编号和订单序列号，这两个确定唯一一条消息?
//通过参数传进来
//2、打印成功改变打印表的状态
if strings.Contains(data, printSuccessText) || strings.Contains(data, printSuccessTextNew) {
status = printMsgSuccess
comment = "回调成功,修改打印状态"
} else {
//打印失败也改变状态并更新失败原因
status = printMsgFail
comment = printErrMap[data[12:14]]
}
//这里序号重复会有问题
if printMsgs, err := dao.GetPrintMsgNoPage(db, printNo, orderNo); err != nil {
globals.SugarLogger.Debugf("changePrintMsg err :[%v]", err)
return err
} else if len(printMsgs) == 0 {
globals.SugarLogger.Debugf("changePrintMsg err ,not found printMsg printNo:[%v], orderNo :[%v]", printNo, orderNo)
} else if len(printMsgs) > 0 {
for _, v := range printMsgs {
v.Comment = comment
v.Status = status
dao.UpdateEntity(db, v, "Comment", "Status")
}
}
return err
}
// 工具函数：判断连接是否关闭
func isClosed(conn net.Conn) bool {
buffer := make([]byte, 0)
// 设置读取超时=极短时间，不阻塞
conn.SetReadDeadline(time.Now().Add(1 * time.Millisecond))
_, err := conn.Read(buffer)
conn.SetReadDeadline(time.Time{}) // 重置超时
if err != nil {
if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
return false // 只是超时，连接正常
}
}
return true // 连接已关闭
}
var mu sync.Mutex
// 自动重连
func reconnectPrinter(printerConn net.Conn) net.Conn {
mu.Lock()
defer mu.Unlock()
if printerConn != nil {
printerConn.Close()
}
// 重试 5 次
for i := 0; i < 5; i++ {
conn, err := net.DialTimeout("tcp", "打印机IP:端口", 3*time.Second)
if err == nil {
printerConn = conn
log.Println("打印机重连成功")
return printerConn
}
time.Sleep(2 * time.Second)
}
log.Println("打印机重连失败")
return nil
}