package controllers import ( "fmt" "io" "log" "net/http" "os" "path" "sync" "time" "git.rosy.net.cn/jx-callback/business/jxstore/event" "git.rosy.net.cn/jx-callback/business/jxutils" "git.rosy.net.cn/jx-callback/business/jxutils/jxcontext" "git.rosy.net.cn/jx-callback/globals" "git.rosy.net.cn/jx-callback/business/model/dao" "git.rosy.net.cn/jx-callback/business/model" "github.com/gorilla/websocket" "git.rosy.net.cn/baseapi/utils" "github.com/astaxie/beego" ) // 操作事件明细相关API type EventController struct { beego.Controller } //连接的客户端,吧每个客户端都放进来 type WSClient struct { Clients map[int]map[string]*websocket.Conn ClientsHeart map[string]*websocket.Conn s *sync.RWMutex } var ( wsClient = &WSClient{} ) //广播频道(通道) var broadcast = make(chan *model.ImMessageRecord) // 配置升级程序(升级为websocket) var upgrader = websocket.Upgrader{} // 定义我们的消息对象 type Message struct { Data interface{} `json:"data"` } const ( audioPath = "/jxdata/cthrgw/dist/audio/" ) // @Title 测试websocket // @Description 测试websocket // @Success 200 {object} controllers.CallResult // @Failure 200 {object} controllers.CallResult // @router /TestWebsocket [get] func (c *EventController) TestWebsocket() { // 解决跨域问题(微信小程序) upgrader.CheckOrigin = func(r *http.Request) bool { return true } //升级将HTTP服务器连接升级到WebSocket协议。 //responseHeader包含在对客户端升级的响应中 //请求。使用responseHeader指定Cookie(设置Cookie)和 //应用程序协商的子目录(Sec WebSocket协议)。 //如果升级失败,则升级将向客户端答复一个HTTP错误 ws, err := upgrader.Upgrade(c.Ctx.ResponseWriter, c.Ctx.Request, nil) if err != nil { log.Fatal(err) } defer ws.Close() var ( userID = c.GetString("userID") clientUser = make(map[string]*websocket.Conn) ) if userID == "" { return } //将当前客户端放入map中 messageGroups, _ := dao.GetUserMessageGroups(dao.GetDB(), userID) if len(messageGroups) == 0 { return } wsClient.s.Lock() clientUser[userID] = ws wsClient.ClientsHeart[userID] = ws for _, v := range messageGroups { if len(wsClient.Clients[v.GroupID]) > 0 { wsClient.Clients[v.GroupID][userID] = ws } else { wsClient.Clients[v.GroupID] = clientUser } } wsClient.s.Unlock() db := dao.GetDB() if globals.IsProductEnv() { _, _, err = jxcontext.New(nil, c.GetString("token"), c.Ctx.ResponseWriter, c.Ctx.Request) if err != nil { msg := &CallResult{ Code: model.ErrCodeGeneralFailed, Desc: err.Error(), } ws.WriteJSON(&msg) } globals.SugarLogger.Debugf("TestWebsocket connection...") } globals.SugarLogger.Debugf("userID :%v ,clients :%v", userID, utils.Format4Output(wsClient.Clients, false)) c.EnableRender = false //Beego不启用渲染 var s *model.ImMessageRecord for { //接收客户端的消息 err := ws.ReadJSON(&s) if err != nil { globals.SugarLogger.Debugf("页面可能断开啦 ws.ReadJSON error: %v", err.Error()) wsClient.s.RLock() for k, _ := range wsClient.Clients { delete(wsClient.Clients[k], userID) } wsClient.s.RUnlock() wsClient.s.Lock() delete(wsClient.ClientsHeart, userID) wsClient.s.Unlock() // delete(clients, ws) //删除map中的客户端 break //结束循环 } else { //接受消息 业务逻辑 broadcast <- s if s.GroupID != 0 { //发聊天消息时,这个组所有的成员包括创建者都在userIDs里 userIDs := []string{} if results, err := dao.GetMessageGroups(db, "", s.GroupID, 0, true, ""); err == nil { for _, v := range results { userIDs = append(userIDs, v.UserID) for _, vv := range v.MessageGroupMembers { userIDs = append(userIDs, vv.UserID) } } } //如果这些人不在这个组的ws池子里就打上未读标记 for _, v := range userIDs { if wsClient.ClientsHeart[v] == nil { messageGroupReads, _ := dao.GetMessageGroupRead(db, v, s.GroupID) for _, vv := range messageGroupReads { vv.UnReadCount++ dao.UpdateEntity(db, vv, "UnReadCount") } } } } utils.CallFuncAsync(func() { if s.GroupID != 0 { dao.WrapAddIDCULDEntity(s, "") dao.CreateEntity(db, s) } }) } } } func init() { clients := make(map[int]map[string]*websocket.Conn) clientsHeart := make(map[string]*websocket.Conn) wsClient.Clients = clients wsClient.ClientsHeart = clientsHeart wsClient.s = new(sync.RWMutex) go handleMessages() } //广播推送消息 func handleMessages() { for { //读取通道中的消息 msg := <-broadcast if msg.GroupID == 0 { // globals.SugarLogger.Debugf("heart %v", utils.Format4Output(msg, false)) if wsClient.ClientsHeart[msg.UserID] != nil { if err := wsClient.ClientsHeart[msg.UserID].WriteJSON(&model.ImMessageRecord{ Key: "pang", }); err != nil { globals.SugarLogger.Debugf("heart client.WriteJSON error: %v", err) wsClient.ClientsHeart[msg.UserID].Close() //关闭 delete(wsClient.ClientsHeart, msg.UserID) } } } else { globals.SugarLogger.Debugf("clients len %v", len(wsClient.Clients)) //循环map客户端 for userID, client := range wsClient.Clients[msg.GroupID] { //把通道中的消息发送给客户端 user, err := dao.GetUser(dao.GetDB(), msg.UserID) if err == nil { msg.UserInfo = user } globals.SugarLogger.Debugf("msg %v", utils.Format4Output(msg, false)) if msg.CreatedAt == utils.ZeroTimeValue { msg.CreatedAt = time.Now() } err = client.WriteJSON(msg) if err != nil { globals.SugarLogger.Debugf("client.WriteJSON error: %v", err) client.Close() //关闭 delete(wsClient.Clients[msg.GroupID], userID) // delete(clients, client) //删除map中的客户端 } } } } } // @Title 查询聊天记录 // @Description 查询聊天记录 // @Param token header string true "认证token" // @Param groupID query int true "组ID" // @Param fromTime query string false "开始时间" // @Param toTime query string false "结束时间" // @Param offset query int false "门店列表起始序号(以0开始,缺省为0)" // @Param pageSize query int false "门店列表页大小(缺省为50,-1表示全部)" // @Success 200 {object} controllers.CallResult // @Failure 200 {object} controllers.CallResult // @router /GetImMessageRecord [get] func (c *EventController) GetImMessageRecord() { c.callGetImMessageRecord(func(params *tEventGetImMessageRecordParams) (retVal interface{}, errCode string, err error) { var db = dao.GetDB() retVal, err = dao.GetImMessageRecord(db, params.GroupID, "", 0, -1, utils.Str2Time(params.FromTime), utils.Str2Time(params.ToTime), params.Offset, params.PageSize) //清除此用户组所有的未读标记 if messageGroupReads, err := dao.GetMessageGroupRead(db, params.Ctx.GetUserID(), params.GroupID); err == nil { for _, v := range messageGroupReads { v.UnReadCount = 0 dao.UpdateEntity(db, v, "UnReadCount") } } return retVal, "", err }) } // @Title 用户未读消息设置 // @Description 用户未读消息设置,用户在退出登录,ws关闭,以及关闭小程序或app时调用 // @Param token header string true "认证token" // @Param payload query string true "messageGroupRead 格式[{"groupID":,"unReadCount":}]" // @Success 200 {object} controllers.CallResult // @Failure 200 {object} controllers.CallResult // @router /UpdateUserMessageGroupRead [post] func (c *EventController) UpdateUserMessageGroupRead() { c.callUpdateUserMessageGroupRead(func(params *tEventUpdateUserMessageGroupReadParams) (retVal interface{}, errCode string, err error) { var reads []*model.MessageGroupRead if err = jxutils.Strings2Objs(params.Payload, &reads); err == nil { err = event.UpdateUserMessageGroupRead(params.Ctx, reads) } return retVal, "", err }) } // @Title 发送聊天消息(限定系统消息) // @Description 发送聊天消息(限定系统消息) // @Param token header string true "认证token" // @Param payload formData string true "immessageRecord 类型" // @Success 200 {object} controllers.CallResult // @Failure 200 {object} controllers.CallResult // @router /SendSysMessage [post] func (c *EventController) SendSysMessage() { c.callSendSysMessage(func(params *tEventSendSysMessageParams) (retVal interface{}, errCode string, err error) { var imMessageRecord *model.ImMessageRecord if err = jxutils.Strings2Objs(params.Payload, &imMessageRecord); err == nil { err = event.SendSysMessage(params.Ctx, imMessageRecord) } return retVal, "", err }) } // @Title 创建聊天组 // @Description 创建聊天组 // @Param token header string true "认证token" // @Param userID formData string true "创建者id" // @Param userID2 formData string false "被拉的id 如果userID2为空就默认为是创建的群聊" // @Param name formData string false "如果是群聊,则要传入群名" // @Param dividePercentage formData int false "如果是群聊,则要传入分成比例" // @Param quitPrice formData int false "如果是群聊,则要传入退团金额" // @Success 200 {object} controllers.CallResult // @Failure 200 {object} controllers.CallResult // @router /CreateMessageGroup [post] func (c *EventController) CreateMessageGroup() { c.callCreateMessageGroup(func(params *tEventCreateMessageGroupParams) (retVal interface{}, errCode string, err error) { retVal, err = event.CreateMessageGroup(params.Ctx, params.UserID, params.UserID2, params.Name, params.DividePercentage, params.QuitPrice) return retVal, "", err }) } // @Title 查询某个用户所有聊天组 // @Description 查询某个用户所有聊天组 // @Param token header string true "认证token" // @Param userID query string true "userid" // @Success 200 {object} controllers.CallResult // @Failure 200 {object} controllers.CallResult // @router /GetMessageGroupByUser [get] func (c *EventController) GetMessageGroupByUser() { c.callGetMessageGroupByUser(func(params *tEventGetMessageGroupByUserParams) (retVal interface{}, errCode string, err error) { retVal, err = event.GetMessageGroupByUser(params.Ctx, params.UserID) return retVal, "", err }) } // @Title 查询聊天组 // @Description 查询聊天组 // @Param token header string true "认证token" // @Param groupID query int true "groupID" // @Param isMember query bool true "是否查询组员" // @Success 200 {object} controllers.CallResult // @Failure 200 {object} controllers.CallResult // @router /GetMessageGroups [get] func (c *EventController) GetMessageGroups() { c.callGetMessageGroups(func(params *tEventGetMessageGroupsParams) (retVal interface{}, errCode string, err error) { retVal, err = dao.GetMessageGroups(dao.GetDB(), "", params.GroupID, model.GroupTypeMulit, params.IsMember, "") return retVal, "", err }) } // @Title 加入用户组 // @Description 加入用户组 // @Param token header string true "认证token" // @Param groupID formData int true "组号" // @Param userID formData string true "被邀请人ID" // @Success 200 {object} controllers.CallResult // @Failure 200 {object} controllers.CallResult // @router /AddMessageGroup [post] func (c *EventController) AddMessageGroup() { c.callAddMessageGroup(func(params *tEventAddMessageGroupParams) (retVal interface{}, errCode string, err error) { err = event.AddMessageGroup(params.Ctx, params.GroupID, params.UserID) return retVal, "", err }) } // @Title 修改群组信息 // @Description 修改群组信息 // @Param token header string true "认证token" // @Param groupID formData int true "组号" // @Param payload formData string true "群组 payload" // @Success 200 {object} controllers.CallResult // @Failure 200 {object} controllers.CallResult // @router /UpdateMessageGroup [put] func (c *EventController) UpdateMessageGroup() { c.callUpdateMessageGroup(func(params *tEventUpdateMessageGroupParams) (retVal interface{}, errCode string, err error) { payload := make(map[string]interface{}) if err = utils.UnmarshalUseNumber([]byte(params.Payload), &payload); err == nil { retVal, err = event.UpdateMessageGroup(params.Ctx, params.GroupID, payload) } return retVal, "", err }) } // @Title 退出用户组(踢人) // @Description 退出用户组(踢人) // @Param token header string true "认证token" // @Param groupID formData int true "组号" // @Param userID formData string true "userID" // @Param flag formData bool false "是否是解散群" // @Success 200 {object} controllers.CallResult // @Failure 200 {object} controllers.CallResult // @router /DeleteMessageGroup [post] func (c *EventController) DeleteMessageGroup() { c.callDeleteMessageGroup(func(params *tEventDeleteMessageGroupParams) (retVal interface{}, errCode string, err error) { errCode, err = event.DeleteMessageGroup(params.Ctx, params.GroupID, params.UserID, params.Flag) return retVal, errCode, err }) } // @Title 转让群主 // @Description 转让群主 // @Param token header string true "认证token" // @Param groupID formData int true "组号" // @Param userID formData string true "userID" // @Success 200 {object} controllers.CallResult // @Failure 200 {object} controllers.CallResult // @router /TransferMessageGroupMaster [post] func (c *EventController) TransferMessageGroupMaster() { c.callTransferMessageGroupMaster(func(params *tEventTransferMessageGroupMasterParams) (retVal interface{}, errCode string, err error) { return retVal, "", err }) } // @Title 上传图片 // @Description 上传图片 // @Param token header string true "认证token" // @Success 200 {object} controllers.CallResult // @Failure 200 {object} controllers.CallResult // @router /UploadImg [post] func (c *EventController) UploadImg() { c.callUploadImg(func(params *tEventUploadImgParams) (retVal interface{}, errCode string, err error) { file, head, err := c.GetFile("rsmImg") defer file.Close() if head.Size > 1024*1024*5 { err = fmt.Errorf("图片太大,请重新选择!") } if path.Ext(head.Filename) != ".jpg" && path.Ext(head.Filename) != ".png" { err = fmt.Errorf("不支持的图片格式,请重新选择!") } if err != nil { return retVal, "", err } fileName := utils.GetUUID() + "_" + time.Now().Format("20060102") + path.Ext(head.Filename) c.SaveToFile("rsmImg", "/jxdata/cthrgw/dist/img/"+fileName) return "https://www.jxcs.net/img/" + fileName, "", err }) } // @Title 上传音频 // @Description 上传音频 // @Param token header string true "认证token" // @Success 200 {object} controllers.CallResult // @Failure 200 {object} controllers.CallResult // @router /UploadAudio [post] func (c *EventController) UploadAudio() { c.callUploadAudio(func(params *tEventUploadAudioParams) (retVal interface{}, errCode string, err error) { var ( timeStr = time.Now().Format("20060102") filePath = audioPath + timeStr ) files := c.Ctx.Request.MultipartForm.File["rsmAudio"] head := files[0] file, err := head.Open() defer file.Close() if path.Ext(head.Filename) != ".mp3" { err = fmt.Errorf("不支持的音频格式,请重新选择!") } if err != nil { return retVal, "", err } fileName := utils.GetUUID() + "_" + timeStr + path.Ext(head.Filename) if _, err = os.Stat(filePath); os.IsNotExist(err) { os.MkdirAll(filePath, os.ModePerm) os.Chmod(filePath, os.ModePerm) } f, err := os.OpenFile(filePath+"/"+fileName, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0666) if err != nil { return retVal, "", err } defer f.Close() io.Copy(f, file) return &model.ImMessageRecord{ Content: "https://www.jxcs.net/audio/" + timeStr + "/" + fileName, }, "", err }) } func GetOnlineUserCount() (count int) { wsClient.s.RLock() defer wsClient.s.RUnlock() count = len(wsClient.ClientsHeart) return count } // @Title 得到用户统计数据 // @Description 得到用户统计数据 // @Param token header string true "认证token" // @Success 200 {object} controllers.CallResult // @Failure 200 {object} controllers.CallResult // @router /GetUserStatistics [get] func (c *EventController) GetUserStatistics() { c.callGetUserStatistics(func(params *tEventGetUserStatisticsParams) (retVal interface{}, errCode string, err error) { retVal, err = event.GetUserStatistics(params.Ctx) result := retVal.(*event.GetUserStatisticsResult) result.OnlineUserCount = GetOnlineUserCount() return result, "", err }) }