MiFaZhan
110 lines
3.4 KiB
Go
110 lines
3.4 KiB
Go
package rolesync
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"github.com/gogf/gf/v2/errors/gerror"
|
|
"github.com/gogf/gf/v2/frame/g"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/consts"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/library/rabbitmq"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
func Start(ctx context.Context) (func(), error) {
|
|
if !g.Cfg().MustGet(ctx, "roleSync.enabled", false).Bool() {
|
|
return func() {}, nil
|
|
}
|
|
if !rabbitmq.Enabled(ctx) {
|
|
return nil, gerror.New("权限同步 Worker 必须启用 RabbitMQ 并配置连接")
|
|
}
|
|
workerCtx, cancel := context.WithCancel(ctx)
|
|
go superviseConsumer(workerCtx)
|
|
g.Log().Info(ctx, "role sync Worker started")
|
|
return cancel, nil
|
|
}
|
|
|
|
func handleMessage(ctx context.Context, message model.AgentRoleSyncMessage) error {
|
|
if err := syncRoleMessage(ctx, &message); err != nil {
|
|
g.Log().Errorf(ctx, "role sync failed, taskId:%s, roleId:%d, attempt:%d, err:%+v", message.TaskId, message.SourceRoleId, message.RetryCount, err)
|
|
message.RetryCount++
|
|
body, marshalErr := json.Marshal(message)
|
|
if marshalErr != nil {
|
|
return marshalErr
|
|
}
|
|
if message.RetryCount > 5 {
|
|
deadLetter, marshalErr := json.Marshal(model.AgentRoleSyncDeadLetterMessage{
|
|
Message: message, Error: err.Error(), FailedAt: time.Now().Format(time.RFC3339),
|
|
})
|
|
if marshalErr != nil {
|
|
return marshalErr
|
|
}
|
|
return publishRoleDeadLetter(ctx, deadLetter)
|
|
}
|
|
delay := time.Duration(1<<uint(message.RetryCount-1)) * 5 * time.Second
|
|
if publishErr := publishRoleRetry(ctx, delay, body); publishErr != nil {
|
|
return gerror.Wrap(publishErr, fmt.Sprintf("role sync retry publish failed: %v", err))
|
|
}
|
|
return nil
|
|
}
|
|
return nil
|
|
}
|
|
|
|
var consumeRoleSync = rabbitmq.ConsumeWithDone
|
|
|
|
var syncRoleMessage = SyncInheritedRole
|
|
|
|
var publishRoleRetry = func(ctx context.Context, delay time.Duration, body []byte) error {
|
|
return rabbitmq.PublishDelayedBytesWithHeaders(ctx, consts.QueueAgentRoleSync+".retry", consts.QueueAgentRoleSync, delay, body, nil)
|
|
}
|
|
|
|
var publishRoleDeadLetter = func(ctx context.Context, body []byte) error {
|
|
return rabbitmq.PublishBytesWithHeaders(ctx, consts.QueueAgentRoleSync+".dlq", body, nil)
|
|
}
|
|
|
|
func superviseConsumer(ctx context.Context) {
|
|
for ctx.Err() == nil {
|
|
done, err := consumeRoleSync(ctx, consts.QueueAgentRoleSync, "iotlcm-worker-agent-role-sync", func(ctx context.Context, body []byte) error {
|
|
var message model.AgentRoleSyncMessage
|
|
invalid := json.Unmarshal(body, &message)
|
|
if invalid != nil || message.SourceRoleId <= 0 || message.SourceDeptId <= 0 || strings.TrimSpace(message.TenantId) == "" || message.RetryCount < 0 {
|
|
return quarantineRoleSync(ctx, body)
|
|
}
|
|
if err := handleMessage(ctx, message); err != nil {
|
|
select {
|
|
case <-ctx.Done():
|
|
case <-time.After(time.Second):
|
|
}
|
|
return err
|
|
}
|
|
return nil
|
|
})
|
|
if err == nil {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-done:
|
|
}
|
|
}
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
g.Log().Warningf(ctx, "role sync consumer disconnected, restarting: %v", err)
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-time.After(3 * time.Second):
|
|
}
|
|
}
|
|
}
|
|
|
|
var quarantineRoleSync = func(ctx context.Context, body []byte) error {
|
|
if err := publishRoleDeadLetter(ctx, body); err != nil {
|
|
return gerror.Wrap(err, "保存无效权限消息到死信队列失败")
|
|
}
|
|
g.Log().Warning(ctx, "无效权限消息已隔离到死信队列")
|
|
return nil
|
|
}
|