cron/cron.go

@@ -16,4 +16,5 @@ func Init() {
 	mic_cron.OnMicCheck()           // 检查上麦
 	group_cron.GroupPowerExpClear() // 清理家族经验/等级
 	group_cron.GroupPowerMonthRankAct()
+	group_cron.GroupInEventInit() // 进房事件
 }

cron/group_cron/group_in.go

+package group_cron
+
+import (
+	"git.hilo.cn/hilo-common/mycontext"
+	"git.hilo.cn/hilo-common/mylogrus"
+	"git.hilo.cn/hilo-common/resource/config"
+	"hilo-group/domain/service/event_s"
+	"time"
+)
+
+// 进房事件
+func GroupInEventInit() {
+	if !config.IsMaster() {
+		return
+	}
+	mylogrus.MyLog.Infof("GroupInEventInit")
+	go func() {
+		ticker := time.NewTicker(time.Millisecond * 500)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ticker.C:
+				//start := time.Now()
+				myCtx := mycontext.CreateMyContext(nil)
+				// 消费进房事件
+				if err := event_s.NewGroupInEventService(myCtx).Consume(); err != nil {
+					myCtx.Log.Errorf("groupInEvent consume fail:%v", err)
+				} else {
+					//myCtx.Log.Infof("groupInEvent consume success,cost:%v", time.Now().Sub(start))
+				}
+			}
+		}
+	}()
+}

domain/event/group_ev/tx_group_in.go

+package group_ev
+
+import (
+	"git.hilo.cn/hilo-common/domain"
+	"git.hilo.cn/hilo-common/resource/mysql"
+)
+
+var txGroupInListen = new(domain.EventBase)
+
+// 进房事件
+type TxGroupInEvent struct {
+	GroupId    string // imGroupId
+	UserId     mysql.ID
+	ExternalId string
+	Nick       string
+	Avatar     string
+	IsMember   bool //是否永久成员
+	IsVip      bool
+	NobleLevel uint16
+}
+
+func AddTxGroupInEventSync(callback func(model *domain.Model, event interface{}) error) {
+	domain.AddEventSync(txGroupInListen, callback)
+}
+
+func AddTxGroupInEventAsync(callback func(model *domain.Model, event interface{}) error) {
+	domain.AddEventAsync(txGroupInListen, callback)
+}
+
+func PublishTxGroupInEvent(model *domain.Model, event interface{}) error {
+	return domain.PublishEvent(txGroupInListen, model, event)
+}

domain/model/event_m/group_in.go

+package event_m
+
+import (
+	"git.hilo.cn/hilo-common/domain"
+	"git.hilo.cn/hilo-common/resource/mysql"
+	"gorm.io/gorm"
+)
+
+// 进房事件消息
+type EventGroupIn struct {
+	mysql.Entity
+	*domain.Model `gorm:"-"`
+	Proto         mysql.Type
+	Payload       []byte
+	Mark          mysql.YesNo
+	MarkHiloGroup mysql.YesNo
+}
+
+func (EventGroupIn) TableName() string {
+	return "event_group_in"
+}
+
+// 偏移值
+type EventGroupInOffsetHiloGroup struct {
+	mysql.Entity
+	*domain.Model `gorm:"-"`
+	MarkOffset    mysql.ID
+}
+
+// 获取当前偏移值
+func GroupOffset(model *domain.Model) (*EventGroupInOffsetHiloGroup, error) {
+	offset := new(EventGroupInOffsetHiloGroup)
+	if err := model.Db.WithContext(model).First(offset).Error; err != nil {
+		if err != gorm.ErrRecordNotFound {
+			model.Log.Errorf("Offset fail:%v", err)
+			return nil, err
+		}
+		// gorm.ErrRecordNotFound
+	}
+	offset.Model = model
+	return offset, nil
+}
+
+// 批量获取进房事件
+func FetchEventGroupIn(model *domain.Model, limit int) ([]*EventGroupIn, *EventGroupInOffsetHiloGroup, error) {
+	offset, err := GroupOffset(model)
+	if err != nil {
+		return nil, nil, err
+	}
+	var events []*EventGroupIn
+	if err := model.Db.WithContext(model).Model(EventGroupIn{}).
+		Where("id > ?", offset.MarkOffset).
+		Order("id asc").Limit(limit).Find(&events).Error; err != nil {
+		model.Log.Errorf("FetchEventGroupIn fail:%v", err)
+		return nil, nil, err
+	}
+	return events, offset, nil
+}
+
+// 标记已完成
+func (p *EventGroupIn) MarkDone() error {
+	p.MarkHiloGroup = mysql.YES
+	return p.Db.WithContext(p.Model).Model(EventGroupIn{}).Where("id = ?", p.ID).Update("mark_hilo_group", p.MarkHiloGroup).Limit(1).Error
+}

domain/model/event_m/repo.go

@@ -9,3 +9,7 @@ func (p *EventGiftSendOffsetHiloGroup) Persistence() error {
 func (p *EventGiftSend) Persistence() error {
 	return model.Persistent(p.Db, p)
 }
+
+func (p *EventGroupInOffsetHiloGroup) Persistence() error {
+	return model.Persistent(p.Db, p)
+}
\ No newline at end of file

domain/model/group_m/mic.go

@@ -98,8 +98,8 @@ func GetMicNumType(model *domain.Model, groupUuid string) (group_e.GroupMicNumTy
 	}
 }
 
-//6个小时
-const expireMinute = 60 * 60 * 12
+// 3天
+const expireMinute = 60 * 60 * 24 * 3
 const micExpire = expireMinute * time.Second
 
 //麦位基本信息。

domain/model/group_m/micData.go

@@ -259,9 +259,9 @@ func MicAllRPush(model *domain.Model, groupUid string, externalId string) error
 	}
 
 	//
-	micContents, err := getMicAllContent(model, groupUid)
+	micContents, err := GetMicAllContent(model, groupUid)
 	if err != nil {
-		model.Log.Errorf("MicChangeRPush MicAllRPush getMicAllContent  err:%+v, micContents:%v groupUuid:%v, externalId:%v", err, micContents, groupUid, externalId)
+		model.Log.Errorf("MicChangeRPush MicAllRPush GetMicAllContent  err:%+v, micContents:%v groupUuid:%v, externalId:%v", err, micContents, groupUid, externalId)
 		return err
 	}
 	for _, micContent := range micContents {
@@ -319,7 +319,7 @@ func MicRPush(model *domain.Model, txGroupId string, msg GroupSystemMsg) error {
 }
 
 //得使用旧的imGroupId
-func getMicAllContent(model *domain.Model, groupUid string) ([]MicContent, error) {
+func GetMicAllContent(model *domain.Model, groupUid string) ([]MicContent, error) {
 
 	txGroupId, err := ToTxGroupId(model, groupUid)
 	if err != nil {
@@ -340,6 +340,9 @@ func getMicAllContent(model *domain.Model, groupUid string) ([]MicContent, error
 	userIds := make([]uint64, 0, len(micUsers))
 	for _, r := range micUsers {
 		userIds = append(userIds, r.UserId)
+		if r.CpUserId > 0 {
+			userIds = append(userIds, r.CpUserId)
+		}
 	}
 
 	model.Log.Infof("MicChangeRPush getMicAllContent groupUid:%v, userIds:%+v", groupUid, userIds)
@@ -350,6 +353,13 @@ func getMicAllContent(model *domain.Model, groupUid string) ([]MicContent, error
 
 	micContents := make([]MicContent, 0, len(mics))
 	for _, r := range mics {
+		var micEffect string
+		cpUserId := micUserMap[r.I].CpUserId
+		micUserData := micUserDataMap[micUserMap[r.I].UserId]
+		if cpUserId > 0 {
+			micEffect = "https://image.whoisamy.shop/hilo/resource/svga/mic_effect_cp.svga"
+			micUserData.MicEffect = micEffect
+		}
 		micContents = append(micContents, MicContent{
 			GroupId:    txGroupId,
 			I:          r.I,
@@ -359,7 +369,7 @@ func getMicAllContent(model *domain.Model, groupUid string) ([]MicContent, error
 			ExternalId: micUserMap[r.I].ExternalId,
 			AgoraId:    uint32(micUserMap[r.I].UserId),
 			Timestamp:  time.Now().UnixNano(),
-			User:       micUserDataMap[micUserMap[r.I].UserId],
+			User:       micUserData,
 		})
 	}
 	return micContents, nil
@@ -448,6 +458,7 @@ func getMicUserDatas(model *domain.Model, userIds []uint64) (map[uint64]*MicUser
 		return nil, err
 	}
 	svips, _ := rpc.MGetUserSvip(model, userIds)
+	cpRelations, _ := rpc.MGetUserCpRelations(model, userIds)
 
 	micUserDataMap := map[uint64]*MicUserData{}
 	for _, id := range userIds {
@@ -461,11 +472,11 @@ func getMicUserDatas(model *domain.Model, userIds []uint64) (map[uint64]*MicUser
 		}
 		var headwearPicUrl string
 		var headwearEffectUrl string
-		var reverseHeadwearEffectUrl string
+		var headwearReverseEffectUrl string
 		if headwearUser, flag := headwearMap[id]; flag {
 			headwearPicUrl = resHeadwearMap[headwearUser.HeadwearId].PicUrl
 			headwearEffectUrl = resHeadwearMap[headwearUser.HeadwearId].EffectUrl
-			reverseHeadwearEffectUrl = resHeadwearMap[headwearUser.HeadwearId].ReverseEffectUrl
+			headwearReverseEffectUrl = resHeadwearMap[headwearUser.HeadwearId].ReverseEffectUrl
 		}
 		micUserDataMap[id] = &MicUserData{
 			Id:                       user.ID,
@@ -478,9 +489,10 @@ func getMicUserDatas(model *domain.Model, userIds []uint64) (map[uint64]*MicUser
 			NobleLeave:               nobleMap[id],
 			HeadwearPicUrl:           headwearPicUrl,
 			HeadwearEffectUrl:        headwearEffectUrl,
-			HeadwearReverseEffectUrl: reverseHeadwearEffectUrl,
+			HeadwearReverseEffectUrl: headwearReverseEffectUrl,
 			SvipLevel:                svips[id].SvipLevel,
 			Svip:                     rpc.CopySimpleSvip(svips[id]),
+			HeadwearIcon:             cpRelations[user.ID].CpUserAvatar,
 		}
 	}
 	return micUserDataMap, nil

domain/service/event_s/event_init.go

@@ -34,8 +34,9 @@ func EventInit() {
 	GroupEvents()
 	GroupImMass()
 	GroupTheme()
-	SendGift() // 送礼事件
-	OnMic()    // 在麦上事件
+	SendGift()              // 送礼事件
+	OnMic()                 // 在麦上事件
+	GroupInMicChangeEvent() // 用户进房推送mic位置信息
 }
 
 func GroupSupportEvents() {

domain/service/event_s/group_in.go

+package event_s
+
+import (
+	"encoding/json"
+	"git.hilo.cn/hilo-common/domain"
+	"git.hilo.cn/hilo-common/mycontext"
+	"git.hilo.cn/hilo-common/resource/mysql"
+	"golang.org/x/sync/errgroup"
+	"hilo-group/domain/event/group_ev"
+	"hilo-group/domain/model/event_m"
+	"runtime/debug"
+)
+
+type GroupInEventService struct {
+	svc *domain.Service
+}
+
+func NewGroupInEventService(myContext *mycontext.MyContext) *GroupInEventService {
+	svc := domain.CreateService(myContext)
+	return &GroupInEventService{svc}
+}
+
+//
+func (s *GroupInEventService) Consume() error {
+	defer func() {
+		if err := recover(); err != nil {
+			s.svc.Log.Errorf("ExceptionHandle GroupInEventService Consume SYSTEM ACTION PANIC: %v, stack: %v", err, string(debug.Stack()))
+		}
+	}()
+	var model = domain.CreateModel(s.svc.CtxAndDb)
+	events, offset, err := event_m.FetchEventGroupIn(model, BatchCount)
+	if err != nil {
+		return err
+	}
+	var wg errgroup.Group
+	for k := range events {
+		cpEvent := &event_m.EventGroupIn{
+			Entity: mysql.Entity{
+				ID:          events[k].ID,
+				CreatedTime: events[k].CreatedTime,
+				UpdatedTime: events[k].UpdatedTime,
+			},
+			Proto:         events[k].Proto,
+			Payload:       events[k].Payload,
+			Mark:          events[k].Mark,
+			MarkHiloGroup: events[k].MarkHiloGroup,
+		}
+		wg.Go(func() error {
+			if cpEvent.MarkHiloGroup == mysql.YES {
+				model.Log.Warnf("already consume msg :%v", cpEvent)
+				return nil
+			}
+			groupInEvent := new(group_ev.TxGroupInEvent)
+			if err := json.Unmarshal(cpEvent.Payload, groupInEvent); err != nil {
+				model.Log.Errorf("json msg fail,event:%v,err:%v", cpEvent, err)
+				return nil
+			}
+			// 发布事件
+			if err := group_ev.PublishTxGroupInEvent(model, groupInEvent); err != nil {
+				model.Log.Errorf("PublishTxGroupInEvent,event:%v,err:%v", cpEvent, err)
+				return err
+			}
+			// 标记已经处理
+			cpEvent.Model = model
+			err = cpEvent.MarkDone()
+			if err != nil {
+				model.Log.Errorf("consume msg fail,event:%v,err:%v", cpEvent, err)
+			}
+			return err
+		})
+	}
+	err = wg.Wait()
+	if err != nil {
+		model.Log.Errorf("batch consume msg has fail,event,err:%v", err)
+		// 暂时先允许丢数据,继续mark offset
+	}
+	// 最后一次提交offset
+	if len(events) > 0 {
+		offset.MarkOffset = events[len(events)-1].ID
+		return offset.Persistence()
+	}
+	return nil
+}

domain/service/event_s/mic_change.go

+package event_s
+
+import (
+	"encoding/json"
+	"git.hilo.cn/hilo-common/domain"
+	"git.hilo.cn/hilo-common/protocol/userProxy"
+	"git.hilo.cn/hilo-common/rpc"
+	uuid "github.com/satori/go.uuid"
+	"hilo-group/domain/event/group_ev"
+	"hilo-group/domain/model/group_m"
+	"hilo-group/myerr/bizerr"
+)
+
+// 用户进房后推送mic信息
+func GroupInMicChangeEvent() {
+	group_ev.AddTxGroupInEventAsync(func(model *domain.Model, event interface{}) error {
+		e, ok := event.(*group_ev.TxGroupInEvent)
+		if !ok {
+			return bizerr.InvalidParameter
+		}
+		return MicAllRPush(model, e.GroupId, e.UserId, e.ExternalId)
+	})
+}
+
+func MicAllRPush(model *domain.Model, imGroupId string, userId uint64, externalId string) error {
+	model.Log.Infof("MicChangeRPush MicAllRPush begin groupUuid:%v, externalId:%v", imGroupId, externalId)
+
+	txGroupId, err := group_m.ToTxGroupId(model, imGroupId)
+	if err != nil {
+		model.Log.Errorf("ToImGroupId fail:%v", err)
+		return err
+	}
+	//
+	micContents, err := group_m.GetMicAllContent(model, imGroupId)
+	if err != nil {
+		model.Log.Errorf("MicChangeRPush MicAllRPush getMicAllContent  err:%+v, micContents:%v groupUuid:%v, externalId:%v", err, micContents, txGroupId, externalId)
+		return err
+	}
+	for _, micContent := range micContents {
+		//麦上是默认值，就不用推
+		if micContent.Forbid == false && micContent.User == nil && micContent.AgoraId == 0 && micContent.Lock == false && micContent.ExternalId == "" && micContent.MicForbid == false {
+			model.Log.Infof("MicChangeRPush MicAllRPush default micContent:%v, groupUuid:%v, externalId:%v, micContent:%+v", micContent, txGroupId, externalId, micContent)
+			continue
+		}
+		micContentStr, err := json.Marshal(micContent)
+		if err != nil {
+			continue
+		}
+		var micExternalId string
+		var micUserData *userProxy.MicUserData
+		if micContent.User != nil {
+			micExternalId = micContent.ExternalId
+			micUserData = &userProxy.MicUserData{
+				Id:                       micContent.User.Id,
+				ExternalId:               micContent.User.ExternalId,
+				Avatar:                   micContent.User.Avatar,
+				Nick:                     micContent.User.Nick,
+				Sex:                      uint32(micContent.User.Sex),
+				Code:                     micContent.User.Code,
+				IsVip:                    micContent.User.IsVip,
+				Noble:                    uint32(micContent.User.NobleLeave),
+				HeadwearPicUrl:           micContent.User.HeadwearPicUrl,
+				HeadwearEffectUrl:        micContent.User.HeadwearEffectUrl,
+				HeadwearReverseEffectUrl: micContent.User.HeadwearReverseEffectUrl,
+				SvipLevel:                uint32(micContent.User.SvipLevel),
+				MicEffect:                micContent.User.MicEffect,
+				HeadwearIcon:             micContent.User.HeadwearIcon,
+				Svip:                     nil,
+			}
+			if micContent.User.Svip.SvipLevel > 0 || len(micContent.User.Svip.Privileges) > 0 {
+				micUserData.Svip = &userProxy.Svip{
+					SvipLevel: uint64(micContent.User.SvipLevel),
+				}
+				for _, v := range micContent.User.Svip.Privileges {
+					micUserData.Svip.Privileges = append(micUserData.Svip.Privileges, &userProxy.SvipPrivilege{
+						Type:        int32(v.Type),
+						CanSwitch:   v.CanSwitch,
+						UserSwitch:  v.UserSwitch,
+						MysteryCode: v.MysteryCode,
+					})
+				}
+			}
+		}
+		seqId := uuid.NewV4().String()
+
+		if err := rpc.SendSocketMicChange(seqId, userId, micExternalId, txGroupId, uint32(micContent.I), micContent.Lock, micContent.Forbid, micContent.MicForbid, micContent.AgoraId, micUserData); err != nil {
+			model.Log.Errorf("MicAllRPush err:%+v, groupUuid:%v, externalId:%v, micContent:%+v", err, txGroupId, externalId, string(micContentStr))
+		} else {
+			model.Log.Infof("MicChangeRPush MicAllRPush Marshal micContent, groupUuid:%v, externalId:%v, micContent:%+v", txGroupId, externalId, string(micContentStr))
+		}
+	}
+	model.Log.Infof("MicAllRPush end groupUuid:%v, externalId:%v", txGroupId, externalId)
+	return nil
+}

go.mod

@@ -93,6 +93,7 @@ require (
 	github.com/tencentyun/tls-sig-api-v2-golang v1.0.0 // indirect
 	github.com/ugorji/go/codec v1.1.7 // indirect
 	golang.org/x/net v0.0.0-20210428140749-89ef3d95e781 // indirect
+	golang.org/x/sync v0.3.0 // indirect
 	golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 // indirect
 	golang.org/x/text v0.3.6 // indirect
 	golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e // indirect

go.sum

@@ -401,6 +401,8 @@ golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.3.0 h1:ftCYgMx6zT/asHUrPw8BLLscYtGznsLAnjq5RH9P66E=
+golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=

route/group_r/group_mic.go

@@ -550,12 +550,13 @@ func GroupMicIn(c *gin.Context) (*mycontext.MyContext, error) {
 		return myContext, err
 	}
 	if micUser != nil {
-		if iStr == "" {
+		// 同房间且上麦序列为空
+		if iStr == "" && micUser.GroupUuid == groupUuid {
 			// 非切麦操作
 			return myContext, bizerr.GroupMicUserHasIn
 		}
 		// 切换麦位,先下麦
-		if err := group_mic_s.NewGroupPowerService(myContext).GroupMicLeave(groupUuid, micUser.I, userId, externalId); err != nil {
+		if err := group_mic_s.NewGroupPowerService(myContext).GroupMicLeave(micUser.GroupUuid, micUser.I, userId, externalId); err != nil {
 			return myContext, err
 		}
 	}