run processor

cmd/main.go

@@ -17,12 +17,9 @@ func main() {
 	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer cancel()
 
-	appl := app.NewApp(app.Deps{
+	appl := app.NewApp(ctx, c)
-		Config:  c,
-		RootCtx: ctx,
-	})
 
-	if err := appl.Run(); err != nil {
+	if err := appl.Run(ctx); err != nil {
 		log.WithError(err).Fatal("Application run failed")
 	}
 }

internal/app/handler.go

@@ -5,75 +5,82 @@ import (
 	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
-	rootCtx   context.Context
 	taskAgent   ta.TaskAgent
+	processor   processor.Processor
+	sendChanLen uint
 }
 
-type Deps struct {
+func NewApp(ctx context.Context, cfg config.Config) *App {
-	Config  config.Config
-	RootCtx context.Context
-}
-
-func NewApp(deps Deps) *App {
-	cfg := deps.Config
-
 	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,
+		},
+		ChanLen: cfg.App.ProcChanLen,
+	})
+
 	return &App{
 		config:      cfg,
-		rootCtx:   deps.RootCtx,
 		taskAgent:   taskAgent,
+		processor:   proc,
+		sendChanLen: cfg.App.ProcChanLen,
 	}
 }
 
-func (app *App) Run() error {
+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() {
-		if err := app.processTasks(); err != nil {
+		if err := app.processor.ProcessTasks(ctx); err != nil {
 			errChan <- err
 		}
 
 		for range mainLoop.C {
-			if err := app.processTasks(); err != nil {
+			if err := app.processor.ProcessTasks(ctx); err != nil {
 				errChan <- err
 			}
 		}
 	}()
 
+	if err := app.processor.SendResults(ctx, app.sendChanLen); err != nil {
+		errChan <- err
+	}
+
 	select {
-	case <-app.rootCtx.Done():
+	case <-ctx.Done():
-		return app.Shutdown()
+		return app.Shutdown(ctx)
 
 	case err := <-errChan:
 		return err
 	}
 }
 
-func (app *App) Shutdown() error {
+func (app *App) Shutdown(ctx context.Context) error {
 	log.Info("App shutting down")
-	ctx, cancel := context.WithTimeout(app.rootCtx, time.Second*10)
+	ctx, cancel := context.WithTimeout(ctx, time.Second*10)
 	defer cancel()
 
 	_ = ctx
 	return nil
 }
-
-func (app *App) processTasks() error {
-	log.Info("Processing tasks")
-	return nil
-}