add push client queue

msgtransfer.go

@@ -46,3 +46,24 @@ func (m *msgReadTransferClient) Push(msg *types.MsgMarkRead) error {
 	}
 	return m.pusher.Push(context.Background(), string(data))
 }
+
+// 消息推送
+type MsgPushClientTransferClient interface {
+	Push(msg *types.MessagePushData) error
+}
+
+type msgPushClientTransferClient struct {
+	pusher *kq.Pusher
+}
+
+func NewMsgPushClientTransferClient(addr []string, topic string, opts ...kq.Pusher) MsgPushClientTransferClient {
+	return &msgPushClientTransferClient{pusher: kq.NewPusher(addr, topic)}
+}
+
+func (m *msgPushClientTransferClient) Push(msg *types.MessagePushData) error {
+	data, err := json.Marshal(msg)
+	if err != nil {
+		return err
+	}
+	return m.pusher.Push(context.Background(), string(data))
+}

types/message.go

@@ -1,5 +1,17 @@
 package types
 
+import "time"
+
+type FrameType uint8
+
+const (
+	FrameData  FrameType = 0x0
+	FramePing  FrameType = 0x1
+	FrameErr   FrameType = 0x9
+	FrameAck   FrameType = 0x2
+	FrameNoAck FrameType = 0x3
+)
+
 type (
 	Msg struct {
 		MType       `mapstructure:"mType"`
@@ -38,4 +50,16 @@ type (
 		ConversationId string   `mapstructure:"conversationId"`
 		MsgIds         []string `mapstructure:"msgIds"`
 	}
+
+	MessagePushData struct {
+		FrameType   `json:"frame_type"` //消息类型
+		Id          string              `json:"id"`
+		AckSeq      int                 `json:"ack_seq"`           // ack seq
+		AckTime     time.Time           `json:"ackTime"`           // ack时间
+		AccErrCount int                 `json:"errCount"`          // ack失败次数
+		Method      string              `json:"method,omitempty"`  // 消息处理方法
+		FormID      int64               `json:"form_id,omitempty"` // 接收者id
+		UserID      int64               `json:"user_id,omitempty"` // 发送者id
+		Data        interface{}         `json:"data,omitempty"`    // 消息内容
+	}
 )