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
}