app/routers/take.go

@@ -1,9 +1,11 @@
 package routers
 
 import (
+	"fmt"
 	"net/http"
 	tasks "queues-go/app/storage/mongo/collections"
-	"sync"
+	"queues-go/app/storage/redis"
+	"time"
 )
 
 type TaskResponse struct {
@@ -16,16 +18,17 @@ type TakeResponse struct {
 	Task *TaskResponse `json:"task"`
 }
 
-var MutexMap map[string]*sync.Mutex
-
 func Take(r *http.Request) (interface{}, int) {
 	queue := r.Header.Get("queue")
-	mutex, ok := MutexMap[queue]
-	if !ok {
-		mutex = &sync.Mutex{}
-		MutexMap[queue] = mutex
+	mutex := redis.Sync.NewMutex(fmt.Sprintf("lock_queues_%s", queue))
+	for {
+		err := mutex.Lock()
+		if err != nil {
+			time.Sleep(time.Millisecond * 5)
+		} else {
+			break
+		}
 	}
-	mutex.Lock()
 	task, err := tasks.Take(queue)
 	mutex.Unlock()
 	if err != nil {

main.go

@@ -6,7 +6,7 @@ import (
 	"net/http"
 	"queues-go/app/routers"
 	client "queues-go/app/storage/mongo"
-	"sync"
+	"queues-go/app/storage/redis"
 	"time"
 )
 
@@ -31,7 +31,7 @@ func handlerWrapper(f func(*http.Request) (interface{}, int)) func(http.Response
 
 func main() {
 	client.Connect()
-	routers.MutexMap = make(map[string]*sync.Mutex)
+	redis.Connect()
 	http.HandleFunc("/api/v1/take", handlerWrapper(routers.Take))
 	http.HandleFunc("/api/v1/finish", handlerWrapper(routers.Finish))
 	http.HandleFunc("/api/v1/put", handlerWrapper(routers.Put))