Michael McGuinness
0ebb8a21a1
Clean Up Testing and Linting * somescriptcleanup * mostgolangci * deplatest * imageversions * usingscratch * movedqueuefrtesting * finishedunittesting * linting * taskfilecontext
52 lines
910 B
Go
52 lines
910 B
Go
package queue
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log"
|
|
)
|
|
|
|
type PollConfig struct {
|
|
Config *Config
|
|
Controller Controller
|
|
}
|
|
|
|
func PollMessages(ctx context.Context, queueConfig *PollConfig) {
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
default:
|
|
err := PollMessage(ctx, queueConfig)
|
|
if err != nil {
|
|
log.Print(err)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func PollMessage(ctx context.Context, queueConfig *PollConfig) error {
|
|
result, err := Receive(ctx, queueConfig.Config, []string{
|
|
"type",
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("message fetch fail: %v", err)
|
|
}
|
|
|
|
for _, message := range result.Messages {
|
|
go func() {
|
|
err := queueConfig.Controller.Process(ctx, queueConfig.Config, &message)
|
|
if err != nil {
|
|
log.Printf("message process fail: %v", err)
|
|
}
|
|
|
|
err = Delete(ctx, queueConfig.Config, &message)
|
|
if err != nil {
|
|
log.Printf("message delete fail: %v", err)
|
|
}
|
|
}()
|
|
}
|
|
|
|
return nil
|
|
}
|