Initial commit

internal/pkg/onexqueue/onexqueue.go

@@ -0,0 +1,71 @@
+package onexqueue
+
+import (
+	"github.com/streadway/amqp"
+	"log"
+)
+
+type DisplayAreaParam struct {
+	Width  int
+	Height int
+}
+
+type MessageParams struct {
+	Player      string
+	VenueId     string
+	DisplayTime int64
+	DisplayArea DisplayAreaParam
+}
+
+type Message struct {
+	Vendor string
+	Params MessageParams
+}
+
+func Connect(queueUrl string) (*amqp.Connection, *amqp.Channel) {
+	conn, err := amqp.Dial(queueUrl)
+	if err != nil {
+		log.Fatalf("Failed to connect to RabbitMQ: %s", err)
+	}
+
+	ch, err := conn.Channel()
+	if err != nil {
+		log.Fatalf("Failed to open a channel: %s", err)
+	}
+
+	return conn, ch
+}
+
+func CreateQueueIfNecessary(channel *amqp.Channel, queueName string) amqp.Queue {
+	queue, err := channel.QueueDeclare(
+		queueName, // name
+		false,     // durable
+		false,     // delete when unused
+		false,     // exclusive
+		false,     // no-wait
+		nil,       // arguments
+	)
+	if err != nil {
+		log.Fatalf("Failed to declare a queue: %s", err)
+	}
+	return queue
+}
+
+func Consume(channel *amqp.Channel, queueName string) <-chan amqp.Delivery {
+	queue := CreateQueueIfNecessary(channel, queueName)
+
+	messages, err := channel.Consume(
+		queue.Name, // queue
+		"",         // consumer
+		true,       // auto-ack
+		false,      // exclusive
+		false,      // no-local
+		false,      // no-wait
+		nil,        // args
+	)
+	if err != nil {
+		log.Fatalf("Failed to register a consumer: %s", err)
+	}
+
+	return messages
+}