sqs worker

sqs.go

package main

import (
	"fmt"
	"os"
	"os/signal"
	"sync"
	"syscall"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/service/sqs"
)

func main() {
	sc := make(chan os.Signal, 1)
	signal.Notify(
		sc,
		syscall.SIGHUP,
		syscall.SIGINT,
		syscall.SIGTERM,
		syscall.SIGQUIT,
	)

	svc = sqs.New(
		session.New(
			&aws.Config{
				Region: aws.String("ap-northeast-01"),
			},
		),
	)

	mc := make(chan *sqs.Message)

	var wg sync.WaitGroup
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go worker(svc, &wg, mc)
	}

	if err := loop(sc, mc); err != nil {
		fmt.Println(err)
	}

	close(mc)

	wg.Wait()
}

func loop(sc <-chan os.Signal, mc chan<- *sqs.Message) error {
	in := &sqs.ReceiveMessageInput{
		QueueUrl:            aws.String("QueueUrl"),
		MaxNumberOfMessages: aws.Int64(int64(1)),
		WaitTimeSeconds:     aws.Int64(int64(1)),
	}
	for {
		select {
		case _ = <-sc:
			return nil
		default:
			res, err := svc.ReceiveMessage(in)
			if err != nil {
				return err
			}

			for _, m := range res.Messages {
				mc <- m
			}
		}
	}
}

func worker(c *sqs.SQS, wg *sync.WaitGroup, mc <-chan *sqs.Message) {
	defer wg.Done()

	for m := range mc {
		c.DeleteMessage(
			&sqs.DeleteMessageInput{
				QueueUrl:      aws.String("QueueUrl"),
				ReceiptHandle: aws.String(*m.ReceiptHandle),
			},
		)
	}
}