Реализован функционал прослушивания RabbitMQ сообщений и соответствующие handlers для событий

docker-compose.base.yml

@@ -2,9 +2,16 @@ version: '2'
 services:
     app:
         ports:
-            - "80:80"
+            - "81:80"
 
     redis:
         image: redis:3.0
         volumes:
             - ./data/redis:/data
+
+    rabbitmq:
+        image: rabbitmq:3.6
+        environment:
+            RABBITMQ_DEFAULT_USER: "ely-skinsystem-app"
+            RABBITMQ_DEFAULT_PASS: "ely-skinsystem-app-password"
+            RABBITMQ_DEFAULT_VHOST: "/ely"

docker-compose.dev.yml

@@ -10,8 +10,14 @@ services:
         command: ["go", "run", "minecraft-skinsystem.go"]
         links:
             - redis
+            - rabbitmq
 
     redis:
         extends:
             file: docker-compose.base.yml
             service: redis
+
+    rabbitmq:
+        extends:
+            file: docker-compose.base.yml
+            service: rabbitmq

docker-compose.prod.yml

@@ -7,6 +7,7 @@ services:
         image: erickskrauch/ely-by-skinsystem:master
         links:
             - redis
+            - rabbitmq
         restart: always
 
     redis:
@@ -14,3 +15,9 @@ services:
             file: docker-compose.base.yml
             service: redis
         restart: always
+
+    rabbitmq:
+        extends:
+            file: docker-compose.base.yml
+            service: rabbitmq
+        restart: always

lib/services/services.go

@@ -2,9 +2,9 @@ package services
 
 import (
 	"github.com/mediocregopher/radix.v2/pool"
-	"github.com/gorilla/mux"
+	"github.com/streadway/amqp"
 )
 
 var RedisPool *pool.Pool
 
-var Router *mux.Router
+var RabbitMQChannel *amqp.Channel

lib/worker/handlers.go

@@ -0,0 +1,52 @@
+package worker
+
+import (
+	"elyby/minecraft-skinsystem/lib/data"
+	"log"
+)
+
+func handleChangeUsername(model usernameChanged) (bool) {
+	if (model.OldUsername == "") {
+		record := data.SkinItem{
+			UserId: model.AccountId,
+			Username: model.NewUsername,
+		}
+
+		record.Save()
+
+		return true
+	}
+
+	record, err := data.FindByUsername(model.OldUsername)
+	if (err != nil) {
+		log.Println("Exit by not found record")
+		// TODO: я не уверен, что это валидное поведение
+		// Суть в том, что здесь может возникнуть ошибка в том случае, если записи в базе нету
+		// а значит его нужно, как минимум, зарегистрировать
+		return true
+	}
+
+	record.Username = model.NewUsername
+	record.Save()
+
+	log.Println("all saved!")
+
+	return true
+}
+
+func handleSkinChanged(model skinChanged) (bool) {
+	record, err := data.FindById(model.AccountId)
+	if (err != nil) {
+		return true
+	}
+
+	record.SkinId = model.SkinId
+	record.Hash   = model.Hash
+	record.Is1_8  = model.Is1_8
+	record.IsSlim = model.IsSlim
+	record.Url    = model.Url
+
+	record.Save()
+
+	return true
+}

lib/worker/models.go

@@ -0,0 +1,17 @@
+package worker
+
+type usernameChanged struct {
+	AccountId   int    `json:"accountId"`
+	OldUsername string `json:"oldUsername"`
+	NewUsername string `json:"newUsername"`
+}
+
+type skinChanged struct {
+	AccountId int    `json:"userId"`
+	SkinId    int    `json:"skinId"`
+	OldSkinId int    `json:"oldSkinId"`
+	Hash      string `json:"hash"`
+	Is1_8     bool   `json:"is1_8"`
+	IsSlim    bool   `json:"isSlim"`
+	Url       string `json:"url"`
+}

lib/worker/worker.go

@@ -0,0 +1,91 @@
+package worker
+
+import (
+	"log"
+
+	"encoding/json"
+
+	"elyby/minecraft-skinsystem/lib/services"
+)
+
+const exchangeName string = "events"
+const queueName string = "skinsystem-accounts-events"
+
+func Listen() {
+	var err error
+	ch := services.RabbitMQChannel
+
+	err = ch.ExchangeDeclare(
+		exchangeName, // name
+		"topic",      // type
+		true,         // durable
+		false,        // auto-deleted
+		false,        // internal
+		false,        // no-wait
+		nil,          // arguments
+	)
+	failOnError(err, "Failed to declare an exchange")
+
+	_, err = ch.QueueDeclare(
+		queueName, // name
+		true,      // durable
+		false,     // delete when usused
+		false,     // exclusive
+		false,     // no-wait
+		nil,       // arguments
+	)
+	failOnError(err, "Failed to declare a queue")
+
+	err = ch.QueueBind(
+		queueName,    // queue name
+		"accounts.#", // routing key
+		exchangeName, // exchange
+		false,
+		nil,
+	)
+	failOnError(err, "Failed to bind a queue")
+
+	msgs, err := ch.Consume(
+		queueName, // queue
+		"",        // consumer
+		false,     // auto-ack
+		false,     // exclusive
+		false,     // no-local
+		false,     // no-wait
+		nil,       // args
+	)
+	failOnError(err, "Failed to register a consumer")
+
+	forever := make(chan bool)
+
+	go func() {
+		for d := range msgs {
+			log.Println("Incoming message with routing key " + d.RoutingKey)
+			var result bool = true;
+			switch d.RoutingKey {
+			case "accounts.username-changed":
+				var model usernameChanged
+				json.Unmarshal(d.Body, &model)
+				result = handleChangeUsername(model)
+			case "accounts.skin-changed":
+				var model skinChanged
+				json.Unmarshal(d.Body, &model)
+				result = handleSkinChanged(model)
+			}
+
+			if (result) {
+				d.Ack(false)
+			} else {
+				d.Reject(true)
+			}
+		}
+	}()
+
+	<-forever
+}
+
+func failOnError(err error, msg string) {
+	if err != nil {
+		log.Fatalf("%s: %s", msg, err)
+	}
+}

minecraft-skinsystem.go

@@ -12,9 +12,13 @@ import (
 	"elyby/minecraft-skinsystem/lib/routes"
 	"elyby/minecraft-skinsystem/lib/services"
 	//"github.com/mediocregopher/radix.v2/redis"
+
+	"github.com/streadway/amqp"
+	"elyby/minecraft-skinsystem/lib/worker"
 )
 
 const redisString string = "redis:6379"
+const rabbitmqString string = "amqp://ely-skinsystem-app:ely-skinsystem-app-password@rabbitmq:5672/%2fely"
 
 func main() {
 	log.Println("Starting...")
@@ -28,6 +32,19 @@ func main() {
 	}
 	log.Println("Connected to redis")
 
+	log.Println("Connecting to rabbitmq")
+	// TODO: rabbitmq становится доступен не сразу. Нужно дождаться, пока он станет доступен, периодически повторяя запросы
+	rabbitConnection, rabbitmqErr := amqp.Dial(rabbitmqString)
+	if rabbitmqErr != nil {
+		log.Fatalf("%s", rabbitmqErr)
+	}
+	log.Println("Connected to rabbitmq. Trying to open a channel")
+	rabbitChannel, rabbitmqErr := rabbitConnection.Channel()
+	if rabbitmqErr != nil {
+		log.Fatalf("%s", rabbitmqErr)
+	}
+	log.Println("Connected to rabbitmq channel")
+
 	router := mux.NewRouter().StrictSlash(true)
 	router.HandleFunc("/skins/{username}", routes.Skin).Methods("GET").Name("skins")
 	router.HandleFunc("/cloaks/{username}", routes.Cape).Methods("GET").Name("cloaks")
@@ -43,7 +60,7 @@ func main() {
 	apiRouter.HandleFunc("/user/{username}/skin", routes.SetSkin).Methods("POST")
 
 	services.RedisPool = redisPool
-	services.Router = router
+	services.RabbitMQChannel = rabbitChannel
 
 	/*go func() {
 		for {
@@ -63,6 +80,8 @@ func main() {
 		}
 	}()*/
 
+	go worker.Listen()
+
 	log.Println("Started");
 	log.Fatal(http.ListenAndServe(":80", router))
 }