fix push error

Author: c-wind

client/client/client.go

@@ -194,18 +194,13 @@ func addFriend(c *candy.CandyClient, reader *bufio.Reader) {
 	data, _, _ = reader.ReadLine()
 	msg := string(data)
 
-	success, err := c.AddFriend(id, true, msg)
+	confirm, err := c.AddFriend(id, true, msg)
 	if err != nil {
 		log.Errorf("addFriend error:%v", err)
 		return
 	}
 
-	if !success {
-		log.Debugf("addFriend failure, userID:%v", userID)
-		return
-	}
-
-	log.Debugf("addFriend success, userID:%v", userID)
+	log.Debugf("addFriend confirm:%v, userID:%v", confirm, userID)
 	fmt.Println("==============================================")
 }
 

gate/server.go

@@ -286,6 +286,7 @@ func (g *Gate) SendMessage(ctx context.Context, req *meta.GateSendMessageRequest
 
 	//防止乱写
 	req.Msg.From = s.getID()
+	req.Msg.Method = meta.Method_NONE
 
 	if err = g.store.newMessage(req.Msg); err != nil {
 		return &meta.GateSendMessageResponse{Header: &meta.ResponseHeader{Code: -1, Msg: err.Error()}}, nil
@@ -301,9 +302,11 @@ func (g *Gate) Ready(msg *meta.Message, stream meta.Gate_ReadyServer) error {
 		return errors.Trace(err)
 	}
 
-	s.addStream(stream)
-
-	return nil
+	for {
+		m := <-s.push
+		log.Debugf("recv push request m:%v", m)
+		stream.Send(m)
+	}
 }
 
 // Heartbeat nil.
@@ -334,19 +337,9 @@ func (g *Gate) Notice(ctx context.Context, req *meta.GateNoticeRequest) (*meta.G
 			log.Debugf("User:%d offline", id.User)
 			continue
 		}
-
-		client := s.getStream()
-		if client == nil {
-			log.Errorf("User:%d client strem is nil", id.User)
-			continue
-		}
-
 		req.Msg.Before = id.Before
-
-		if err := client.Send(req.Msg); err != nil {
-			log.Errorf("client  Send msg:%v err:%v", req.Msg, err)
-			continue
-		}
+		log.Debugf("will push user:%d, msg:%v", id.User, req.Msg)
+		s.push <- req.Msg
 	}
 
 	log.Debugf("end PushID:%v msg:%v ok", req.ID, req.Msg)

gate/session.go

@@ -13,27 +13,16 @@ const (
 )
 
 type session struct {
-	id     int64
-	state  int
-	last   int64
-	addr   string
-	stream meta.Gate_ReadyServer
+	id    int64
+	state int
+	last  int64
+	addr  string
+	push  chan *meta.Message
 }
 
 func newSession(addr string) *session {
 	log.Debugf("addr:%s", addr)
-	return &session{addr: addr}
-}
-
-func (s *session) addStream(stream meta.Gate_ReadyServer) {
-	s.stream = stream
-}
-
-func (s *session) getStream() meta.Gate_ReadyServer {
-	if s.stream != nil {
-		return s.stream
-	}
-	return nil
+	return &session{addr: addr, push: make(chan *meta.Message)}
 }
 
 func (s *session) online(id int64) {

meta/common.pb.go

@@ -27,19 +27,22 @@ enum FriendRelation {
 }
 
 enum Method {
+
+    // 客户端发送上来的消息
+    NONE = 0;
     // 有人请求添加好友
-    FRIEND_ADD  = 0;
+    FRIEND_ADD  = 1;
     // 添加好友成功
-    FRIEND_CONFIRM = 1;
+    FRIEND_CONFIRM = 2;
     // 被删除好友
-    FRIEND_DEL  = 2;
+    FRIEND_DEL  = 3;
 
     // 申请加群
-    GROUP_ADD   = 3;
+    GROUP_ADD   = 4;
     // 退出群组
-    GROUP_EXIT  = 4;
+    GROUP_EXIT  = 5;
     // 邀请指定用户入群
-    GROUP_INVITE= 5;
+    GROUP_INVITE= 6;
 }
 
 // PushID 用户的上一条消息ID,用户ID