send_gift.go 2.14 KB
Newer Older
hujiebin's avatar
hujiebin committed
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32
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"
	"hilo-group/domain/event/gift_ev"
	"hilo-group/domain/model/event_m"
	"runtime/debug"
)

// 每次处理500条
const BatchCount = 500

type GiftSendEventService struct {
	svc *domain.Service
}

func NewGiftSendEventService(myContext *mycontext.MyContext) *GiftSendEventService {
	svc := domain.CreateService(myContext)
	return &GiftSendEventService{svc}
}

//
func (s *GiftSendEventService) Consume() error {
	defer func() {
		if err := recover(); err != nil {
			s.svc.Log.Errorf("ExceptionHandle GiftSendEventService Consume SYSTEM ACTION PANIC: %v, stack: %v", err, string(debug.Stack()))
		}
	}()
	var model = domain.CreateModel(s.svc.CtxAndDb)
33 34 35 36 37 38 39
	events, offset, err := event_m.FetchEventGiftSend(model, BatchCount)
	if err != nil {
		return err
	}
	for k := range events {
		// 一个事件一个事务
		err = model.Transaction(func(model *domain.Model) error {
hujiebin's avatar
hujiebin committed
40 41 42 43 44 45 46 47 48 49 50 51
			cpEvent := &event_m.EventGiftSend{
				Entity: mysql.Entity{
					ID:          events[k].ID,
					CreatedTime: events[k].CreatedTime,
					UpdatedTime: events[k].UpdatedTime,
				},
				Proto:         events[k].Proto,
				Payload:       events[k].Payload,
				MarkHiloGroup: events[k].MarkHiloGroup,
			}
			if cpEvent.MarkHiloGroup == mysql.YES {
				model.Log.Warnf("already consume msg :%v", cpEvent)
52
				return nil
hujiebin's avatar
hujiebin committed
53 54 55 56
			}
			sendGiftEvent := new(gift_ev.SendGiftEvent)
			if err := json.Unmarshal(cpEvent.Payload, sendGiftEvent); err != nil {
				model.Log.Errorf("json msg fail,event:%v,err:%v", cpEvent, err)
57
				return nil
hujiebin's avatar
hujiebin committed
58 59 60 61 62 63 64 65 66 67 68 69
			}
			if err := gift_ev.PublishSendGiftEvent(model, sendGiftEvent); err != nil {
				model.Log.Errorf("PublishSendGiftEvent fail,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
			}
70 71 72 73 74 75 76 77 78
			return nil
		})
	}
	// 最后一次提交offset
	if len(events) > 0 {
		offset.MarkOffset = events[len(events)-1].ID
		return offset.Persistence()
	}
	return nil
hujiebin's avatar
hujiebin committed
79
}