MTP20_ACCESS/client/client.go

package client

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

	"github.com/gorilla/websocket"
	"github.com/mitchellh/mapstructure"
)

var Clients map[int]*Client // key:SessionID

type Client struct {
	LoginRedis

	mtx             sync.RWMutex
	curSerialNumber uint32                // 当前业务流水号
	asyncTasks      map[string]*AsyncTask // key:SessionId_FuncodeRsp_SerialNumber

	wsConn      *websocket.Conn             // 终端WebSocket连接
	quoteSubs   []request.QuoteSubscribeReq // 当前已订阅行情的商品
	ch          chan interface{}            // 接收实时行情订阅channel
	unSubscribe chan struct{}               // 取消订阅信号
	writeChan   chan []byte                 // 推送队列 QuoteServer -> Client
	wsCloseChan 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) SetQuoteSubs(req []request.QuoteSubscribeReq) {
	r.mtx.Lock()
	defer r.mtx.Unlock()

	r.quoteSubs = req

	if r.ch != nil {
		r.unSubscribe <- struct{}{}
		global.M2A_Publish.Unsubscribe(publish.Topic_Quote, r.ch)
	}
	r.ch = global.M2A_Publish.Subscribe(publish.Topic_Quote)

	go func() {
		for {
			select {
			case msg, ok := <-r.ch:
				if !ok {
					return // 管道已关闭，退出循环
				}
				// 向客户端发送行情信息
				if p, ok := msg.(*packet.MiQuotePacket); ok {
					DispatchRealQuote(p, r)
				}
			case <-r.unSubscribe:
				// 取消订阅信息
				return
			}
		}
	}()
}

// WriteWsBuf 向客户端发送实时行情
func (r *Client) WriteWsBuf(buf []byte) (err error) {
	r.writeChan <- buf

	return
}

// 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
}

func (r *Client) SetWebSocket(ws *websocket.Conn) (err error) {
	r.mtx.Lock()
	defer r.mtx.Unlock()

	if r.wsConn != nil {
		r.wsConn.Close()
	}
	r.wsConn = ws

	// 开始读取客户端发送信息
	go r.readClientWsMessage()
	// 开始推送客户端信息循环
	r.writeChan = make(chan []byte, 100)
	go r.writeClientWsMessage()

	return
}

// readClientWsMessage 处理终端发过来的websocket数据
// 注意: 阻塞式, 直到websocket关闭才退出
func (r *Client) readClientWsMessage() {
	for {
		mt, msg, err := r.wsConn.ReadMessage()
		if err != nil {
			fmt.Println(err)
			break
		}

		switch mt {
		case websocket.PingMessage:
			_ = r.wsConn.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.wsConn.WriteMessage(mt, msg)
		}
	}
}

// writeClientWsMessage 由于websocket非线程安全,
// 所以由统一协程写入
func (r *Client) writeClientWsMessage() {
	// defer r.close()
	for {
		select {
		case buf := <-r.writeChan:
			err := r.wsConn.WriteMessage(websocket.BinaryMessage, buf)
			if err != nil {
				return
			}
		case <-r.wsCloseChan: // 与终端连接关闭信息
			return
		}
	}
}

// clientToQuoteAgentMsg 处理客户端发上来的包
func (r *Client) clientToQuoteAgentMsg(msg []byte) error {
	var p packet.MiQuotePacket
	err := p.UnPackHead(msg)
	if err != nil {
		// logger.Logger().Errorf("[%v c->s] invalid packet: %v", r.ProxyId, err)
		return err
	}
	// 长度15 是心跳包, 目前只允许客户端上行心跳包
	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)
		if err != nil {
			return err
		}
	}

	return nil
}

// 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)
			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: 由于本服务改用短连，所以每次提交请交请求可能会不一样，后期可判断是否在中间件中进行拦截
}

// 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
}