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-03-16 15:54:18 +08:00
parent be1d4be19e
commit 9ebe42830f
2 changed files with 163 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -181,12 +181,7 @@ func (t *TcpClient) HandleTcpMessages(printNo string) {
} }
go func() { go func() {
for { for {
select { if t.TimeoutMap[printNo] == true {
case <-t.TimeoutMap[printNo]:
globals.SugarLogger.Debugf("HandleTcpMessages timeout")
return
default:
//一直读？
timeNow := time.Now() timeNow := time.Now()
timeStart := time.Date(timeNow.Year(), timeNow.Month(), timeNow.Day(), 0, 0, 0, 0, timeNow.Location()) 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()) timeEnd := time.Date(timeNow.Year(), timeNow.Month(), timeNow.Day(), 23, 59, 59, 0, timeNow.Location())
@@ -200,7 +195,30 @@ func (t *TcpClient) HandleTcpMessages(printNo string) {
} }
} }
} }
} else {
globals.SugarLogger.Debugf("HandleTcpMessages timeout")
return
} }
//select {
//case <-t.TimeoutMap[printNo]:
// globals.SugarLogger.Debugf("HandleTcpMessages timeout")
// return
//default:
// //一直读？
// 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)
// }
// }
// }
//}
} }
}() }()
} }
@@ -208,7 +226,8 @@ func (t *TcpClient) HandleTcpMessages(printNo string) {
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]
return t.TimeoutMap[key]
} }
func (t *TcpClient) doPrint(key string) (err error) { func (t *TcpClient) doPrint(key string) (err error) {
@@ -221,11 +240,7 @@ func (t *TcpClient) doPrint(key string) (err error) {
} }
go func() { go func() {
for { for {
select { if t.TimeoutMap[key] == true {
case <-t.TimeoutMap[key]:
globals.SugarLogger.Debugf("doPrint timeout")
return
default:
select { select {
case printMsg, ok := <-t.MsgMap[key]: case printMsg, ok := <-t.MsgMap[key]:
if !ok { if !ok {
@@ -258,12 +273,14 @@ func (t *TcpClient) doPrint(key string) (err error) {
printMsg.Status = printMsgErr printMsg.Status = printMsgErr
printMsg.Comment = err.Error() printMsg.Comment = err.Error()
dao.UpdateEntity(db, printMsg, "Status", "Comment") dao.UpdateEntity(db, printMsg, "Status", "Comment")
close(t.TimeoutMap[key]) //close(t.TimeoutMap[key])
delete(t.TimeoutMap, key)
} else { } else {
if c != nil { if c != nil {
if _, err = c.Write(data); err != nil { if _, err = c.Write(data); err != nil {
globals.SugarLogger.Debugf("handleTcpMessages err [%v]", err) globals.SugarLogger.Debugf("handleTcpMessages err [%v]", err)
close(t.TimeoutMap[key]) //close(t.TimeoutMap[key])
delete(t.TimeoutMap, key)
} else { } else {
//等待回调 //等待回调
dataStr := <-t.CallBackMap[key] dataStr := <-t.CallBackMap[key]
@@ -309,7 +326,99 @@ func (t *TcpClient) doPrint(key string) (err error) {
globals.SugarLogger.Debugf("msgMap is nil") globals.SugarLogger.Debugf("msgMap is nil")
} }
} }
} else {
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 return err
@@ -319,25 +428,46 @@ func (t *TcpClient) doPrint(key string) (err error) {
func (t *TcpClient) HandleCheckTcpHeart(key string) { func (t *TcpClient) HandleCheckTcpHeart(key string) {
go func() { go func() {
for { for {
select { if t.TimeoutMap[key] == true {
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) statusTime := t.getPrintStatusTime(key)
if !utils.IsTimeZero(statusTime) { if !utils.IsTimeZero(statusTime) {
//1分钟内没心跳判断打印机掉线了 //1分钟内没心跳判断打印机掉线了
if time.Now().Sub(statusTime) > time.Minute+time.Second*10 { if time.Now().Sub(statusTime) > time.Minute+time.Second*10 {
changePrinterStatus(key, printerStatusOffline) changePrinterStatus(key, printerStatusOffline)
//t.clear(key) //t.clear(key)
close(t.TimeoutMap[key]) //close(t.TimeoutMap[key])
delete(t.TimeoutMap, key)
} }
} }
} else {
t.getClients(key).C.Close()
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{} //keys := []string{}
//t.RLock() //t.RLock()
//for k, v := range t.Clients { //for k, v := range t.Clients {

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

@@ -130,7 +130,8 @@ type TcpClient struct {
Clients map[string]*PrintInfo //放tcp连接的，printNo 为key Clients map[string]*PrintInfo //放tcp连接的，printNo 为key
MsgMap map[string]chan *model.PrintMsg //放打印信息的，printNo为key MsgMap map[string]chan *model.PrintMsg //放打印信息的，printNo为key
CallBackMap map[string]chan string //放打印信息回调信息的，printNo为key CallBackMap map[string]chan string //放打印信息回调信息的，printNo为key
TimeoutMap map[string]chan bool //退出channel //TimeoutMap map[string]chan bool //退出channel
TimeoutMap map[string]bool //退出channel
*sync.RWMutex *sync.RWMutex
} }
@@ -177,14 +178,16 @@ func (t *TcpClient) buildCallBackMap(key string) {
func (t *TcpClient) buildTimeoutMap(key string) { func (t *TcpClient) buildTimeoutMap(key string) {
t.Lock() t.Lock()
defer t.Unlock() defer t.Unlock()
dataChan := make(chan bool) //dataChan := make(chan bool)
t.TimeoutMap[key] = dataChan //t.TimeoutMap[key] = dataChan
t.TimeoutMap[key] = true
} }
func (t *TcpClient) getTimeOut(key string) bool { func (t *TcpClient) getTimeOut(key string) bool {
t.RLock() t.RLock()
defer t.RUnlock() defer t.RUnlock()
return <-t.TimeoutMap[key] //return <-t.TimeoutMap[key]
return t.TimeoutMap[key]
} }
func (t *TcpClient) buildAllMap(key string) { func (t *TcpClient) buildAllMap(key string) {
@@ -192,7 +195,7 @@ func (t *TcpClient) buildAllMap(key string) {
defer t.Unlock() defer t.Unlock()
t.MsgMap[key] = make(chan *model.PrintMsg, 1024) t.MsgMap[key] = make(chan *model.PrintMsg, 1024)
t.CallBackMap[key] = make(chan string, 1024) t.CallBackMap[key] = make(chan string, 1024)
t.TimeoutMap[key] = make(chan bool, 1) t.TimeoutMap[key] = true
} }
func (t *TcpClient) getPrintStatus(key string) int { func (t *TcpClient) getPrintStatus(key string) int {
@@ -332,7 +335,7 @@ func NewTcpClient() *TcpClient {
Clients: make(map[string]*PrintInfo), Clients: make(map[string]*PrintInfo),
CallBackMap: make(map[string]chan string), CallBackMap: make(map[string]chan string),
MsgMap: make(map[string]chan *model.PrintMsg), MsgMap: make(map[string]chan *model.PrintMsg),
TimeoutMap: make(map[string]chan bool), TimeoutMap: make(map[string]bool, 0),
} }
t.RWMutex = new(sync.RWMutex) t.RWMutex = new(sync.RWMutex)
return t return t