// 注册/push路由
func initPushRouter(r *gin.Engine) {
	pushGroup := r.Group("/push")
	pushGroup.Use(CheckSessionId())
	{
		pushGroup.POST("/push", handler.Push)
		pushGroup.POST("/pushRoom", handler.PushRoom) // 发消息到聊天室
		pushGroup.POST("/count", handler.Count)
		pushGroup.POST("/getRoomInfo", handler.GetRoomInfo)
	}

}

func PushRoom(c *gin.Context) {

	// 获取 http 请求的form数据
	var formRoom FormRoom
	if err := c.ShouldBindBodyWith(&formRoom, binding.JSON); err != nil {
		tools.FailWithMsg(c, err.Error())
		return
	}
	authToken := formRoom.AuthToken
	msg := formRoom.Msg
	roomId := formRoom.RoomId
	checkAuthReq := &proto.CheckAuthRequest{AuthToken: authToken}

	// rpc 请求 logic服务（校验 authToken），返回 userid
	authCode, fromUserId, fromUserName := rpc.RpcLogicObj.CheckAuth(checkAuthReq)
	if authCode == tools.CodeFail {
		tools.FailWithMsg(c, "rpc fail get self info")
		return
	}

	req := &proto.Send{
		Msg:          msg,               // 用户发的消息
		FromUserId:   fromUserId,        // 用户id
		FromUserName: fromUserName,      // 用户名
		RoomId:       roomId,            // 房间id
		Op:           config.OpRoomSend, // 向房间发送
	}

	// rpc 请求 logic服务
	code, msg := rpc.RpcLogicObj.PushRoom(req)
	if code == tools.CodeFail {
		tools.FailWithMsg(c, "rpc push room msg fail!")
		return
	}
	// 回复前端，发送ok
	tools.SuccessWithMsg(c, "ok", msg)
}

func (rpc *RpcLogic) PushRoom(req *proto.Send) (code int, msg string) {
	reply := &proto.SuccessReply{}
	// 通过rpc 调用logic服务中的PushRoom函数，代码位于logic/rpc.go
	LogicRpcClient.Call(context.Background(), "PushRoom", req, reply)
	code = reply.Code
	msg = reply.Msg
	return
}

func (rpc *RpcLogic) PushRoom(ctx context.Context, args *proto.Send, reply *proto.SuccessReply) (err error) {
	reply.Code = config.FailReplyCode
	sendData := args
	roomId := sendData.RoomId
	logic := new(Logic)

	roomUserKey := logic.getRoomUserKey(strconv.Itoa(roomId))      // gochat_room_xxxx
	roomUserInfo, err := RedisClient.HGetAll(roomUserKey).Result() // 从redis中获取房间信息
	if err != nil {
		logrus.Errorf("logic,PushRoom redis hGetAll err:%s", err.Error())
		return
	}

	var bodyBytes []byte
	sendData.RoomId = roomId                     // 房间id
	sendData.Msg = args.Msg                      // 发送的消息
	sendData.FromUserId = args.FromUserId        // 发送人id
	sendData.FromUserName = args.FromUserName    // 发送人名
	sendData.Op = config.OpRoomSend              // 向房间发送
	sendData.CreateTime = tools.GetNowDateTime() // 时间
	bodyBytes, err = json.Marshal(sendData)      // 转成json字符串

	if err != nil {
		logrus.Errorf("logic,PushRoom Marshal err:%s", err.Error())
		return
	}
	// 保存到redis链表 gochat_queue 中
	err = logic.RedisPublishRoomInfo(roomId, len(roomUserInfo), roomUserInfo, bodyBytes)
	if err != nil {
		logrus.Errorf("logic,PushRoom err:%s", err.Error())
		return
	}
	reply.Code = config.SuccessReplyCode
	return
}

func (logic *Logic) RedisPublishRoomInfo(roomId int, count int, RoomUserInfo map[string]string, msg []byte) (err error) {
	var redisMsg = &proto.RedisMsg{ // 这里又包装了一层信息
		Op:           config.OpRoomSend, // 向房间发送
		RoomId:       roomId,            // 房间id
		Count:        count,             // 房间人数
		Msg:          msg,               // 这个就是外部的json字符串后的消息
		RoomUserInfo: RoomUserInfo,      // 房间信息
	}

	redisMsgByte, err := json.Marshal(redisMsg) // 有进行了一次 json 字符串
	if err != nil {
		logrus.Errorf("logic,RedisPublishRoomInfo redisMsg error : %s", err.Error())
		return
	}

	// 保存到 redis 链表 gochat_queue 中 ， 通过 LRANGE gochat_queue 0 -1 命令行在redis中，可以查看其中的消息
	err = RedisClient.LPush(config.QueueName, redisMsgByte).Err()

	if err != nil {
		logrus.Errorf("logic,RedisPublishRoomInfo redisMsg error : %s", err.Error())
		return
	}
	return
}

func (task *Task) InitQueueRedisClient() (err error) {

	// redis的配置信息
	redisOpt := tools.RedisOption{
		Address:  config.Conf.Common.CommonRedis.RedisAddress,
		Password: config.Conf.Common.CommonRedis.RedisPassword,
		Db:       config.Conf.Common.CommonRedis.Db,
	}

	// 获取Redis客户端
	RedisClient = tools.GetRedisInstance(redisOpt)

	// 检测是否ping的通
	if pong, err := RedisClient.Ping().Result(); err != nil {
		logrus.Infof("RedisClient Ping Result pong: %s,  err: %s", pong, err)
	}
	go func() {
		for { // 死循环
			var result []string
			//10s timeout
			result, err = RedisClient.BRPop(time.Second*10, config.QueueName).Result()
			if err != nil {
				logrus.Infof("task queue block timeout,no msg err:%s", err.Error())
			}
			// 从 redis 链表gochat_queue中，读取元素，保存到 task.Push 中
			if len(result) >= 2 {
				task.Push(result[1])
			}
		}
	}()
	return
}

func (task *Task) Push(msg string) {

	// 解析redis中的数据包
	m := &proto.RedisMsg{}
	if err := json.Unmarshal([]byte(msg), m); err != nil {
		logrus.Infof(" json.Unmarshal err:%v ", err)
	}
	logrus.Infof("push msg info %d,op is:%d", m.RoomId, m.Op)
	switch m.Op {
	case config.OpSingleSend: // 发送到个人的数据包
		// 随机选一个 pushChannel 保存 PushParams （ 处理的逻辑在👆上面的 task.pushSingleToConnect 函数中）
		pushChannel[rand.Int()%config.Conf.Task.TaskBase.PushChan] <- &PushParams{
			ServerId: m.ServerId,
			UserId:   m.UserId,
			Msg:      m.Msg,
		}
	case config.OpRoomSend: // 发送到房间中的数据包
		task.broadcastRoomToConnect(m.RoomId, m.Msg)
	case config.OpRoomCountSend:
		task.broadcastRoomCountToConnect(m.RoomId, m.Count)
	case config.OpRoomInfoSend:
		task.broadcastRoomInfoToConnect(m.RoomId, m.RoomUserInfo)
	}
}

func (task *Task) broadcastRoomToConnect(roomId int, msg []byte) {
	pushRoomMsgReq := &proto.PushRoomMsgRequest{
		RoomId: roomId,
		Msg: proto.Msg{
			Ver:       config.MsgVersion,
			Operation: config.OpRoomSend,
			SeqId:     tools.GetSnowflakeId(),
			Body:      msg,
		},
	}
	reply := &proto.SuccessReply{}
	rpcList := RClient.GetAllConnectTypeRpcClient()
	for _, rpc := range rpcList {
		logrus.Infof("broadcastRoomToConnect rpc  %v", rpc)

		// 通过rpc请求，调用 connect服务中的 PushRoomMsg 函数
		rpc.Call(context.Background(), "PushRoomMsg", pushRoomMsgReq, reply)
		logrus.Infof("reply %s", reply.Msg)
	}
}

func (rpc *RpcConnectPush) PushRoomMsg(ctx context.Context, pushRoomMsgReq *proto.PushRoomMsgRequest, successReply *proto.SuccessReply) (err error) {
	successReply.Code = config.SuccessReplyCode
	successReply.Msg = config.SuccessReplyMsg
	logrus.Infof("PushRoomMsg msg % v", pushRoomMsgReq)

	// 【注意】这里对所有的buckets；原因在于 同一个roomid 可以存在于不同的 Buckets中
	for _, bucket := range DefaultServer.Buckets {
		bucket.BroadcastRoom(pushRoomMsgReq)
	}
	return
}

func (b *Bucket) BroadcastRoom(pushRoomMsgReq *proto.PushRoomMsgRequest) {
	// 按照轮询的方式,选择 b.routines
	num := atomic.AddUint64(&b.routinesNum, 1) % b.bucketOptions.RoutineAmount
	b.routines[num] <- pushRoomMsgReq
}

func NewBucket(bucketOptions BucketOptions) (b *Bucket) {
	b = new(Bucket)
	b.chs = make(map[int]*Channel, bucketOptions.ChannelSize) // 1024 Channel 对象
	b.bucketOptions = bucketOptions
	b.routines = make([]chan *proto.PushRoomMsgRequest, bucketOptions.RoutineAmount) // 32 个 channel
	b.rooms = make(map[int]*Room, bucketOptions.RoomSize)                            // 1024 Room 房间

	for i := uint64(0); i < b.bucketOptions.RoutineAmount; i   {
		c := make(chan *proto.PushRoomMsgRequest, bucketOptions.RoutineSize) // 初始化chan，容量 20
		b.routines[i] = c
		go b.PushRoom(c) // 启动32个协程，消费32个chan
	}
	return
}


func (b *Bucket) PushRoom(ch chan *proto.PushRoomMsgRequest) {
	for {
		var (
			arg  *proto.PushRoomMsgRequest
			room *Room
		)
		arg = <-ch
		// 基于消息中的房间id， b.Room(arg.RoomId) 获取 *Room 房间对象
		if room = b.Room(arg.RoomId); room != nil { // room != nil 在该bucket下存在房间的时候，才会向房间中发送消息
			room.Push(&arg.Msg)
		}
	}
}


func (r *Room) Push(msg *proto.Msg) {
	r.rLock.RLock()

	// 遍历房间中的所有的客户端socket（tcp or websocket）
	for ch := r.next; ch != nil; ch = ch.Next {
		// 将消息发送给客户端
		if err := ch.Push(msg); err != nil {
			logrus.Infof("push msg err:%s", err.Error())
		}
	}
	r.rLock.RUnlock()
}


// 这里没有直接调用 conn/connTcp 发送，而是放到 broadcast 通道中（writeDataToTcp/writePump 函数读取broadcast将数据发送出去 ）
func (ch *Channel) Push(msg *proto.Msg) (err error) {
	select {
	case ch.broadcast <- msg:
	default: // 如果消息很多，把broadcast容量打满了，就存在消息的丢弃。。。
	}
	return
}

// 调用一次serveWs函数，就建立一个新的【客户端和服务器的】socket长连接
func (c *Connect) serveWs(server *Server, w http.ResponseWriter, r *http.Request) {

	//... 省略 ...

	// 创建一个 *Channel对象：里面保存的就是【客户端的连接句柄】
	ch := NewChannel(server.Options.BroadcastSize) // 512
	ch.conn = conn

	// 启动协程，死循环发送数据
	go server.writePump(ch, c)
	
}


// 负责读取 ch.broadcast 中的数据，然后通过websocket发送给客户端
func (s *Server) writePump(ch *Channel, c *Connect) {
	//... 省略 ...

	// 死循环
	for {
		select {
		// 读取 ch.broadcast 数据
		case message, ok := <-ch.broadcast:
			//write data dead time , like http timeout , default 10s
			ch.conn.SetWriteDeadline(time.Now().Add(s.Options.WriteWait))
			if !ok { // 说明 ch.broadcast 关闭了
				logrus.Warn("SetWriteDeadline not ok")
				ch.conn.WriteMessage(websocket.CloseMessage, []byte{})
				return
			}

			// 写入文本
			w, err := ch.conn.NextWriter(websocket.TextMessage)
			if err != nil {
				logrus.Warnf(" ch.conn.NextWriter err :%s  ", err.Error())
				return
			}
			logrus.Infof("message write body:%s", message.Body)
			// 向前端发送数据
			w.Write(message.Body)
			if err := w.Close(); err != nil {
				return
			}
		case <-ticker.C:
			ch.conn.SetWriteDeadline(time.Now().Add(s.Options.WriteWait))
			logrus.Infof("websocket.PingMessage :%v", websocket.PingMessage)
			// ping测试下 websocket连接是否已经断开
			if err := ch.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
				return
			}
		}
	}
}