Restructure callback and schedule interface

Co-authored-by: Matthias Linhuber <[email protected]>

HadesScheduler/k8s.go

@@ -0,0 +1,13 @@
+package main
+
+import "github.com/Mtze/HadesCI/shared/payload"
+
+type JobScheduler interface {
+	ScheduleJob(job payload.BuildJob) error
+}
+
+type K8sScheduler struct{}
+
+func (k *K8sScheduler) ScheduleJob(job payload.BuildJob) error {
+	return nil
+}

HadesScheduler/main.go

@@ -11,7 +11,6 @@ import (
 	"github.com/Mtze/HadesCI/shared/queue"
 	"github.com/Mtze/HadesCI/shared/utils"
 
-	amqp "github.com/rabbitmq/amqp091-go"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -63,12 +62,8 @@ func main() {
 
 	var forever chan struct{}
 
-	f := func(ch <-chan amqp.Delivery) {
-		for d := range ch {
-			log.Printf("Received a message: %s", d.Body)
-		}
-	}
-	BuildQueue.Dequeue(f)
+	scheduler := K8sScheduler{}
+	BuildQueue.Dequeue(scheduler.ScheduleJob)
 
 	log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
 	<-forever

docker-compose.yml

@@ -36,6 +36,9 @@ services:
       - "5672:5672"
     networks:
       - hades
+    environment:
+      - RABBITMQ_DEFAULT_USER
+      - RABBITMQ_DEFAULT_PASS
     healthcheck:
       test: ["CMD", "rabbitmq-diagnostics", "ping"]
       interval: 30s

shared/queue/queue.go

@@ -79,7 +79,7 @@ func (q *Queue) Enqueue(ctx context.Context, msg payload.BuildJob) error {
 	return nil
 }
 
-func (q *Queue) Dequeue(callback func(<-chan amqp.Delivery)) error {
+func (q *Queue) Dequeue(callback func(job payload.BuildJob) error) error {
 	msgs, err := q.channel.Consume(
 		q.queue.Name, // queue
 		"",           // consumer
@@ -94,6 +94,14 @@ func (q *Queue) Dequeue(callback func(<-chan amqp.Delivery)) error {
 		return err
 	}
 
-	go callback(msgs)
+	for d := range msgs {
+		var buildJob payload.BuildJob
+		err := json.Unmarshal(d.Body, &buildJob)
+		if err != nil {
+			log.WithError(err).Warn("error unmarshalling message %s", d.Body)
+		}
+
+		go callback(buildJob)
+	}
 	return nil
 }