feat: implement worker pkg

.support/docker/docker-compose.dev.yml

@@ -1,7 +1,7 @@
 version: '3.7'
 
 services:
-  db:
+  postgres:
     image: postgres:latest
     restart: on-failure
     ports:

cmd/gilfoyle/worker.go

@@ -14,7 +14,8 @@ func init() {
 
 var workerCmd = &cobra.Command{
 	Use:     "worker",
-	Short:   "Launch background task worker",
+	Short:   "Launch a background task worker node",
+	Long:    "Multiple worker nodes compose a worker pool. We usually recommend to launch a minimum of 2 worker nodes to ensure fail over.",
 	Example: "gilfoyle worker",
 	Run: func(cmd *cobra.Command, args []string) {
 		logger := gilfoyle.Logger
@@ -34,6 +35,33 @@ var workerCmd = &cobra.Command{
 		}
 		defer w.Close()
 
-		w.Init()
+		forever := make(chan bool)
+
+		err = w.Init()
+		if err != nil {
+			logger.Fatal("Failed to initialize worker queues", zap.Error(err))
+		}
+
+		w.Consume()
+
+		ch, err := w.Client.Channel()
+		if err != nil {
+			logger.Fatal("Failed to create message queue channel", zap.Error(err))
+		}
+		defer ch.Close()
+
+		//time.Sleep(2 * time.Second)
+		//err = ch.Publish("", worker.VideoTranscodingQueue, false, false, amqp.Publishing{
+		//	DeliveryMode: amqp.Persistent,
+		//	ContentType:  "text/plain",
+		//	Body:         []byte("hello!!!"),
+		//})
+		//if err != nil {
+		//	logger.Error("Failed to publish a message", zap.Error(err))
+		//}
+
+		logger.Info("Worker is now ready to handle incoming jobs")
+
+		<-forever
 	},
 }

worker/consumers.go

@@ -0,0 +1,20 @@
+package worker
+
+import (
+	"github.com/google/uuid"
+	"github.com/streadway/amqp"
+	"go.uber.org/zap"
+	"time"
+)
+
+type VideoTranscodingParams struct {
+	MediaUUID uuid.UUID
+}
+
+func videoTranscodingQueueConsumer(w *Worker, msgs <-chan amqp.Delivery) {
+	for d := range msgs {
+		w.Logger.Info("Received a message", zap.ByteString("body", d.Body))
+		time.Sleep(3 * time.Second)
+		w.Logger.Info("Done")
+	}
+}

worker/worker.go

@@ -7,11 +7,51 @@ import (
 )
 
 const (
-//videoTranscodingQueue string = "videoTranscoding"
-//thumbnailGenerationQueue string = "thumbnailGeneration"
-//previewGenerationQueue string = "previewGeneration"
+	VideoTranscodingQueue    string = "VideoTranscoding"
+	ThumbnailGenerationQueue string = "ThumbnailGeneration"
+	PreviewGenerationQueue   string = "PreviewGeneration"
 )
 
+type Queue struct {
+	Name       string
+	Durable    bool
+	AutoDelete bool
+	Exclusive  bool
+	NoWait     bool
+	Args       amqp.Table
+	Handler    func(*Worker, <-chan amqp.Delivery)
+}
+
+var queues = []Queue{
+	{
+		Name:       VideoTranscodingQueue,
+		Durable:    false,
+		AutoDelete: false,
+		Exclusive:  false,
+		NoWait:     false,
+		Args:       nil,
+		Handler:    videoTranscodingQueueConsumer,
+	},
+	{
+		Name:       ThumbnailGenerationQueue,
+		Durable:    false,
+		AutoDelete: false,
+		Exclusive:  false,
+		NoWait:     false,
+		Args:       nil,
+		Handler:    func(*Worker, <-chan amqp.Delivery) {},
+	},
+	{
+		Name:       PreviewGenerationQueue,
+		Durable:    false,
+		AutoDelete: false,
+		Exclusive:  false,
+		NoWait:     false,
+		Args:       nil,
+		Handler:    func(*Worker, <-chan amqp.Delivery) {},
+	},
+}
+
 type Options struct {
 	Host     string
 	Port     int16
@@ -21,8 +61,9 @@ type Options struct {
 }
 
 type Worker struct {
-	client *amqp.Connection
-	logger *zap.Logger
+	Queues map[string]amqp.Queue
+	Logger *zap.Logger
+	Client *amqp.Connection
 }
 
 func New(opts Options) (*Worker, error) {
@@ -38,15 +79,68 @@ func New(opts Options) (*Worker, error) {
 	}
 
 	return &Worker{
-		client: conn,
-		logger: opts.Logger,
+		Queues: map[string]amqp.Queue{},
+		Client: conn,
+		Logger: opts.Logger,
 	}, nil
 }
 
-func (w *Worker) Init() {
-	w.logger.Info("test")
+func (w *Worker) Init() error {
+	ch, err := w.Client.Channel()
+	if err != nil {
+		return err
+	}
+
+	for _, q := range queues {
+		queue, err := ch.QueueDeclare(
+			q.Name,       // name
+			q.Durable,    // durable
+			q.AutoDelete, // delete when unused
+			q.Exclusive,  // exclusive
+			q.NoWait,     // no-wait
+			q.Args,       // arguments
+		)
+		if err != nil {
+			return err
+		}
+
+		w.Queues[q.Name] = queue
+	}
+
+	return nil
+}
+
+func (w *Worker) Consume() {
+	ch, err := w.Client.Channel()
+	if err != nil {
+		w.Logger.Fatal("Error creating channel", zap.Error(err))
+		return
+	}
+
+	for _, q := range queues {
+		msgs, err := ch.Consume(
+			q.Name, // queue
+			"",     // consumer
+			true,   // auto-ack
+			false,  // exclusive
+			false,  // no-local
+			false,  // no-wait
+			map[string]interface{}{},
+		)
+		if err != nil {
+			w.Logger.Sugar().Fatalf("Error consuming %s queue: %e", q.Name, err)
+			return
+		}
+
+		go q.Handler(w, msgs)
+	}
 }
 
 func (w *Worker) Close() error {
-	return w.client.Close()
+	err := w.Client.Close()
+	if err != nil {
+		return err
+	}
+
+	return nil
 }