You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
70 lines
1.5 KiB
Go
70 lines
1.5 KiB
Go
package consumer
|
|
|
|
import (
|
|
"context"
|
|
"joybuywork/internal/app"
|
|
"joybuywork/internal/consts"
|
|
"joybuywork/internal/plugins/jsonm"
|
|
"joybuywork/internal/plugins/printer"
|
|
"joybuywork/internal/utils"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/Shopify/sarama"
|
|
"github.com/rs/xid"
|
|
)
|
|
|
|
type Do func(ctx *Ctx) error
|
|
|
|
type Ctx struct {
|
|
ctx context.Context
|
|
name string
|
|
message *sarama.ConsumerMessage
|
|
do Do
|
|
}
|
|
|
|
func NewCtx(name string, message *sarama.ConsumerMessage, do Do) *Ctx {
|
|
ctx := context.WithValue(context.Background(), consts.CtxMsgId, xid.New().String())
|
|
for _, header := range message.Headers {
|
|
ctx = context.WithValue(ctx, utils.Bs(header.Key), utils.Bs(header.Value))
|
|
}
|
|
|
|
return &Ctx{
|
|
ctx: ctx,
|
|
name: name,
|
|
message: message,
|
|
do: do,
|
|
}
|
|
}
|
|
|
|
func (s *Ctx) Done() {
|
|
if s == nil {
|
|
return
|
|
}
|
|
|
|
req := utils.Bs(s.message.Value)
|
|
defer app.TryCatch(s.ctx, fmt.Sprintf("执行任务[%s]失败，数据: %s", s.name, req))
|
|
|
|
var err error
|
|
tn := time.Now()
|
|
sl := printer.NewSlowLog(s.ctx, printer.SetSlow(printer.TenNormalSlow))
|
|
app.Collect(s.name)
|
|
sl.Start(fmt.Sprintf("运行[%s], 执行间隔：%12v, 数据：%s", s.name, time.Since(s.message.Timestamp), req))
|
|
defer func() {
|
|
sl.End(fmt.Sprintf("执行时长：%12v, 运行错误：%v", time.Since(tn), err), err, s.name, req, "")
|
|
}()
|
|
|
|
err = s.do(&Ctx{
|
|
ctx: s.ctx,
|
|
message: s.message,
|
|
})
|
|
}
|
|
|
|
func (s *Ctx) Parse(data any) error {
|
|
return jsonm.Unmarshal(s.message.Value, data)
|
|
}
|
|
|
|
func (s *Ctx) Context() context.Context {
|
|
return s.ctx
|
|
}
|