tickserver/framework/msq/message.go

package msq

import "sync/atomic"
import "errors"
import "fmt"
import "time"
import "sync"

const (
	MsgInit = iota
	MsgShutDown
	MsgClose
	MsgMsqBindPrivate
	MsgMsqBindPublic
	MsgTimeout
	MsgEcho
	MsgLog
	MsgMatch
	MsgMulti
	MsgCheckReqId
	MsgCancelOrder
	MsgPlaceOrder
	MsgNotify
	MsgConnect
	MsgMarketInfo
	MsgLoadAccount
	MsgDeposit
	MsgWithdraw
	MsgWalletInfo
	MsgActiveOrderList
	MsgHistoryOrderList
	MsgGetCount
	MsgTopAccount
	MsgEmpty
	MsgTKDown
	MsgTKGetTicks
	MsgTKSub
	MsgTKHis
	MsgTK
	MsgTKS
	MsgInss
	MsgIns
    MsgGetPage
    MsgGetPageRange
    MsgDelete
    MsgGetSnapRange
    MsgRecoverSnap
    MsgSetConfig
    MsgGetConfig
	MsgCount
)

var msgname = []string{
	"MsgInit",
	"MsgShutDown",
	"MsgClose",
	"MsgMsqBindPrivate",
	"MsgMsqBindPublic",
	"MsgTimeout",
	"MsgEcho",
	"MsgLog",
	"MsgMatch",
	"MsgMulti",
	"MsgCheckReqId",
	"MsgCancelOrder",
	"MsgPlaceOrder",
	"MsgNotify",
	"MsgConnect",
	"MsgMarketInfo",
	"MsgLoadAccount",
	"MsgDeposit",
	"MsgWithdraw",
	"MsgWalletInfo",
	"MsgActiveOrderList",
	"MsgHistoryOrderList",
	"MsgGetCount",
	"MsgTopAccount",
	"MsgEmpty",
	"MsgTKDown",
	"MsgTKGetTicks",
	"MsgTKSub",
	"MsgTKHis",
	"MsgTK",
	"MsgTKS",
	"MsgInss",
	"MsgIns",
	"MsgGetPage",
	"MsgGetPageRange",
    "MsgDelete",
    "MsgGetSnapRange",
    "MsgRecoverSnap",
    "MsgSetConfig",
    "MsgGetConfig",
	"MsgCount",
}

var custommsg = make(map[int]string)

const (
	NeedLog = 1 << iota
	SendAsyn
	Public
	Private
)

var ErrTimeout = errors.New("timeout")
var messageId int64

func RegMessage(ty int, msgname string) error {
	if ty < MsgCount {
		return errors.New("message type less than msgcount error")
	}
	custommsg[ty] = msgname
	return nil
}

func getMsgName(ty int) string {
	if ty < MsgCount {
		return msgname[ty]
	}
	return custommsg[ty]
}

type Message struct {
	Type     int
	Index    int32
	Flag     int32
	ClientId int64
	Id       int64
	err      error
	SendTime int64
	RecvTime int64
	mu       sync.Mutex
	data     interface{}
	Ch       chan *Message
}

func NewMessage(mtype int, clientId int64, data interface{}, id int64) *Message {
	err := error(nil)
	if id == 0 {
		id = atomic.AddInt64(&messageId, 1)
	}
	if mtype == MsgTimeout {
		err = ErrTimeout
	}
	return &Message{mtype, 0, 0, clientId, id, err, 0, 0, sync.Mutex{}, data, nil}
}

func (msg *Message) SetErr(err error) {
	msg.mu.Lock()
	defer msg.mu.Unlock()
	msg.err  = err
}

func (msg *Message) GetErr() (err error) {
	msg.mu.Lock()
	defer msg.mu.Unlock()
	return msg.err
}

func (msg *Message) SetData(data interface{}) {
	msg.mu.Lock()
	defer msg.mu.Unlock()
	msg.data = data
}

func (msg *Message) GetData() interface{} {
	msg.mu.Lock()
	defer msg.mu.Unlock()
	return msg.data
}

func (msg *Message) GetId() int64 {
	return atomic.LoadInt64(&msg.Id)
}

func (msg *Message) SetId(id int64) {
	atomic.StoreInt64(&msg.Id, id)
}

func (msg *Message) Name() string {
	return getMsgName(msg.Type)
}

func (msg *Message) String() string {
	return "[" + getMsgName(msg.Type) + "]" + fmt.Sprint(msg.Id, msg.GetErr(), msg.GetData(), "index = ", msg.GetIndex())
}

func (msg *Message) IncIndex() int32 {
	return atomic.AddInt32(&msg.Index, 1)
}

func (msg *Message) GetIndex() int32 {
	return atomic.LoadInt32(&msg.Index)
}

func getTime() int64 {
	return time.Now().UnixNano()
}