MTP20_ACCESS/client/client.go

package client

import (
	"bytes"
	"encoding/json"
	"fmt"
	"mtp20access/global"
	rsp "mtp20access/model/mq/response"
	"mtp20access/packet"
	"mtp20access/publish"
	"sync"
	"time"

	"github.com/gorilla/websocket"
	"github.com/mitchellh/mapstructure"
	"go.uber.org/zap"
)

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

	wsTradeConn    *websocket.Conn  // 终端交易WebSocket连接
	tradeWriteChan chan []byte      // 推送队列 Access -> Client
	tradeChan      chan interface{} // 接收交易通知channel RabbitMQ -> Access
	// wsTradeCloseChan chan struct{}    // 终端WebSocket连接关闭信号
}

// GetSerialNumber 获取可用流水号
func (r *Client) GetSerialNumber() uint32 {
	r.mtx.Lock()
	defer func() {
		r.mtx.Unlock()
	}()
	r.curSerialNumber += 1

	return r.curSerialNumber
}

// SetQuoteSubs 设置交易通知订阅信息
func (r *Client) SetTradeNft() {
	// global.M2A_LOG.Info("SetTradeNft 11111111111111111111111111111")
	// r.tmtx.Lock()
	// defer r.tmtx.Unlock()

	// 从订阅中心订阅交易通知
	if r.tradeChan != nil {
		global.M2A_Publish.Unsubscribe(publish.Topic_Trading, r.tradeChan)
	}
	r.tradeChan = global.M2A_Publish.Subscribe(publish.Topic_Trading)
	global.M2A_LOG.Info("SetTradeNft 添加交易通知订阅", zap.Any("r.tradeChan", r.tradeChan))

	go func() {
		for {
			msg, ok := <-r.tradeChan
			if !ok {
				return // 管道已关闭，退出循环
			}
			// 向客户端发送交易通知信息
			if m, ok := msg.(map[string][]byte); ok {
				// 判断是否自己的通知
				if p, ok := m[r.SessionID]; ok {
					// 发送给客户端
					if r.tradeWriteChan != nil {
						r.tradeWriteChan <- p
					}
				}
			}
		}
	}()
}

// GetAsyncTask 获取目标异步任务
// key:SessionId_FuncodeRsp
func (r *Client) GetAsyncTask(key string) (asyncTask *AsyncTask) {
	r.mtx.RLock()
	defer func() {
		r.mtx.RUnlock()
	}()

	asyncTask, ok := r.asyncTasks[key]
	if ok {
		return asyncTask
	} else {
		return nil
	}
}

// SetAsyncTask 设置异步任务
func (r *Client) SetAsyncTask(asyncTask *AsyncTask, key string) {
	r.mtx.Lock()
	defer func() {
		r.mtx.Unlock()
	}()

	if r.asyncTasks == nil {
		r.asyncTasks = make(map[string]*AsyncTask, 0)
	}
	delete(r.asyncTasks, key)
	r.asyncTasks[key] = asyncTask
}

// DeleteAsyncTask 删除异步任务
func (r *Client) DeleteAsyncTask(key string) {
	r.mtx.Lock()
	defer func() {
		r.mtx.Unlock()
	}()
	delete(r.asyncTasks, key)
}

func (r *Client) GetAllAsyncTask() *map[string]*AsyncTask {
	return &r.asyncTasks
}

// **************** Trade WebSocket ****************

func (r *Client) SetTradeWebSocket(ws *websocket.Conn) (err error) {
	// global.M2A_LOG.Info("交易长连 SetTradeWebSocket 33333333333333333333")

	r.mtx.Lock()
	defer r.mtx.Unlock()

	// if r.wsTradeConn != nil {
	// 	r.wsTradeConn.Close()
	// }
	r.wsTradeConn = ws
	r.tradeWriteChan = make(chan []byte, 100)

	r.wsTradeConn.SetCloseHandler(func(code int, text string) error {
		global.M2A_LOG.Info("交易长连发生断开", zap.Any("LoginID", r.LoginID), zap.Any("SessionID", r.SessionID), zap.Any("code", code), zap.Any("text", text))
		return nil
	})

	// 开始读取客户端发送信息
	go r.readClientWsTradeMessage()
	// 开始推送客户端信息循环
	go r.writeClientWsTradeMessage()

	// 设置交易通知订阅信息
	r.SetTradeNft()

	global.M2A_LOG.Info("交易长连客户端接入", zap.Any("LoginID", r.LoginID), zap.Any("SessionID", r.SessionID))

	return
}

// readClientWsQuoteMessage 处理终端发过来的websocket数据
// 注意: 阻塞式, 直到websocket关闭才退出
func (r *Client) readClientWsTradeMessage() {
	for {
		fmt.Println("readClientWsTradeMessage start")

		// 40秒心跳超时
		r.wsTradeConn.SetReadDeadline(time.Now().Add(40 * time.Second))

		mt, msg, err := r.wsTradeConn.ReadMessage()
		if err != nil {
			global.M2A_LOG.Info("readClientWsTradeMessage::ReadMessage发生错误", zap.Any("err", err))
			return
		}

		switch mt {
		case websocket.PingMessage:
			if err = r.wsTradeConn.WriteMessage(mt, msg); err != nil {
				global.M2A_LOG.Info("readClientWsTradeMessage::PingMessage发生错误", zap.Any("err", err))
			}
		case websocket.CloseMessage:
			return
		case websocket.BinaryMessage:
			if err := r.clientToTradeAgentMsg(msg); err != nil {
				global.M2A_LOG.Info("readClientWsTradeMessage::BinaryMessage发生错误", zap.Any("err", err))
				return
			}
		case websocket.TextMessage:
			if err = r.wsTradeConn.WriteMessage(mt, msg); err != nil {
				global.M2A_LOG.Info("readClientWsTradeMessage::TextMessage发生错误", zap.Any("err", err))
				return
			}
		}
	}
}

// writeClientWsQuoteMessage 由于websocket非线程安全,
// 所以由统一协程写入
func (r *Client) writeClientWsTradeMessage() {
	for {
		buf, ok := <-r.tradeWriteChan
		if !ok {
			// 通道已关闭
			global.M2A_LOG.Info("writeClientWsTradeMessage 通道已关闭", zap.Any("LoginID", r.LoginID), zap.Any("SessionID", r.SessionID))
			return
		}
		err := r.wsTradeConn.WriteMessage(websocket.BinaryMessage, buf)
		if err != nil {
			global.M2A_LOG.Info("writeClientWsTradeMessage 向客户端发送信息时发生错误", zap.Any("LoginID", r.LoginID), zap.Any("SessionID", r.SessionID), zap.Any("err", err))
			return
		}
	}
}

// clientToTradeAgentMsg 处理客户端发上来的包
func (r *Client) clientToTradeAgentMsg(msg []byte) error {
	var p packet.MiPacket
	err := p.UnPackHead(msg)
	if err != nil {
		// logger.Logger().Errorf("[%v c->s] invalid packet: %v", r.ProxyId, err)
		return err
	}
	// 长度24 是心跳包, 目前只允许客户端上行心跳包
	if p.Length == 24 {
		// logger.Logger().Debugf("%v [%v c->s] %s", r.ClientIP, r.ProxyId, p.HeaderInfo())
		// 将心跳发回给客户端
		err = r.wsTradeConn.WriteMessage(websocket.BinaryMessage, msg)
		if err != nil {
			return err
		}
	}

	return nil
}

// **************** RabbitMQ ****************

// MQPacket 与总线交互的数据体
type MQPacket struct {
	FunCode   uint32  // 功能码
	SessionId uint32  // 数据包的sid
	Data      *[]byte // 业务数据体
}

// AsyncTask 异步任务结构体
type AsyncTask struct {
	PacketRsp    chan MQPacket // 总线数据处理通道
	FuncodeRsp   uint32        // 回复功能码
	SerialNumber uint32        // 通信流水号
	Own          *Client
	IsEncrypted  bool // 是否加密

	Rsp     chan rsp.MQBodyRsp // 回调
	doClose sync.Once          // 仅关闭通道一次
}

// Finish 完成
func (r *AsyncTask) Finish() {
	r.doClose.Do(
		func() {
			close(r.Rsp)
			key := fmt.Sprintf("%v_%v_%v", r.Own.SessionID, r.FuncodeRsp, r.SerialNumber)
			// 银行服务相关的回复流水号是错误的，所以需要特殊处理
			if int(r.FuncodeRsp) == global.T2bBankSignRsp ||
				int(r.FuncodeRsp) == global.T2bSMSVerificationCodeRsp ||
				int(r.FuncodeRsp) == global.T2bBankCancelSignRsp ||
				int(r.FuncodeRsp) == global.T2bBankWithdrawRsp ||
				int(r.FuncodeRsp) == global.T2bBankDepositRsp {
				key = fmt.Sprintf("%v_%v", r.Own.SessionID, r.FuncodeRsp)
			}
			r.Own.DeleteAsyncTask(key)
		})
}

type LoginRedis struct {
	LoginID   string `json:"loginId" redis:"loginId"`     // 登陆账号
	UserID    string `json:"userId" redis:"userId"`       // 用户ID
	SessionID string `json:"sessionId" redis:"sessionId"` // 终端sid
	Token     string `json:"token" redis:"token"`         // 令牌
	Group     string `json:"group" redis:"group"`         // 终端分组
	Addr      string `json:"addr" redis:"addr"`           // 客户端地址信息 // FIXME: 由于本服务改用短连，所以每次提交请交请求可能会不一样，后期可判断是否在中间件中进行拦截
	OldToken  string `json:"-" redis:"-"`                 // 旧接入Token
}

// FromMap Map to Struct
func (r *LoginRedis) FromMap(val map[string]interface{}) error {
	return mapstructure.Decode(val, r)
}

// ToMap Struct to Map
func (r *LoginRedis) ToMap() (val map[string]interface{}, err error) {
	if marshalContent, err := json.Marshal(r); err != nil {
		return nil, err
	} else {
		d := json.NewDecoder(bytes.NewReader(marshalContent))
		d.UseNumber() // 设置将float64转为一个number
		if err := d.Decode(&val); err != nil {
			fmt.Println(err)
		} else {
			for k, v := range val {
				val[k] = v
			}
		}
	}

	return
}