MiFaZhan
44 lines
1.2 KiB
Go
44 lines
1.2 KiB
Go
package rabbitmq
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/gogf/gf/v2/frame/g"
|
|
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/consts"
|
|
mq "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/library/rabbitmq"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model"
|
|
)
|
|
|
|
func (c *Consumers) initWorkerDiagnostic(ctx context.Context) error {
|
|
if !mq.Enabled(ctx) {
|
|
g.Log().Info(ctx, "worker diagnostic consumer skipped, RabbitMQ disabled")
|
|
return nil
|
|
}
|
|
if err := mq.ConsumeJSON[model.WorkerDiagnosticMessage](
|
|
ctx,
|
|
consts.QueueWorkerDiagnostic,
|
|
"iotlcm-worker-diagnostic",
|
|
c.handleWorkerDiagnostic,
|
|
); err != nil {
|
|
return err
|
|
}
|
|
g.Log().Infof(ctx, "worker diagnostic consumer started, queue:%s", consts.QueueWorkerDiagnostic)
|
|
return nil
|
|
}
|
|
|
|
func (c *Consumers) handleWorkerDiagnostic(ctx context.Context, message model.WorkerDiagnosticMessage) error {
|
|
if message.TraceID == "" {
|
|
g.Log().Errorf(ctx, "worker diagnostic message dropped, invalid traceId, message:%+v", message)
|
|
return nil
|
|
}
|
|
g.Log().Infof(
|
|
ctx,
|
|
"worker diagnostic message consumed, traceId:%s, sentAt:%d, message:%s",
|
|
message.TraceID,
|
|
message.SentAt,
|
|
message.Message,
|
|
)
|
|
return nil
|
|
}
|