89 lines
1.8 KiB
Go
89 lines
1.8 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
log "github.com/sirupsen/logrus"
|
|
"net"
|
|
"task-processor/config"
|
|
"task-processor/internal/processor"
|
|
ta "task-processor/internal/taskAgent"
|
|
"time"
|
|
)
|
|
|
|
type App struct {
|
|
config config.Config
|
|
taskAgent ta.TaskAgent
|
|
processor processor.Processor
|
|
sendChanLen uint
|
|
}
|
|
|
|
func NewApp(ctx context.Context, cfg config.Config) *App {
|
|
taskAgent := ta.NewHandler(ta.Deps{
|
|
Addr: net.JoinHostPort(cfg.TasksSource.Host, cfg.TasksSource.Port),
|
|
Timeout: cfg.TasksSource.Timeout,
|
|
})
|
|
|
|
proc := processor.NewHandler(processor.Deps{
|
|
Ctx: ctx,
|
|
TA: taskAgent,
|
|
Addr: processor.Addr{
|
|
Host: cfg.Rabbit.Host,
|
|
Port: cfg.Rabbit.Port,
|
|
User: cfg.Rabbit.User,
|
|
Pass: cfg.Rabbit.Pass,
|
|
Vhost: cfg.Rabbit.Vhost,
|
|
},
|
|
LoggingEnabled: cfg.Rabbit.LoggingEnabled,
|
|
ChanLen: cfg.App.ProcChanLen,
|
|
})
|
|
|
|
return &App{
|
|
config: cfg,
|
|
taskAgent: taskAgent,
|
|
processor: proc,
|
|
sendChanLen: cfg.App.ProcChanLen,
|
|
}
|
|
}
|
|
|
|
func (app *App) Run(ctx context.Context) error {
|
|
log.Info("App started")
|
|
errChan := make(chan error, 3)
|
|
|
|
mainLoop := time.NewTicker(app.config.App.CheckPeriod)
|
|
defer mainLoop.Stop()
|
|
|
|
go func() {
|
|
log.Info("Process tasks after start")
|
|
if err := app.processor.ProcessTasks(ctx); err != nil {
|
|
errChan <- err
|
|
}
|
|
|
|
for range mainLoop.C {
|
|
log.WithField("period", app.config.App.CheckPeriod).Info("Repeated process tasks")
|
|
if err := app.processor.ProcessTasks(ctx); err != nil {
|
|
errChan <- err
|
|
}
|
|
}
|
|
}()
|
|
|
|
if err := app.processor.SendResults(ctx, app.sendChanLen); err != nil {
|
|
errChan <- err
|
|
}
|
|
|
|
select {
|
|
case <-ctx.Done():
|
|
return app.Shutdown(ctx)
|
|
|
|
case err := <-errChan:
|
|
return err
|
|
}
|
|
}
|
|
|
|
func (app *App) Shutdown(ctx context.Context) error {
|
|
log.Info("App shutting down")
|
|
ctx, cancel := context.WithTimeout(ctx, time.Second*10)
|
|
defer cancel()
|
|
|
|
_ = ctx
|
|
return nil
|
|
}
|