增加支持HolderCloseReq总线请求

api/trade/trade.go

@@ -0,0 +1,17 @@
+package trade
+
+import (
+	"mtp20access/model/common/response"
+	"mtp20access/service/trade"
+
+	"github.com/gin-gonic/gin"
+)
+
+func Trade(c *gin.Context) {
+	if err := trade.TradeConn(c); err != nil {
+		response.FailWithMessage(err.Error(), c)
+		// return
+	}
+
+	// response.OkWithMessage("连接成功", c)
+}

client/client.go

@@ -21,16 +21,22 @@ var Clients map[int]*Client // key:SessionID
 type Client struct {
 	LoginRedis
 
+	// FIXME: - 这里是否需要定义多个锁？
 	mtx             sync.RWMutex
 	curSerialNumber uint32                // 当前业务流水号
 	asyncTasks      map[string]*AsyncTask // key:SessionId_FuncodeRsp_SerialNumber
 
-	wsConn      *websocket.Conn      // 终端WebSocket连接
-	quoteSubs   []request.QuoteGoods // 当前已订阅行情的商品
-	ch          chan interface{}     // 接收实时行情订阅channel
-	unSubscribe chan struct{}        // 取消订阅信号
-	writeChan   chan []byte          // 推送队列 QuoteServer -> Client
-	wsCloseChan chan struct{}        // 终端WebSocket连接关闭信号
+	wsQuoteConn      *websocket.Conn      // 终端行情WebSocket连接
+	quoteWriteChan   chan []byte          // 推送队列 Access -> Client
+	quoteChan        chan interface{}     // 接收实时行情订阅channel QuoteServer -> Access
+	quoteSubs        []request.QuoteGoods // 当前已订阅行情的商品
+	unSubscribe      chan struct{}        // 取消订阅信号
+	wsQuoteCloseChan chan struct{}        // 终端WebSocket连接关闭信号
+
+	wsTradeConn      *websocket.Conn  // 终端交易WebSocket连接
+	tradeWriteChan   chan []byte      // 推送队列 Access -> Client
+	tradeChan        chan interface{} // 接收交易通知channel RabbitMQ -> Access
+	wsTradeCloseChan chan struct{}    // 终端WebSocket连接关闭信号
 }
 
 // GetSerialNumber 获取可用流水号
@@ -51,16 +57,16 @@ func (r *Client) SetQuoteSubs(req request.QuoteSubscribeReq) {
 
 	r.quoteSubs = req.QuoteGoodses
 
-	if r.ch != nil {
+	if r.quoteChan != nil {
 		// r.unSubscribe <- struct{}{}
-		global.M2A_Publish.Unsubscribe(publish.Topic_Quote, r.ch)
+		global.M2A_Publish.Unsubscribe(publish.Topic_Quote, r.quoteChan)
 	}
-	r.ch = global.M2A_Publish.Subscribe(publish.Topic_Quote)
+	r.quoteChan = global.M2A_Publish.Subscribe(publish.Topic_Quote)
 
 	go func() {
 		for {
 			select {
-			case msg, ok := <-r.ch:
+			case msg, ok := <-r.quoteChan:
 				if !ok {
 					return // 管道已关闭，退出循环
 				}
@@ -86,7 +92,7 @@ func (r *Client) GetQuoteSubs() (quoteSubs []request.QuoteGoods) {
 
 // WriteWsBuf 向客户端发送实时行情
 func (r *Client) WriteWsBuf(buf []byte) (err error) {
-	r.writeChan <- buf
+	r.quoteWriteChan <- buf
 
 	return
 }
@@ -134,22 +140,24 @@ func (r *Client) GetAllAsyncTask() *map[string]*AsyncTask {
 	return &r.asyncTasks
 }
 
-func (r *Client) SetWebSocket(ws *websocket.Conn) (err error) {
+// **************** Quote WebSocket ****************
+
+func (r *Client) SetQuoteWebSocket(ws *websocket.Conn) (err error) {
 	r.mtx.Lock()
 	defer r.mtx.Unlock()
 
-	if r.wsConn != nil {
-		r.wsConn.Close()
+	if r.wsQuoteConn != nil {
+		r.wsQuoteConn.Close()
 	}
-	r.wsConn = ws
-	r.writeChan = make(chan []byte, 100)
+	r.wsQuoteConn = ws
+	r.quoteWriteChan = make(chan []byte, 100)
 
 	// 开始读取客户端发送信息
-	go r.readClientWsMessage()
+	go r.readClientWsQuoteMessage()
 	// 开始推送客户端信息循环
-	go r.writeClientWsMessage()
+	go r.writeClientWsQuoteMessage()
 
-	// r.wsConn.SetCloseHandler(func(code int, text string) error {
+	// r.wsQuoteConn.SetCloseHandler(func(code int, text string) error {
 	// 	close(r.wsCloseChan)
 	// 	return nil
 	// })
@@ -157,14 +165,14 @@ func (r *Client) SetWebSocket(ws *websocket.Conn) (err error) {
 	return
 }
 
-// readClientWsMessage 处理终端发过来的websocket数据
+// readClientWsQuoteMessage 处理终端发过来的websocket数据
 // 注意: 阻塞式, 直到websocket关闭才退出
-func (r *Client) readClientWsMessage() {
+func (r *Client) readClientWsQuoteMessage() {
 	for {
 		// 40秒心跳超时
-		r.wsConn.SetReadDeadline(time.Now().Add(40 * time.Second))
+		r.wsQuoteConn.SetReadDeadline(time.Now().Add(40 * time.Second))
 
-		mt, msg, err := r.wsConn.ReadMessage()
+		mt, msg, err := r.wsQuoteConn.ReadMessage()
 		if err != nil {
 			fmt.Println(err)
 			break
@@ -172,7 +180,7 @@ func (r *Client) readClientWsMessage() {
 
 		switch mt {
 		case websocket.PingMessage:
-			_ = r.wsConn.WriteMessage(mt, msg)
+			_ = r.wsQuoteConn.WriteMessage(mt, msg)
 		case websocket.CloseMessage:
 			return
 		case websocket.BinaryMessage:
@@ -181,7 +189,7 @@ func (r *Client) readClientWsMessage() {
 			}
 		case websocket.TextMessage:
 			fmt.Println(string(msg))
-			_ = r.wsConn.WriteMessage(mt, msg)
+			_ = r.wsQuoteConn.WriteMessage(mt, msg)
 		}
 
 		// FIXME: - 这里要判断是否有问题
@@ -192,18 +200,18 @@ func (r *Client) readClientWsMessage() {
 	}
 }
 
-// writeClientWsMessage 由于websocket非线程安全,
+// writeClientWsQuoteMessage 由于websocket非线程安全,
 // 所以由统一协程写入
-func (r *Client) writeClientWsMessage() {
+func (r *Client) writeClientWsQuoteMessage() {
 	// defer r.close()
 	for {
 		select {
-		case buf := <-r.writeChan:
-			err := r.wsConn.WriteMessage(websocket.BinaryMessage, buf)
+		case buf := <-r.quoteWriteChan:
+			err := r.wsQuoteConn.WriteMessage(websocket.BinaryMessage, buf)
 			if err != nil {
 				return
 			}
-		case <-r.wsCloseChan: // 与终端连接关闭信息
+		case <-r.wsQuoteCloseChan: // 与终端连接关闭信息
 			return
 		}
 	}
@@ -221,7 +229,7 @@ func (r *Client) clientToQuoteAgentMsg(msg []byte) error {
 	if p.Length == 15 {
 		// logger.Logger().Debugf("%v [%v c->s] %s", r.ClientIP, r.ProxyId, p.HeaderInfo())
 		// 将心跳发回给客户端
-		err = r.wsConn.WriteMessage(websocket.BinaryMessage, msg)
+		err = r.wsQuoteConn.WriteMessage(websocket.BinaryMessage, msg)
 		if err != nil {
 			return err
 		}
@@ -230,6 +238,74 @@ func (r *Client) clientToQuoteAgentMsg(msg []byte) error {
 	return nil
 }
 
+// **************** Trade WebSocket ****************
+
+func (r *Client) SetTradeWebSocket(ws *websocket.Conn) (err error) {
+	r.mtx.Lock()
+	defer r.mtx.Unlock()
+
+	if r.wsTradeConn != nil {
+		r.wsTradeConn.Close()
+	}
+	r.wsTradeConn = ws
+	r.tradeWriteChan = make(chan []byte, 100)
+
+	// 开始读取客户端发送信息
+	go r.readClientWsTradeMessage()
+	// 开始推送客户端信息循环
+	go r.writeClientWsTradeMessage()
+
+	return
+}
+
+// readClientWsQuoteMessage 处理终端发过来的websocket数据
+// 注意: 阻塞式, 直到websocket关闭才退出
+func (r *Client) readClientWsTradeMessage() {
+	for {
+		// 40秒心跳超时
+		r.wsTradeConn.SetReadDeadline(time.Now().Add(40 * time.Second))
+
+		mt, msg, err := r.wsTradeConn.ReadMessage()
+		if err != nil {
+			fmt.Println(err)
+			break
+		}
+
+		switch mt {
+		case websocket.PingMessage:
+			_ = r.wsTradeConn.WriteMessage(mt, msg)
+		case websocket.CloseMessage:
+			return
+		case websocket.BinaryMessage:
+			if err := r.clientToQuoteAgentMsg(msg); err != nil {
+				return
+			}
+		case websocket.TextMessage:
+			fmt.Println(string(msg))
+			_ = r.wsTradeConn.WriteMessage(mt, msg)
+		}
+	}
+}
+
+// writeClientWsQuoteMessage 由于websocket非线程安全,
+// 所以由统一协程写入
+func (r *Client) writeClientWsTradeMessage() {
+	// defer r.close()
+	for {
+		select {
+		case buf := <-r.tradeWriteChan:
+			err := r.wsTradeConn.WriteMessage(websocket.BinaryMessage, buf)
+			if err != nil {
+				return
+			}
+		case <-r.wsTradeCloseChan: // 与终端连接关闭信息
+			return
+		}
+	}
+}
+
+// **************** RabbitMQ ****************
+
 // MQPacket 与总线交互的数据体
 type MQPacket struct {
 	FunCode   uint32  // 功能码

client/msgRealQuote.go

@@ -41,7 +41,7 @@ func DispatchRealQuote(p *packet.MiQuotePacket, clinet *Client) {
 		sendBuf := quote.EnPack()
 
 		// 发送给客户端
-		clinet.writeChan <- sendBuf
+		clinet.quoteChan <- sendBuf
 	}
 }
 

global/funcode.go

@@ -1,6 +1,8 @@
 package global
 
 var (
+	MoneyChangedNtf = 131076 // 资金变化通知
+
 	AccountFundInfoReq = 262150 // 账户资金信息请求
 	AccountFundInfoRsp = 262151 // 账户资金信息响应
 
@@ -88,6 +90,8 @@ var (
 	OrderRsp       = 196610 // 交易委托应答
 	CancelOrderReq = 196611 // 撤单请求
 	CancelOrderRsp = 196612 // 撤单应答
+	HolderCloseReq = 196713 // 按单平仓请求
+	HolderCloseRsp = 196714 // 按单平仓应答
 )
 
 // 通过请求功能码获取对应主题的方法

global/topic.go

@@ -1,6 +1,8 @@
 package global
 
 var (
+	TOPIC_RSP_NTF = "notify_rsp" // 回报主题（通知）
+
 	TOPIC_REQ_QKERNEL = "query_req" // 查询相关请求
 	TOPIC_RSP_QKERNEL = "query_rsp" // 查询相关回应
 

initialize/rabbitmq.go

@@ -50,41 +50,48 @@ func (t *MQProc) Process(topic, queuename string, msg *[]byte) {
 	// global.M2A_LOG.Info(info)
 
 	if funcode, sessionId, bytes, serialNumber, err := t.getRspProtobuf(msg); err == nil && bytes != nil {
-		// 尝试获取对应异步任务
-		if client, exists := client.Clients[int(sessionId)]; exists {
-			key := fmt.Sprintf("%v_%v_%v", sessionId, funcode, serialNumber)
-			// 银行服务相关的回复流水号是错误的，所以需要特殊处理
-			if int(funcode) == global.T2bBankSignRsp ||
-				int(funcode) == global.T2bBankCancelSignRsp ||
-				int(funcode) == global.T2bBankWithdrawRsp ||
-				int(funcode) == global.T2bBankDepositRsp {
-				key = fmt.Sprintf("%v_%v", sessionId, funcode)
-			}
-			asyncTask := client.GetAsyncTask(key)
-			if asyncTask != nil {
-				rspData := string(*bytes)
-				// 判断是否要加密
-				if asyncTask.IsEncrypted {
-					if b, err := packet.Encrypt(*bytes, packet.AESKey, true); err != nil {
-						global.M2A_LOG.Error("总线回复数据加密失败", zap.Error(err))
-						return
-					} else {
-						rspData = base64.StdEncoding.EncodeToString(b)
-					}
+		if sessionId == 0 {
+			// 通知类
+
+		} else {
+			// 请求回复
+
+			// 尝试获取对应异步任务
+			if client, exists := client.Clients[int(sessionId)]; exists {
+				key := fmt.Sprintf("%v_%v_%v", sessionId, funcode, serialNumber)
+				// 银行服务相关的回复流水号是错误的，所以需要特殊处理
+				if int(funcode) == global.T2bBankSignRsp ||
+					int(funcode) == global.T2bBankCancelSignRsp ||
+					int(funcode) == global.T2bBankWithdrawRsp ||
+					int(funcode) == global.T2bBankDepositRsp {
+					key = fmt.Sprintf("%v_%v", sessionId, funcode)
 				}
-				// 给客户端回调
-				global.M2A_LOG.Info("[S->C]", zap.Any("rsp", funcode), zap.Any("sessionId", sessionId), zap.Any("data", string(rspData)))
-				r := rsp.MQBodyRsp{
-					FunCode:     funcode,
-					IsEncrypted: asyncTask.IsEncrypted,
-					Data:        rspData,
+				asyncTask := client.GetAsyncTask(key)
+				if asyncTask != nil {
+					rspData := string(*bytes)
+					// 判断是否要加密
+					if asyncTask.IsEncrypted {
+						if b, err := packet.Encrypt(*bytes, packet.AESKey, true); err != nil {
+							global.M2A_LOG.Error("总线回复数据加密失败", zap.Error(err))
+							return
+						} else {
+							rspData = base64.StdEncoding.EncodeToString(b)
+						}
+					}
+					// 给客户端回调
+					global.M2A_LOG.Info("[S->C]", zap.Any("rsp", funcode), zap.Any("sessionId", sessionId), zap.Any("data", string(rspData)))
+					r := rsp.MQBodyRsp{
+						FunCode:     funcode,
+						IsEncrypted: asyncTask.IsEncrypted,
+						Data:        rspData,
+					}
+					asyncTask.Rsp <- r
+				} else {
+					global.M2A_LOG.Info("找不到对应KEY的异步任务", zap.Any("key", key), zap.Any("AsyncMap", client.GetAllAsyncTask()))
 				}
-				asyncTask.Rsp <- r
 			} else {
-				global.M2A_LOG.Info("找不到对应KEY的异步任务", zap.Any("key", key), zap.Any("AsyncMap", client.GetAllAsyncTask()))
+				global.M2A_LOG.Info("找不到对应的client", zap.Any("sessionId", sessionId))
 			}
-		} else {
-			global.M2A_LOG.Info("找不到对应的client", zap.Any("sessionId", sessionId))
 		}
 	}
 }
@@ -99,6 +106,19 @@ func (t *MQProc) getRspProtobuf(msg *[]byte) (funcode uint32, sessionId uint32,
 	global.M2A_LOG.Info("收到总线消息", zap.Any("funcode", funcode), zap.Any("sessionId", sessionId), zap.Any("len", len(b)))
 
 	switch int(funcode) {
+	case global.MoneyChangedNtf: // 资金变化通知
+		var p pb.MoneyChangedNtf
+		if err = proto.Unmarshal(b, &p); err != nil {
+			global.M2A_LOG.Error("总线通知数据反序列化失败", zap.Error(err))
+			return
+		}
+		if bs, e := protojson.Marshal(&p); e != nil {
+			global.M2A_LOG.Error("总线通知数据反序列化失败", zap.Error(err))
+			return
+		} else {
+			bytes = &bs
+			serialNumber = p.GetHeader().GetRequestID()
+		}
 	case global.ModifyPwdRsp: // 修改账户密码应答
 		var p pb.ModifyPwdRsp
 		if err = proto.Unmarshal(b, &p); err != nil {
@@ -567,6 +587,19 @@ func (t *MQProc) getRspProtobuf(msg *[]byte) (funcode uint32, sessionId uint32,
 			bytes = &bs
 			serialNumber = p.GetHeader().GetRequestID()
 		}
+	case global.HolderCloseRsp:
+		var p pb.HolderCloseRsp
+		if err = proto.Unmarshal(b, &p); err != nil {
+			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
+			return
+		}
+		if bs, e := protojson.Marshal(&p); e != nil {
+			global.M2A_LOG.Error("总线回复数据反序列化失败", zap.Error(err))
+			return
+		} else {
+			bytes = &bs
+			serialNumber = p.GetHeader().GetRequestID()
+		}
 	}
 
 	return
@@ -575,6 +608,10 @@ func (t *MQProc) getRspProtobuf(msg *[]byte) (funcode uint32, sessionId uint32,
 // RabbitMQSubscribeTopic 订阅主题
 func RabbitMQSubscribeTopic() (err error) {
 	// 订阅需要的总线响应主题
+	if err = rabbitmq.SubscribeTopic(global.TOPIC_RSP_NTF); err != nil {
+		global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
+		return
+	}
 	if err = rabbitmq.SubscribeTopic(global.TOPIC_RSP_QKERNEL); err != nil {
 		global.M2A_LOG.Error("rabbitmq subscribe topic failed, err:", zap.Error(err))
 		return
@@ -723,5 +760,6 @@ func InitFuncodeTopic() {
 	global.M2A_FuncodeTopic[global.TOPIC_REQ_ORDER] = []int{
 		global.OrderReq,
 		global.CancelOrderReq,
+		global.HolderCloseReq,
 	}
 }

initialize/router.go

@@ -40,6 +40,7 @@ func Routers() *gin.Engine {
 		}
 		router.InitAccountPublicRouter(publicGroup)
 		router.InitQuotePublicRouter(publicGroup)
+		router.InitTradePublicRouter(publicGroup)
 	}
 
 	// 鉴权组

model/account/account.go

@@ -44,3 +44,82 @@ func (r *Loginaccount) TableName() string {
 func (r *Loginaccount) Get() (has bool, err error) {
 	return global.M2A_DB.Get(r)
 }
+
+// Taaccount 资金账户表 - 导历史
+type Taaccount struct {
+	ACCOUNTID             int64     `json:"accountid" xorm:"ACCOUNTID"`                         // 资金账户ID
+	USERID                int64     `json:"userid" xorm:"USERID"`                               // 用户ID
+	CURRENCYID            int64     `json:"currencyid" xorm:"CURRENCYID"`                       // 货币ID
+	CHANGEFLAG            int32     `json:"changeflag" xorm:"CHANGEFLAG"`                       // 变动标志(当前账户资金有任何变动更新为1系统结算时更新0;供清算时使用) 0：无变动 1：有变动
+	PASSWORD              string    `json:"password" xorm:"PASSWORD"`                           // 资金密码
+	TRADESTATUS           int32     `json:"tradestatus" xorm:"TRADESTATUS"`                     // 交易状态 - 1:正常 2:受限 3:冻结(停用) 4:禁止建仓(人工受限) 5:禁止交易(人工冻结) 6:待激活 7:已删除(注销)
+	SIGNSTATUS            int32     `json:"signstatus" xorm:"SIGNSTATUS"`                       // 签约状态 - 1:未签约 2:签约待审核 3:签约中 4:已签约 5:解约待审核 6:解约中 7:已解约 8:已解绑 9:绑卡中
+	ISMAIN                int32     `json:"ismain" xorm:"ISMAIN"`                               // 是否母账号 0:不是母账户 1:是母账户
+	PARENTACCOUNTID       int64     `json:"parentaccountid" xorm:"PARENTACCOUNTID"`             // 所属根账号
+	RELATEDUSERID         int64     `json:"relateduserid" xorm:"RELATEDUSERID"`                 // 关联用户
+	RELATEDACCOUNTSTATUS  int32     `json:"relatedaccountstatus" xorm:"RELATEDACCOUNTSTATUS"`   // 关联用户交易权限状态 - 1:正常(可交易) 2:受限(可平仓，不可建仓) 3:冻结(不可交易)
+	TAACCOUNTTYPE         int32     `json:"taaccounttype" xorm:"TAACCOUNTTYPE"`                 // 账号类型 - 1:外部账号 2:内部账号 3:内部做市自营账号 4:内部做市接单账号
+	ISRECKONACCOUNT       int32     `json:"isreckonaccount" xorm:"ISRECKONACCOUNT"`             // 是否机构分润账号 0:不是 1:是
+	ISMARKETACCOUNT       int32     `json:"ismarketaccount" xorm:"ISMARKETACCOUNT"`             // 是否机构接单账号 0:不是 1:是
+	BALANCE               float64   `json:"balance" xorm:"BALANCE"`                             // 期初余额
+	ORIFREEZEMARGIN       float64   `json:"orifreezemargin" xorm:"ORIFREEZEMARGIN"`             // 期初冻结保证金
+	ORIUSEDMARGIN         float64   `json:"oriusedmargin" xorm:"ORIUSEDMARGIN"`                 // 期初占用保证金
+	ORIOTHERFREEZEMARGIN  float64   `json:"oriotherfreezemargin" xorm:"ORIOTHERFREEZEMARGIN"`   // 期初其他冻结保证金(出金冻结资金 交割买方冻结 申购冻结 全款买入 商城买入)
+	ORIFREEZECHARGE       float64   `json:"orifreezecharge" xorm:"ORIFREEZECHARGE"`             // 期初手续费冻结
+	ORIMORTGAGECREDIT     float64   `json:"orimortgagecredit" xorm:"ORIMORTGAGECREDIT"`         // 期初授信金额
+	ORIOTHERCREDIT        float64   `json:"oriothercredit" xorm:"ORIOTHERCREDIT"`               // 期初其它授信金额
+	ORIOUTAMOUNTFREEZE    float64   `json:"orioutamountfreeze" xorm:"ORIOUTAMOUNTFREEZE"`       // 期初出金冻结
+	FREEZEMARGIN          float64   `json:"freezemargin" xorm:"FREEZEMARGIN"`                   // 冻结保证金
+	USEDMARGIN            float64   `json:"usedmargin" xorm:"USEDMARGIN"`                       // 占用保证金
+	OTHERFREEZEMARGIN     float64   `json:"otherfreezemargin" xorm:"OTHERFREEZEMARGIN"`         // 其他冻结保证金(交割买方冻结 申购冻结 全款买入 商城买入)
+	FREEZECHARGE          float64   `json:"freezecharge" xorm:"FREEZECHARGE"`                   // 手续费冻结
+	MORTGAGECREDIT        float64   `json:"mortgagecredit" xorm:"MORTGAGECREDIT"`               // 授信金额
+	OTHERCREDIT           float64   `json:"othercredit" xorm:"OTHERCREDIT"`                     // 其它授信金额
+	OUTAMOUNTFREEZE       float64   `json:"outamountfreeze" xorm:"OUTAMOUNTFREEZE"`             // 出金冻结
+	INAMOUNT              float64   `json:"inamount" xorm:"INAMOUNT"`                           // 今日入金金额(包括三方入金)
+	CURRENTBALANCE        float64   `json:"currentbalance" xorm:"CURRENTBALANCE"`               // 期末余额
+	OUTAMOUNT             float64   `json:"outamount" xorm:"OUTAMOUNT"`                         // 今日出金金额(包括三方出金)
+	PAYCHARGE             float64   `json:"paycharge" xorm:"PAYCHARGE"`                         // 今日手续费支出
+	CLOSEPL               float64   `json:"closepl" xorm:"CLOSEPL"`                             // 今日平仓盈亏
+	CREDITINCREASE        float64   `json:"creditincrease" xorm:"CREDITINCREASE"`               // 今日授信增加
+	CREDITDECREASE        float64   `json:"creditdecrease" xorm:"CREDITDECREASE"`               // 今日授信减少
+	OTHERCREDITINCREASE   float64   `json:"othercreditincrease" xorm:"OTHERCREDITINCREASE"`     // 今日其它授信增加
+	OTHERCREDITDECREASE   float64   `json:"othercreditdecrease" xorm:"OTHERCREDITDECREASE"`     // 今日其它授信减少
+	TRANSFERAMOUNT        float64   `json:"transferamount" xorm:"TRANSFERAMOUNT"`               // 今日划转金额(母子账号资金划转，从划入账号为正，从账号划出为负)
+	OTHERPAY              float64   `json:"otherpay" xorm:"OTHERPAY"`                           // 其他支出(交割付款 申购付款 全款买入 商城买入 卖家退货)
+	OTHERINCOME           float64   `json:"otherincome" xorm:"OTHERINCOME"`                     // 其他收入(交割收款 申购收款 全款卖出 商城卖出 买家退货 会员手续费收入)
+	OUTTHRESHOLD          float64   `json:"outthreshold" xorm:"OUTTHRESHOLD"`                   // 出金阈值
+	TRADESTATUSCHANGETIME time.Time `json:"tradestatuschangetime" xorm:"TRADESTATUSCHANGETIME"` // 激活时间
+	CHANGETIME            time.Time `json:"changetime" xorm:"CHANGETIME"`                       // 账户状态变更时间
+	THIRDINAMOUNT         float64   `json:"thirdinamount" xorm:"THIRDINAMOUNT"`                 // 今日三方入金
+	THIRDOUTAMOUNT        float64   `json:"thirdoutamount" xorm:"THIRDOUTAMOUNT"`               // 今日三方出金
+	CAPITALBALANCE        float64   `json:"capitalbalance" xorm:"CAPITALBALANCE"`               // 本金余额[外部子母账户实际出入金余额]
+	FROMACCOUNTID         int64     `json:"fromaccountid" xorm:"FROMACCOUNTID"`                 // 所属上级账户
+	SUBLEVELPATH          string    `json:"sublevelpath" xorm:"SUBLEVELPATH"`                   // 账号层级路径(逗号分隔,首尾加逗号)
+	SERIVCEGROUP          int64     `json:"serivcegroup" xorm:"SERIVCEGROUP"`                   // 服务分组
+	ACCOUNTNAME           string    `json:"accountname" xorm:"ACCOUNTNAME"`                     // 账户名称
+	ACCOUNTFLAG           int32     `json:"accountflag" xorm:"ACCOUNTFLAG"`                     // 账户标识 - 0\1 (默认为0， 当上级账户与本账户的关联用户均为自己时更新为1)
+	ORIDEBT               float64   `json:"oridebt" xorm:"ORIDEBT"`                             // 期初欠款
+	CURDEBT               float64   `json:"curdebt" xorm:"CURDEBT"`                             // 期末欠款
+	MODIFYTIME            time.Time `json:"modifytime" xorm:"MODIFYTIME"`                       // 修改时间
+	MODIFYSRC             int32     `json:"modifysrc" xorm:"MODIFYSRC"`                         // 修改来源 - 1:管理端 2:终端
+	MODIFIERID            int64     `json:"modifierid" xorm:"MODIFIERID"`                       // 修改人
+}
+
+// TableName is TAACCOUNT
+func (r *Taaccount) TableName() string {
+	return "TAACCOUNT"
+}
+
+// Logintaaccount 登录账户资金权限表
+type Logintaaccount struct {
+	LOGINID    int64     `json:"loginid" xorm:"LOGINID"`       // 登录账号ID
+	ACCOUNTID  int64     `json:"accountid" xorm:"ACCOUNTID"`   // 资金账号ID
+	MODIFYTIME time.Time `json:"modifytime" xorm:"MODIFYTIME"` // 修改时间
+	MODIFIERID int64     `json:"modifierid" xorm:"MODIFIERID"` // 修改人
+}
+
+// TableName is LOGINTAACCOUNT
+func (r *Logintaaccount) TableName() string {
+	return "LOGINTAACCOUNT"
+}

model/mq/request/mq.go

@@ -734,6 +734,24 @@ func (r *MQBodyReq) GetProtoBytes(serialNumber *uint32) (bytes *[]byte, err erro
 		} else {
 			bytes = &b
 		}
+	case global.HolderCloseReq:
+		m := pb.HolderCloseReq{}
+		if err = r.reflect(data, &m); err != nil {
+			return
+		}
+		if m.Header != nil {
+			m.Header.RequestID = serialNumber
+		} else {
+			err = errors.New("请求信息序列化失败")
+			return
+		}
+		if b, e := proto.Marshal(&m); e != nil {
+			global.M2A_LOG.Error(e.Error(), zap.Error(e))
+			err = errors.New("请求信息序列化失败")
+			return
+		} else {
+			bytes = &b
+		}
 	}
 
 	return

packet/packet.go

@@ -0,0 +1,256 @@
+package packet
+
+import (
+	"bytes"
+	"compress/zlib"
+	"errors"
+	"fmt"
+	"hash/crc32"
+	"io"
+	"log"
+	"mtp20access/utils"
+	"net"
+)
+
+/************协议格式************************************
+
+WTAS协议是交易系统与管理客户端/会员系统之间的通信协议，由报文头 + 报文体两部分组成
+标识	            类型	字节数	内容	说明
+HeadTag	        byte	1	    0xFF	头标记
+Length	        uint	4		包总长度，包括头长度
+FunCode	        uint	4		功能号（为0代表心跳，心跳数据体长度为0）
+SessionId	    uint	4		会话ID，由交易接入(代理)维护
+Mode	        byte	1		内容类型，0:ProtoBuff，1:Json，2:Zip + ProtBuff，…
+Version	        byte	1		版本号
+SerialNumber	uint	4		通讯报文序号
+数据体				            业务结构体，ProtoBuf. 数据体格式： 头部 4 节字 + 内容 + 8 节字 掩码, 解密的时候只解内容部分
+CRC	            byte	4		报文校验和
+FootTag	        byte	1	0x00	尾标记
+
+数据体前,包头长度=19, 空包长度=24
+#define MAXKEY    "B0FB83E39A5EBFAABE471362A58393FF"
+#define TRANSKEY  "F7A72DE7D6264530F01BA49BC73EB873"
+*******************************************************/
+
+// MiPacket 协议包结构体
+type MiPacket struct {
+	Length       uint32 // 包总长度
+	FunCode      uint32 // 功能能
+	SessionId    uint32 // 数据包的sid
+	Mode         uint32 // 业务数据体的格式 0:ProtoBuff，1:Json，2:Zip + ProtoBuff
+	SerialNumber uint32 // 通信流水号
+	Data         []byte // 业务数据体
+	CRC          uint32 // CRC
+	OriMsg       []byte // 原始包数据
+}
+
+// EnPack 打包,根据现有的数据内容重新打包,无任何数据时打出来的是心跳包
+// 这里不对数据体进行加密, 如需加密，请先加密再设置进来
+func (p *MiPacket) EnPack() []byte {
+	p.Length = uint32(len(p.Data)) + 24 // 空包长度24
+
+	buf := make([]byte, 0)                                     // 缓冲区
+	buf = append(buf, byte(0xFF))                              // HeadTag 0xFF
+	buf = append(buf, utils.UintToBytes(p.Length)...)          // Length
+	buf = append(buf, utils.UintToBytes(p.FunCode)...)         // FunCode
+	buf = append(buf, utils.UintToBytes(p.SessionId)...)       // SessionId
+	buf = append(buf, byte(0))                                 // Mode
+	buf = append(buf, byte(0))                                 // Version
+	buf = append(buf, utils.UintToBytes(p.SerialNumber)...)    // SerialNumber
+	buf = append(buf, p.Data...)                               // 数据体 body
+	p.CRC = crc32.Update(58861227, crc32.IEEETable, buf[0:19]) //
+	buf = append(buf, utils.UintToBytes(p.CRC)...)             // CRC
+	buf = append(buf, byte(0))                                 // FootTag
+
+	return buf
+}
+
+// UnPackHead 解包头
+// @buf length must be >=19
+func (p *MiPacket) UnPackHead(buf []byte) error {
+	if len(buf) < 19 {
+		return errors.New("header len error")
+	}
+
+	if buf[0] != 0xFF {
+		return errors.New("header flag error")
+	}
+
+	p.Length = utils.BytesToUint32(buf[1:5])         // Length
+	p.FunCode = utils.BytesToUint32(buf[5:9])        // FunCode
+	p.SessionId = utils.BytesToUint32(buf[9:13])     // SessionId
+	p.SerialNumber = utils.BytesToUint32(buf[15:19]) // SerialNumber
+	p.Mode = uint32(buf[13])                         // Mode
+
+	if p.Length > 1024*10000 || p.Length < 24 {
+		fmt.Println("packet too big or len error, len:", p.Length)
+		return fmt.Errorf("invalid len, must in[24,1024000]")
+	}
+
+	return nil
+}
+
+// HeaderLen 包头长度, HeaderTag~SerialNumber 的长度
+// 用于tcp接收数据时第一部分的长度
+func (p *MiPacket) HeaderLen() uint32 {
+	return 19
+}
+
+// BodyLen 除包头外的长度, 含FootTag
+// 用于tcp接收数据时第二部分的长度
+func (p *MiPacket) BodyLen() uint32 {
+	return p.Length - 19
+}
+
+// FuncCode 功能号
+func (p *MiPacket) FuncCode() int {
+	return int(p.FunCode)
+}
+
+// UnPack 从指定的缓冲区解包
+// 注意：不解密业务数据体, 可用 DecodeData 方法进行解密
+// @buf 完整的包数据内容
+func (p *MiPacket) UnPack(buf []byte) error {
+	err := p.UnPackHead(buf[:19])
+	if err != nil {
+		return err
+	}
+	if len(buf) != int(p.Length) {
+		err := fmt.Errorf("packet length err")
+		log.Println(err)
+		return err
+	}
+	//长度24为心跳包, 非心跳包时保存数据体内容
+	//包头前面19字节+4个字节数据体加密长度, 所以从23开始
+	//包尾5个字节,但是加密的数据体后8个字节掩码不需要, 所以是 p.length-13
+	if len(buf) != 24 {
+		p.Data = buf[23 : p.Length-13]
+	}
+	p.OriMsg = buf[:]
+	return nil
+}
+
+// DecodeData 解密业务数据体内容
+// @mode 数据体类型 0:ProtoBuff，1:Json，2:Zip + ProtoBuff
+// @msg  业务数据体
+func (p *MiPacket) DecodeData(mode uint32, msg []byte) ([]byte, error) {
+	if mode == 2 {
+		b := bytes.NewReader(msg)
+		r, err := zlib.NewReader(b)
+		if err != nil {
+			return nil, err
+		}
+		var out bytes.Buffer
+		_, _ = io.Copy(&out, r)
+		_ = r.Close()
+		msg = out.Bytes()
+	}
+
+	return Decrypt(msg, AESKey, true)
+}
+
+// EncodeData 加密码业务数据体内容
+func (p *MiPacket) EncodeData(mode uint32, msg []byte) ([]byte, error) {
+	buf := msg
+	if mode == 2 {
+		var b bytes.Buffer
+		w := zlib.NewWriter(&b)
+		_, err := w.Write(msg)
+		if err != nil {
+			return nil, err
+		}
+		_ = w.Close()
+		buf = b.Bytes()
+	}
+	return Encrypt(buf, AESKey, true)
+}
+
+// SetData 设置业务数据, 不含通信包头、CRC等字段
+func (p *MiPacket) SetData(buf []byte) {
+	p.Data = buf[:]
+}
+
+// SetOriMsg 设置原始包数据
+func (p *MiPacket) SetOriMsg(arg ...[]byte) {
+	if p.OriMsg == nil {
+		p.OriMsg = make([]byte, 0)
+	}
+	for i := range arg {
+		p.OriMsg = append(p.OriMsg, arg[i]...)
+	}
+}
+
+// GetOriMsg 获取解包的原始数据
+func (p *MiPacket) GetOriMsg() []byte {
+	return p.OriMsg
+}
+
+// RebuildForNewSid 设置新的通信流水号(sessionId)且重新打包
+func (p *MiPacket) RebuildForNewSid(sessionId uint32) {
+	// 从某个位置开始, 替换一段内容
+	f := func(buf []byte, pos int, newBuf []byte) {
+		for i := 0; i < len(newBuf); i++ {
+			buf[pos] = newBuf[i]
+			pos++
+		}
+	}
+	p.SessionId = sessionId
+	s := utils.UintToBytes(p.SessionId)
+	f(p.OriMsg, 9, s)
+	p.CRC = crc32.Update(58861227, crc32.IEEETable, p.OriMsg[0:19])
+	crc := utils.UintToBytes(p.CRC)
+	f(p.OriMsg, len(p.OriMsg)-5, crc)
+}
+
+// ReadMessage 从指定tcp链接读取一个协议包
+// @返回值 []byte 未解包的原始数据包, 如果需获取业务数据内容,
+// 请调用UnPack方法后取成员变量 Data 的内容
+func (p *MiPacket) ReadMessage(conn *net.Conn) ([]byte, error) {
+	p.OriMsg = make([]byte, 0)
+	if conn == nil {
+		return p.OriMsg, fmt.Errorf("conn is nil")
+	}
+
+	headerBuf := make([]byte, p.HeaderLen())
+	nRead, err := io.ReadFull(*conn, headerBuf)
+	if err != nil || nRead != len(headerBuf) {
+		return p.OriMsg, fmt.Errorf("read header data error, maybe conn closed:%v", err)
+	}
+
+	err = p.UnPackHead(headerBuf)
+	if err != nil {
+		return p.OriMsg, err
+	}
+
+	dataBuf := make([]byte, p.BodyLen())
+	nRead, err = io.ReadFull(*conn, dataBuf)
+	if err != nil || nRead != len(dataBuf) {
+		return p.OriMsg, fmt.Errorf("read data error, maybe conn closed:%v", err)
+	}
+
+	p.SetOriMsg(headerBuf, dataBuf)
+	return p.OriMsg, nil
+}
+
+// HeaderInfo 头部信息, 功能号、sid、流水号、长度等
+func (p *MiPacket) HeaderInfo() string {
+	return fmt.Sprintf("funcode[%d] sid[%d] serial[%d] iLen:%d",
+		p.FunCode, p.SessionId, p.SerialNumber, p.Length)
+}
+
+// BuildPacket 创建包
+// @bCrypto 是否对数据进行加密
+func BuildPacket(funCode, sessionId, serialNum uint32, msg []byte, bCrypto bool) ([]byte, error) {
+	p := new(MiPacket)
+	p.FunCode = funCode
+	p.SessionId = sessionId
+	p.SerialNumber = serialNum
+	if bCrypto && len(msg) > 0 {
+		buf, _ := Encrypt(msg, AESKey, true)
+		p.Data = buf
+		return p.EnPack(), nil
+	}
+
+	p.Data = msg
+	return p.EnPack(), nil
+}

res/pb/mtp2.proto

@@ -26,6 +26,13 @@ message NotifyHead {
 		repeated uint64 AccountIDs = 5; // 目标账号集合（以上均不匹配时，最后检查此集合）
 }
 
+// 资金变化通知
+message MoneyChangedNtf {
+	optional MessageHead Header = 1; // 消息头
+		optional NotifyHead NtfHeader = 2; // NotifyHead 公共消息头
+		optional uint64 AccountID = 3; // uint64 交易账号
+}
+
 // 文件列表
 message FileDetail {
 		optional string FileName = 1; // 文件名
@@ -1216,4 +1223,28 @@ message CancelOrderRsp {
 	optional string RetDesc = 3; // 描述信息
 		optional uint64 ExchActionID = 4; // 交易所操作号
 		optional string OrderTime = 5; // 接收委托交易的时间
+}
+
+// 按单平仓请求
+message HolderCloseReq {
+	optional MessageHead Header = 1;
+		optional uint64 TradeID = 2; // 持仓单号
+		optional string ClientSerialNo = 3; // 客户端流水号
+		optional string ClientOrderTime = 4; // 客户端委托时间
+		optional uint32 ClientType = 5; // 终端类型
+		optional uint64 LoginID = 6; // 登陆账号
+		optional uint64 AccountID = 7; // 交易账号
+		optional uint32 GoodsID = 8; // 商品ID
+		optional uint32 BuyOrSell = 9; // 买卖方向
+		optional uint32 MarketID = 10; // 市场ID
+		optional uint32 OrderSrc = 11; // 单据来源
+		optional uint64 OperatorID = 12; // 操作员账号ID
+}
+// 按单平仓应答
+message HolderCloseRsp {
+	optional MessageHead Header = 1; // 消息头
+	optional int32 RetCode = 2; // 返回码
+	optional string RetDesc = 3; // 描述信息
+		optional uint64 OrderID = 4; // 一级生成的订单号
+		optional string OrderTime = 5; // 接收委托交易的时间
 }

router/trade.go

@@ -0,0 +1,14 @@
+package router
+
+import (
+	"mtp20access/api/trade"
+
+	"github.com/gin-gonic/gin"
+)
+
+func InitTradePublicRouter(r *gin.RouterGroup) {
+	tradeR := r.Group("Trade").Use()
+	{
+		tradeR.GET("WS", trade.Trade)
+	}
+}

service/account/login.go

@@ -213,3 +213,8 @@ func RestoreLoginWithToken(loginID int, group int, token string) (err error) {
 
 	return
 }
+
+func GetClientsByAccountID(accountID uint64) (clients []*client.Client, err error) {
+
+	return
+}

service/quote/quote.go

@@ -95,7 +95,7 @@ func QuoteConn(c *gin.Context) (err error) {
 		return
 	}
 
-	client.SetWebSocket(ws)
+	client.SetQuoteWebSocket(ws)
 
 	return
 }

service/trade/trade.go

@@ -0,0 +1,99 @@
+package trade
+
+import (
+	"errors"
+	"mtp20access/client"
+	"mtp20access/global"
+	"mtp20access/model/common/response"
+	"mtp20access/utils"
+	"net/http"
+
+	"github.com/gin-gonic/gin"
+	"github.com/gorilla/websocket"
+	"go.uber.org/zap"
+)
+
+// TradeConn 终端连接WebSocket
+func TradeConn(c *gin.Context) (err error) {
+	// 获取请求账号信息
+	// s, exists := c.Get("claims")
+	// if !exists {
+	// 	err = errors.New("获取请求账号信息异常")
+	// 	global.M2A_LOG.Error(err.Error())
+	// 	return
+	// }
+	// claims := s.(*commonRequest.CustomClaims)
+
+	// 我们这里jwt鉴权取头部信息 x-token 登录时回返回token信息 这里前端需要把token存储到cookie或者本地localStorage中 不过需要跟后端协商过期时间 可以约定刷新令牌或者重新登录
+	token := c.Request.Header.Get("Sec-WebSocket-Protocol")
+	if token == "" {
+		response.FailWithDetailed(gin.H{"reload": true}, "未登录或非法访问", c)
+		c.Abort()
+		return
+	}
+
+	j := utils.NewJWT()
+	// parseToken 解析token包含的信息
+	claims, err := j.ParseToken(token)
+	if err != nil {
+		if err == utils.ErrTokenExpired {
+			response.FailWithCodeAndDetail(gin.H{"reload": true}, response.ERROR_TOKEN_EXPIRED, "授权已过期", c)
+			// c.Abort()
+			return
+		}
+		response.FailWithDetailed(gin.H{"reload": true}, err.Error(), c)
+		// c.Abort()
+		return
+	}
+	// 从Redis获取登录信息
+	loginMap, err := j.GetRedisLogin(claims.LoginID, claims.Group)
+	if err != nil {
+		response.FailWithCodeAndDetail(gin.H{"reload": true}, response.ERROR_TOKEN_OTHER_LOGIN, "您的帐户异地登陆或令牌失效", c)
+		// c.Abort()
+		return
+	}
+	// 判断Token是否有效
+	if redisToken, isHas := loginMap["token"]; isHas {
+		if redisToken != token {
+			response.FailWithCodeAndDetail(gin.H{"reload": true}, response.ERROR_TOKEN_OTHER_LOGIN, "您的帐户异地登陆或令牌失效", c)
+			// c.Abort()
+			return
+		}
+	} else {
+		response.FailWithDetailed(gin.H{"reload": true}, "未登录或非法访问", c)
+		// c.Abort()
+		return
+	}
+
+	// 获取登录账户信息
+	client, exists := client.Clients[claims.SessionID]
+	if !exists {
+		err = errors.New("获取登录账户信息异常")
+		global.M2A_LOG.Error(err.Error())
+		return
+	}
+
+	// 需要注意的是，如果前端通过websocket连接时指定了Sec-WebSocket-Protocol，后端接收到连接后，必须原封不动的将Sec-WebSocket-Protocol头信息返回给前端，否则连接会抛出异常。
+	// c.Header("Sec-WebSocket-Protocol", token)
+	// c.String(200, "ok")
+	// 将http连接升级为websocket连接
+	upGrader := websocket.Upgrader{
+		ReadBufferSize:  1000000,
+		WriteBufferSize: 1000000,
+		CheckOrigin: func(r *http.Request) bool {
+			return true
+		},
+		//将获取的参数放进这个数组
+		Subprotocols: []string{token},
+	}
+	ws, err := upGrader.Upgrade(c.Writer, c.Request, nil)
+	if err != nil {
+		err = errors.New("连接错误")
+		global.M2A_LOG.Error("将http连接升级为websocket连接失败", zap.Any("err", err))
+		return
+	}
+
+	client.SetQuoteWebSocket(ws)
+
+	return
+}