feat: 重构项目结构,添加命令处理程序并移除冗余文件

This commit is contained in:
Rogee
2025-09-12 19:43:06 +08:00
parent 4c8205212f
commit dd097e9aec
12 changed files with 6 additions and 61 deletions

View File

@@ -0,0 +1,24 @@
package queue
import (
"context"
"github.com/riverqueue/river"
"github.com/riverqueue/river/rivertype"
log "github.com/sirupsen/logrus"
)
type CustomErrorHandler struct{}
func (*CustomErrorHandler) HandleError(ctx context.Context, job *rivertype.JobRow, err error) *river.ErrorHandlerResult {
log.Infof("Job errored with: %s\n", err)
return nil
}
func (*CustomErrorHandler) HandlePanic(ctx context.Context, job *rivertype.JobRow, panicVal any, trace string) *river.ErrorHandlerResult {
log.Infof("Job panicked with: %v\n", panicVal)
log.Infof("Stack trace: %s\n", trace)
return &river.ErrorHandlerResult{
SetCancelled: true,
}
}

View File

@@ -0,0 +1,66 @@
package queue
import (
"context"
"go.ipao.vip/atom"
"go.ipao.vip/atom/container"
"go.ipao.vip/atom/contracts"
"{{.ModuleName}}/app/jobs"
"{{.ModuleName}}/providers/app"
"{{.ModuleName}}/providers/job"
"{{.ModuleName}}/providers/postgres"
log "github.com/sirupsen/logrus"
"github.com/spf13/cobra"
"go.uber.org/dig"
)
func defaultProviders() container.Providers {
return srv.Default(container.Providers{
postgres.DefaultProvider(),
job.DefaultProvider(),
}...)
}
func Command() atom.Option {
return atom.Command(
atom.Name("queue"),
atom.Short("start queue processor"),
atom.RunE(Serve),
atom.Providers(
defaultProviders().
With(
jobs.Provide,
),
),
)
}
type Service struct {
dig.In
App *app.Config
Job *job.Job
Initials []contracts.Initial `group:"initials"`
CronJobs []contracts.CronJob `group:"cron_jobs"`
}
func Serve(cmd *cobra.Command, args []string) error {
return container.Container.Invoke(func(ctx context.Context, svc Service) error {
log.SetFormatter(&log.JSONFormatter{})
if svc.App.IsDevMode() {
log.SetLevel(log.DebugLevel)
}
if err := svc.Job.Start(ctx); err != nil {
return err
}
defer svc.Job.Close()
<-ctx.Done()
return nil
})
}