declare queue with opts quick fix
All checks were successful
/ Make image (push) Successful in 44s

This commit is contained in:
nquidox 2026-04-27 00:18:39 +03:00
parent c8a384efcb
commit 9361319f0a
3 changed files with 66 additions and 23 deletions

View file

@ -3,7 +3,6 @@ package processor
import (
"context"
"encoding/json"
"fmt"
log "github.com/sirupsen/logrus"
rabbit "repo.nqws.ru/merch-tracker-v2/mt-rabbit"
"task-processor/internal/structs"
@ -62,8 +61,16 @@ func (s *service) SendResults(ctx context.Context, chanLen uint) error {
log.Debugf("%v Results sender start", pkgLogHeader)
runCtx, cancel := context.WithCancel(ctx)
qn := "tasks-results"
consumerClient, err := rabbit.NewClient(s.brokerAddr, qn, rabbit.WithLogging(s.rabbitLoggingEnabled))
taskResults := rabbit.QueueOpts{
QueueName: "tasks-results",
Durable: false,
AutoDelete: false,
Exclusive: false,
NoWait: false,
Args: nil,
}
consumerClient, err := rabbit.NewClient(s.brokerAddr, taskResults, rabbit.WithLogging(s.rabbitLoggingEnabled))
if err != nil {
cancel()
return err
@ -71,7 +78,7 @@ func (s *service) SendResults(ctx context.Context, chanLen uint) error {
resultsConsumer := rabbit.NewConsumer(consumerClient)
resultChan := resultsConsumer.Start(runCtx, chanLen)
log.Debugf("%v Results consumer started: %v", pkgLogHeader, qn)
log.Debugf("%v Results consumer started: %v", pkgLogHeader, taskResults.QueueName)
go func() {
defer cancel()
@ -145,26 +152,62 @@ func (s *service) convertResult(b []byte) *structs.Result {
// TODO refactor this later: get origins from merch api and remove ctx pass via deps
func (s *service) makeTaskPublishers(ctx context.Context, addr rabbit.Address, chanLen uint) {
origins := [...]string{
"surugaya",
"mandarake",
"amiami",
}
publishers := make(map[string]chan<- []byte)
for _, origin := range origins {
qn := fmt.Sprintf("task-publisher-%s", origin)
pubClient, err := rabbit.NewClient(addr, qn, rabbit.WithLogging(s.rabbitLoggingEnabled))
if err != nil {
log.WithError(err).Error("Failed to create publisher")
continue
}
publishers[origin] = rabbit.NewPublisher(pubClient).Start(ctx, chanLen)
log.Debugf("%v Publisher queue created: %v", pkgLogHeader, qn)
// surugaya
surugayaOpts := rabbit.QueueOpts{
QueueName: "task-publisher-surugaya",
Durable: false,
AutoDelete: false,
Exclusive: false,
NoWait: false,
Args: nil,
}
surugayaClient, err := rabbit.NewClient(addr, surugayaOpts, rabbit.WithLogging(s.rabbitLoggingEnabled))
if err != nil {
log.WithError(err).Error("Failed to create publisher")
}
publishers["surugaya"] = rabbit.NewPublisher(surugayaClient).Start(ctx, chanLen)
log.Debugf("%v Publisher queue created: %v", pkgLogHeader, surugayaOpts.QueueName)
//mandarake
mandarakeOpts := rabbit.QueueOpts{
QueueName: "task-publisher-mandarake",
Durable: false,
AutoDelete: false,
Exclusive: false,
NoWait: false,
Args: nil,
}
mandarakeClient, err := rabbit.NewClient(addr, mandarakeOpts, rabbit.WithLogging(s.rabbitLoggingEnabled))
if err != nil {
log.WithError(err).Error("Failed to create publisher")
}
publishers["mandarake"] = rabbit.NewPublisher(mandarakeClient).Start(ctx, chanLen)
log.Debugf("%v Publisher queue created: %v", pkgLogHeader, mandarakeOpts.QueueName)
//amiami
amiamiOpts := rabbit.QueueOpts{
QueueName: "task-publisher-amiami",
Durable: true,
AutoDelete: false,
Exclusive: false,
NoWait: false,
Args: nil,
}
amiamiClient, err := rabbit.NewClient(addr, amiamiOpts, rabbit.WithLogging(s.rabbitLoggingEnabled))
if err != nil {
log.WithError(err).Error("Failed to create publisher")
}
publishers["amiami"] = rabbit.NewPublisher(amiamiClient).Start(ctx, chanLen)
log.Debugf("%v Publisher queue created: %v", pkgLogHeader, amiamiOpts.QueueName)
s.taskPublishers = publishers
}