Merge pull request 'fix' (#10) from master into dev

Reviewed-on: #10

.deploy/deploy-dev.yaml

@@ -9,6 +9,8 @@ services:
     environment:
       MONGO_HOST: "mongo.develop.sprinthub.ru"
       MONGO_PASSWORD: $MONGO_PASSWORD_DEV
+      REDIS_HOST: "redis.develop.sprinthub.ru"
+      REDIS_PASSWORD: $REDIS_PASSWORD_DEV
       STAGE: "development"
     deploy:
       mode: replicated

.deploy/deploy-prod.yaml

@@ -9,6 +9,8 @@ services:
     environment:
       MONGO_HOST: "mongo.sprinthub.ru"
       MONGO_PASSWORD: $MONGO_PASSWORD_PROD
+      REDIS_HOST: "redis.sprinthub.ru"
+      REDIS_PASSWORD: $REDIS_PASSWORD_PROD
       STAGE: "production"
     deploy:
       mode: replicated

.gitea/workflows/deploy-dev.yaml

@@ -40,4 +40,5 @@ jobs:
       - name: deploy
         env:
           MONGO_PASSWORD_DEV: ${{ secrets.MONGO_PASSWORD_DEV }}
+          REDIS_PASSWORD_DEV: ${{ secrets.REDIS_PASSWORD_DEV }}
         run: docker stack deploy --with-registry-auth -c ./.deploy/deploy-dev.yaml infra-development

.gitea/workflows/deploy-prod.yaml

@@ -40,4 +40,5 @@ jobs:
       - name: deploy
         env:
           MONGO_PASSWORD_PROD: ${{ secrets.MONGO_PASSWORD_PROD }}
+          REDIS_PASSWORD_PROD: ${{ secrets.REDIS_PASSWORD_PROD }}
         run: docker stack deploy --with-registry-auth -c ./.deploy/deploy-prod.yaml infra

app/routers/take.go

@@ -1,8 +1,11 @@
 package routers
 
 import (
+	"fmt"
 	"net/http"
 	tasks "queues-go/app/storage/mongo/collections"
+	"queues-go/app/storage/redis"
+	"time"
 )
 
 type TaskResponse struct {
@@ -17,7 +20,17 @@ type TakeResponse struct {
 
 func Take(r *http.Request) (interface{}, int) {
 	queue := r.Header.Get("queue")
+	mutex := redis.Sync.NewMutex(fmt.Sprintf("lock_queues_%s", queue))
+	for {
+		err := mutex.Lock()
+		if err != nil {
+			time.Sleep(time.Millisecond * 5)
+		} else {
+			break
+		}
+	}
 	task, err := tasks.Take(queue)
+	mutex.Unlock()
 	if err != nil {
 		return nil, http.StatusInternalServerError
 	}

app/storage/redis/locks.go

@@ -0,0 +1,28 @@
+package redis
+
+import (
+	"fmt"
+	"queues-go/app/utils"
+
+	"github.com/go-redsync/redsync/v4"
+	"github.com/go-redsync/redsync/v4/redis/goredis/v9"
+	goredislib "github.com/redis/go-redis/v9"
+)
+
+var Sync redsync.Redsync
+
+func Connect() {
+	client := goredislib.NewClient(getOptions())
+	pool := goredis.NewPool(client)
+	Sync = *redsync.New(pool)
+}
+
+func getOptions() *goredislib.Options {
+	addr := fmt.Sprintf("%s:6379", utils.GetEnv("REDIS_HOST", "localhost"))
+	password := utils.GetEnv("REDIS_PASSWORD", "password")
+
+	return &goredislib.Options{
+		Addr:     addr,
+		Password: password,
+	}
+}

go.mod

@@ -2,10 +2,17 @@ module queues-go
 
 go 1.23.4
 
+require github.com/go-redsync/redsync/v4 v4.13.0
+
 require (
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/golang/snappy v0.0.4 // indirect
+	github.com/hashicorp/errwrap v1.1.0 // indirect
+	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/klauspost/compress v1.13.6 // indirect
 	github.com/montanaflynn/stats v0.7.1 // indirect
+	github.com/redis/go-redis/v9 v9.5.1 // indirect
 	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
 	github.com/xdg-go/scram v1.1.2 // indirect
 	github.com/xdg-go/stringprep v1.0.4 // indirect

go.sum

@@ -1,9 +1,22 @@
+github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
+github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
+github.com/go-redsync/redsync/v4 v4.13.0 h1:49X6GJfnbLGaIpBBREM/zA4uIMDXKAh1NDkvQ1EkZKA=
+github.com/go-redsync/redsync/v4 v4.13.0/go.mod h1:HMW4Q224GZQz6x1Xc7040Yfgacukdzu7ifTDAKiyErQ=
 github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
 github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
+github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I=
+github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
+github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo=
+github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM=
 github.com/klauspost/compress v1.13.6 h1:P76CopJELS0TiO2mebmnzgWaajssP/EszplttgQxcgc=
 github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
 github.com/montanaflynn/stats v0.7.1 h1:etflOAAHORrCC44V+aR6Ftzort912ZU+YLiSTuV8eaE=
 github.com/montanaflynn/stats v0.7.1/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow=
+github.com/redis/go-redis/v9 v9.5.1 h1:H1X4D3yHPaYrkL5X06Wh6xNVM/pX0Ft4RV0vMGvLBh8=
+github.com/redis/go-redis/v9 v9.5.1/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0b/CLO2V2M=
 github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
 github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
 github.com/xdg-go/scram v1.1.2 h1:FHX5I5B4i4hKRVRBCFRxq1iQRej7WO3hhBuJf+UUySY=

main.go

@@ -6,6 +6,7 @@ import (
 	"net/http"
 	"queues-go/app/routers"
 	client "queues-go/app/storage/mongo"
+	"queues-go/app/storage/redis"
 	"time"
 )
 
@@ -30,6 +31,7 @@ func handlerWrapper(f func(*http.Request) (interface{}, int)) func(http.Response
 
 func main() {
 	client.Connect()
+	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))