61 lines
2.2 KiB
Go
61 lines
2.2 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"hyapp/pkg/outboxpartition"
|
|
"hyapp/pkg/outboxreplay"
|
|
"hyapp/pkg/rocketmqx"
|
|
serviceapp "hyapp/pkg/servicekit/app"
|
|
servicemq "hyapp/pkg/servicekit/mq"
|
|
"hyapp/services/user-service/internal/config"
|
|
mysqlstorage "hyapp/services/user-service/internal/storage/mysql"
|
|
)
|
|
|
|
// ReplayUserOutboxOnce processes a process-wide bounded App batch without
|
|
// starting user-service listeners or its unrelated consumers. Claiming one row
|
|
// per iteration keeps the existing 30-second row lease attached only to the
|
|
// message currently being published.
|
|
func ReplayUserOutboxOnce(ctx context.Context, cfg config.Config, scope outboxreplay.Scope) (outboxreplay.Result, error) {
|
|
validatedScope, err := outboxreplay.NewScope(scope.AppCode, scope.Limit, 500)
|
|
if err != nil {
|
|
return outboxreplay.Result{}, err
|
|
}
|
|
if !cfg.RocketMQ.UserOutbox.Enabled {
|
|
return outboxreplay.Result{}, errors.New("user RocketMQ outbox is not enabled")
|
|
}
|
|
repository, err := mysqlstorage.Open(ctx, cfg.MySQLDSN)
|
|
if err != nil {
|
|
return outboxreplay.Result{}, err
|
|
}
|
|
defer func() { _ = repository.Close() }()
|
|
producer, err := rocketmqx.NewProducer(userOutboxProducerConfig(cfg.RocketMQ))
|
|
if err != nil {
|
|
return outboxreplay.Result{}, err
|
|
}
|
|
if err := servicemq.StartProducers([]*rocketmqx.Producer{producer}); err != nil {
|
|
return outboxreplay.Result{}, err
|
|
}
|
|
defer servicemq.ShutdownProducers([]*rocketmqx.Producer{producer})
|
|
|
|
replayConfig := cfg
|
|
replayConfig.OutboxWorker.BatchSize = 1
|
|
replayConfig.OutboxWorker.AppCodes = []string{validatedScope.AppCode}
|
|
workers := serviceapp.NewBackground(ctx)
|
|
defer workers.StopAndWait()
|
|
replayApp := &App{
|
|
mysqlRepo: repository,
|
|
userOutboxProducer: producer,
|
|
userOutboxPartitions: outboxpartition.New(replayConfig.OutboxWorker.AppCodes, outboxpartition.DefaultDiscoveryRefreshInterval),
|
|
cfg: replayConfig,
|
|
workers: workers,
|
|
}
|
|
workerID := fmt.Sprintf("ops-user-outbox-%s-%d", validatedScope.AppCode, time.Now().UTC().UnixMilli())
|
|
return outboxreplay.Run(ctx, validatedScope, func(context.Context) (int, error) {
|
|
return replayApp.processUserOutboxBatch(workerID)
|
|
})
|
|
}
|