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:
richboo111
2023-05-11 18:21:09 +08:00
parent 6d314dbacc
commit 918485a7c4
6 changed files with 78 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
business/jxstore/misc/misc.go
View File

@@ -2,13 +2,14 @@ package misc
import (
"fmt"
"sync"
"time"
"git.rosy.net.cn/baseapi/platformapi/dingdingapi"
"git.rosy.net.cn/jx-callback/business/bidding"
"git.rosy.net.cn/jx-callback/business/jxcallback/auto_delivery"
"git.rosy.net.cn/jx-callback/business/jxutils/ddmsg"
"git.rosy.net.cn/jx-callback/business/partner/delivery"
"sync"
"time"
"git.rosy.net.cn/jx-callback/business/partner/purchase/jdshop"
@@ -261,11 +262,11 @@ func Init() {
}, 5*time.Second, 10*time.Minute)
// 抖音更新门店商品 ,接口收费暂不使用
if beego.BConfig.RunMode != "jxgy" {
ScheduleTimerFunc("RefreshSyncSkuList", func() {
syncStoreSkuTiktok()
}, []string{"13:00:00"})
}
//if beego.BConfig.RunMode != "jxgy" {
// ScheduleTimerFunc("RefreshSyncSkuList", func() {
// syncStoreSkuTiktok()
// }, []string{"13:00:00"})
//}
//刷新抖音门店token
if beego.BConfig.RunMode != "jxgy" {

1
business/model/dao/store.go
View File

@@ -1653,7 +1653,6 @@ func GetStoreBaseByVendorStoreID(vendorStoreID string, vendorID int) (storeDetai
return nil, errors.New("vendorStoreID不能为空")
}
//DefaultTimeValue := utils.Str2Time("1970-01-01 00:00:00")
sql := `SELECT t.* FROM store t WHERE t.id = (SELECT s.store_id FROM store_map s WHERE s.vendor_store_id= ? AND s.vendor_id= ? AND s.deleted_at= '1970-01-01 00:00:00' )`
if err := GetRow(GetDB(), &storeDetail, sql, []interface{}{vendorStoreID, vendorID}); err != nil {
return nil, err

105
business/partner/purchase/im/im.go
View File

@@ -44,6 +44,7 @@ func SendToVendor(msg []byte) {
return
}
Send(temp, sendData.Data.(map[string]interface{})["app_id"])
return
}
if sendData.VendorID == VendorIDELM {
param := sendData.Data.(ebaiapi.BusinessSendMsgReq)
@@ -53,18 +54,6 @@ func SendToVendor(msg []byte) {
}
}
//if err != nil {
// ClientRender(Fail, FailMsg, map[string]string{
// "errMsg": fmt.Sprintf("%v", err),
// })
// return err
//} else {
// ClientRender(SuccessCode, SuccessMsg, map[string]interface{}{
// "vendorID": sendData.VendorID,
// "msg": "ok",
// })
// return nil
//}
return
}
@@ -92,16 +81,16 @@ func Send(data []byte, appID interface{}) {
for {
_, msg, err := conn.ReadMessage()
temp := string(msg)
res := JsonCommon(HeartSuccessWord)
fmt.Printf("Send %s receive: %s\n", conn.RemoteAddr(), string(msg))
if err != nil {
break
} else if temp == res {
continue
} else {
if temp != HeartSuccessWord {
ReadMsgFromVendor(VendorIDMT, "", msg)
}
ReadMsgFromVendor(VendorIDMT, "", msg)
}
fmt.Printf("%s receive: %s\n", conn.RemoteAddr(), string(msg))
}
return
}
// MtInit 发送心跳
@@ -135,53 +124,57 @@ func MtInit() {
for {
_, msg, err := conn.ReadMessage()
temp := string(msg)
if err != nil || temp != "HB" {
res := JsonCommon(HeartCheckSuccess)
fmt.Printf("MtInit %s receive: %s\n", conn.RemoteAddr(), string(msg))
if err != nil {
break
} else if temp == res {
continue
} else {
ReadMsgFromVendor(VendorIDMT, "", msg)
}
fmt.Printf("%s receive: %s\n", conn.RemoteAddr(), string(msg))
}
}, url.ClientIDMain)
//副连接
if url.UrlSub != "" {
jxutils.CallMsgHandlerAsync(func() {
connSub, respSub, errSub := websocket.DefaultDialer.Dial(url.UrlSub, nil)
if errSub != nil || respSub.StatusCode != 101 {
fmt.Printf("连接失败：%v http响应不成功", errSub)
}
//关闭
defer func(conn *websocket.Conn) {
err := conn.Close()
if err != nil {
return
}
}(connSub)
//client连接事件
client := NewClient(url.ClientIDSub, connSub, ClientTypeMt)
Manager.Connect <- client
errSub = connSub.WriteMessage(websocket.TextMessage, data)
if errSub != nil {
fmt.Println(errSub)
}
for {
_, msg, err := connSub.ReadMessage()
temp := string(msg)
if err != nil || temp != HeartCheckSuccess {
break
} else {
ReadMsgFromVendor(VendorIDMT, "", msg)
}
fmt.Printf("%s connSub:receive: %s\n", connSub.RemoteAddr(), string(msg))
}
}, url.ClientIDSub)
}
//if url.UrlSub != "" {
// jxutils.CallMsgHandlerAsync(func() {
// connSub, respSub, errSub := websocket.DefaultDialer.Dial(url.UrlSub, nil)
// if errSub != nil || respSub.StatusCode != 101 {
// fmt.Printf("连接失败：%v http响应不成功", errSub)
// }
//
// //关闭
// defer func(conn *websocket.Conn) {
// err := conn.Close()
// if err != nil {
// return
// }
// }(connSub)
//
// //client连接事件
// client := NewClient(url.ClientIDSub, connSub, ClientTypeMt)
// Manager.Connect <- client
//
// errSub = connSub.WriteMessage(websocket.TextMessage, data)
// if errSub != nil {
// fmt.Println(errSub)
// }
//
// for {
// _, msg, err := connSub.ReadMessage()
// temp := string(msg)
// res := JsonCommon(HeartCheckSuccess)
// if err != nil || temp == res {
// break
// } else {
// ReadMsgFromVendor(VendorIDMT, "", msg)
// }
// fmt.Printf("MtInit %s connSub:receive: %s\n", connSub.RemoteAddr(), string(msg))
// }
//
// }, url.ClientIDSub)
//}
}

9
business/partner/purchase/im/im_model.go
View File

@@ -153,7 +153,7 @@ var (
CommonSetting = &commonConf{}
GlobalSetting = &global{}
//心跳相关
heartbeatInterval = 60 * time.Second // 心跳间隔
heartbeatInterval = 20 * time.Second // 心跳间隔
HeartCheckMsg = "~#HHHBBB#~" //心跳检测消息
HeartCheckSuccess = "HB" //成功发送返回心跳消息
HeartSuccessWord = "成功" //成功发送返回心跳消息
@@ -345,3 +345,10 @@ func RandString() string {
}
return string(bytes)
}
// JsonCommon json格式化
func JsonCommon(str string) (retVal string) {
temp, _ := json.Marshal(str)
_ = json.Unmarshal(temp, &retVal)
return retVal
}

18
business/partner/purchase/im/im_server.go
View File

@@ -90,18 +90,18 @@ func PingTimer() {
if conn.ClientType == ClientTypeJx {
if err := conn.Socket.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(time.Second)); err != nil {
Manager.DisConnect <- conn
globals.SugarLogger.Debugf("发送心跳失败: %s 总连接数：%d", clientId, Manager.Count())
fmt.Printf("发送心跳失败: %s 总连接数：%d", clientId, Manager.Count())
}
if err := ConnRender(conn.Socket, renderData{ClientId: clientId}); err != nil {
return
}
} else {
if err := conn.Socket.WriteMessage(websocket.TextMessage, []byte(HeartCheckMsg)); err != nil {
fmt.Printf("PingTimer mtHeartBeat err:%v", err)
//对美团重新建立连接
MtInit()
}
}
//globals.SugarLogger.Debugf("发送心跳 clientId=%s,i=%d", clientId, i)
}
}
}()
@@ -114,11 +114,11 @@ func WriteMessage() {
clientInfo := <-ToClientChan
//广播发送通知所有京西客户端
i++
fmt.Printf("WriteMessage clientInfo=%s i=%d", utils.Format4Output(clientInfo, false), i)
fmt.Printf("WriteMessage clientInfo=%s i=%d\n", utils.Format4Output(clientInfo, false), i)
if Manager.AllClient() != nil {
for _, conn := range Manager.AllClient() {
if conn.ClientType == ClientTypeJx { //只发送给京西
globals.SugarLogger.Debugf("WriteMessage conn.ClientId=%s", conn.ClientId)
fmt.Printf("WriteMessage conn.ClientId=%s\n", conn.ClientId)
if err := Render(conn.Socket, clientInfo.MessageId, clientInfo.Code, clientInfo.Msg, clientInfo.Data); err != nil {
Manager.DisConnect <- conn
}
@@ -128,11 +128,6 @@ func WriteMessage() {
globals.SugarLogger.Debugf("无客户端连接，请检查")
return
}
//if conn, err := Manager.GetByClientId(clientInfo.ClientId); err == nil && conn != nil {
// if err := Render(conn.Socket, clientInfo.MessageId, clientInfo.Code, clientInfo.Msg, clientInfo.Data); err != nil {
// Manager.DisConnect <- conn
// }
//}
}
}
@@ -155,8 +150,8 @@ func (c *Client) Read() {
go func() {
for {
messageType, msg, err := c.Socket.ReadMessage()
temp := string(msg)
fmt.Print(temp)
//temp := string(msg)
//fmt.Print(temp)
if err != nil {
if messageType == -1 && websocket.IsCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) {
Manager.DisConnect <- c
@@ -165,6 +160,7 @@ func (c *Client) Read() {
return
}
}
fmt.Printf("Client Read:receive: %s\n", string(msg))
SendToVendor(msg)
}
}()