feature/bodySize

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", 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/service/event_s/event_init.go

@@ -35,6 +35,7 @@ func EventInit() {
 	GroupTheme()
 	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 (
+	"git.hilo.cn/hilo-common/domain"
+	"git.hilo.cn/hilo-common/rpc"
+	"hilo-group/domain/event/group_ev"
+	"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 rpc.SendSocketMicChange(e.UserId) // todo
+	})
+}

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=