75 lines
1.8 KiB
Go
75 lines
1.8 KiB
Go
package queues
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"hotgo/internal/consts"
|
|
"hotgo/internal/library/imkafka"
|
|
"hotgo/internal/library/queue"
|
|
|
|
"github.com/gogf/gf/v2/frame/g"
|
|
"github.com/gogf/gf/v2/os/gtime"
|
|
)
|
|
|
|
func init() {
|
|
queue.RegisterConsumer(ImDeadLetter)
|
|
}
|
|
|
|
var ImDeadLetter = &qImDeadLetter{}
|
|
|
|
type qImDeadLetter struct{}
|
|
|
|
func (q *qImDeadLetter) GetTopic() string {
|
|
return consts.QueueImDeadLetterTopic
|
|
}
|
|
|
|
func (q *qImDeadLetter) Handle(ctx context.Context, mqMsg queue.MqMsg) (err error) {
|
|
var envelope imkafka.Envelope
|
|
if err = json.Unmarshal(mqMsg.Body, &envelope); err != nil {
|
|
return err
|
|
}
|
|
payload := map[string]any{}
|
|
if len(envelope.Payload) > 0 {
|
|
_ = json.Unmarshal(envelope.Payload, &payload)
|
|
}
|
|
_, err = g.DB().Model("im_kafka_dead_letter_events").Safe().Ctx(ctx).Data(g.Map{
|
|
"event_id": envelope.EventID,
|
|
"parent_event_id": envelope.ParentEventID,
|
|
"tenant_code": envelope.TenantCode,
|
|
"event_type": envelope.EventType,
|
|
"biz_type": envelope.BizType,
|
|
"biz_id": envelope.BizID,
|
|
"source_service": envelope.SourceService,
|
|
"target_service": envelope.TargetService,
|
|
"dead_id": stringValue(payload["dead_id"]),
|
|
"reason": firstString(payload["reason"], payload["last_reason"]),
|
|
"message": firstString(payload["message"], payload["last_error"]),
|
|
"payload": string(envelope.Payload),
|
|
"envelope": string(mqMsg.Body),
|
|
"received_at": gtime.Now(),
|
|
}).InsertIgnore()
|
|
return err
|
|
}
|
|
|
|
func stringValue(v any) string {
|
|
if v == nil {
|
|
return ""
|
|
}
|
|
switch value := v.(type) {
|
|
case string:
|
|
return value
|
|
default:
|
|
b, _ := json.Marshal(value)
|
|
return string(b)
|
|
}
|
|
}
|
|
|
|
func firstString(values ...any) string {
|
|
for _, v := range values {
|
|
if text := stringValue(v); text != "" && text != "null" {
|
|
return text
|
|
}
|
|
}
|
|
return ""
|
|
}
|