Merge pull request #52 from tinh-tinh/feat/ren/51-add-processor

feat: add processor to easy handle

Author: ren0503

go.mod

@@ -1,19 +1,20 @@
 module github.com/tinh-tinh/queue/v2
 
-go 1.22.0
+go 1.23.0
+
 toolchain go1.24.1
 
 require (
 	github.com/go-redsync/redsync/v4 v4.13.0
-	github.com/redis/go-redis/v9 v9.6.3
-	github.com/robfig/cron/v3 v3.0.0
+	github.com/redis/go-redis/v9 v9.10.0
+	github.com/robfig/cron/v3 v3.0.1
 	github.com/stretchr/testify v1.9.0
-	github.com/tinh-tinh/tinhtinh/v2 v2.0.0
+	github.com/tinh-tinh/tinhtinh/v2 v2.1.3
 	golang.org/x/crypto v0.35.0
 )
 
 require (
-	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect

go.sum

@@ -4,6 +4,8 @@ github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
 github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 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/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
@@ -27,16 +29,24 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/redis/go-redis/v9 v9.6.3 h1:8Dr5ygF1QFXRxIH/m3Xg9MMG1rS8YCtAgosrsewT6i0=
 github.com/redis/go-redis/v9 v9.6.3/go.mod h1:0C0c6ycQsdpVNQpxb1njEQIqkx5UcsM8FJCQLgE9+RA=
+github.com/redis/go-redis/v9 v9.10.0 h1:FxwK3eV8p/CQa0Ch276C7u2d0eNC9kCmAYQ7mCXCzVs=
+github.com/redis/go-redis/v9 v9.10.0/go.mod h1:huWgSWd8mW6+m0VPhJjSSQ+d6Nh1VICQ6Q5lHuCH/Iw=
 github.com/redis/rueidis v1.0.19 h1:s65oWtotzlIFN8eMPhyYwxlwLR1lUdhza2KtWprKYSo=
 github.com/redis/rueidis v1.0.19/go.mod h1:8B+r5wdnjwK3lTFml5VtxjzGOQAC+5UmujoD12pDrEo=
 github.com/robfig/cron/v3 v3.0.0 h1:kQ6Cb7aHOHTSzNVNEhmp8EcWKLb4CbiMW9h9VyIhO4E=
 github.com/robfig/cron/v3 v3.0.0/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
+github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
+github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/stvp/tempredis v0.0.0-20181119212430-b82af8480203 h1:QVqDTf3h2WHt08YuiTGPZLls0Wq99X9bWd0Q5ZSBesM=
 github.com/stvp/tempredis v0.0.0-20181119212430-b82af8480203/go.mod h1:oqN97ltKNihBbwlX8dLpwxCl3+HnXKV/R0e+sRLd9C8=
 github.com/tinh-tinh/tinhtinh/v2 v2.0.0 h1:+AUbzU4EHS2HuXA7pk+bswRSURd9duF4vBLRTjg4vmk=
 github.com/tinh-tinh/tinhtinh/v2 v2.0.0/go.mod h1:4nppE7KAIswZKutI9ElMqAD9kyash7aea0Ewowsqj5g=
+github.com/tinh-tinh/tinhtinh/v2 v2.1.2 h1:rd3m0YMt2Og1VTgF9ktWkj86FY3+caCvJNxhSw/qpnk=
+github.com/tinh-tinh/tinhtinh/v2 v2.1.2/go.mod h1:4nppE7KAIswZKutI9ElMqAD9kyash7aea0Ewowsqj5g=
+github.com/tinh-tinh/tinhtinh/v2 v2.1.3 h1:ug+9XFhKH/Z6jY5TqV5ioWq2kOhO/9q1+jG7wGjHp2s=
+github.com/tinh-tinh/tinhtinh/v2 v2.1.3/go.mod h1:4nppE7KAIswZKutI9ElMqAD9kyash7aea0Ewowsqj5g=
 golang.org/x/crypto v0.35.0 h1:b15kiHdrGCHrP6LvwaQ3c03kgNhhiMgvlhxHQhmg2Xs=
 golang.org/x/crypto v0.35.0/go.mod h1:dy7dXNW32cAb/6/PRuTNsix8T+vJAqvuIy5Bli/x0YQ=
 golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=

processor.go

@@ -0,0 +1,23 @@
+package queue
+
+import "github.com/tinh-tinh/tinhtinh/v2/core"
+
+type Processor struct {
+	core.DynamicProvider
+	module core.Module
+	name   string
+}
+
+func NewProcessor(name string, module core.Module) *Processor {
+	return &Processor{
+		module: module,
+		name:   name,
+	}
+}
+
+func (p *Processor) Process(jobFnc JobFnc) {
+	pQueue := Inject(p.module, p.name)
+	if pQueue != nil {
+		pQueue.Process(jobFnc)
+	}
+}

processor_test.go

@@ -0,0 +1,91 @@
+package queue_test
+
+import (
+	"encoding/json"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+	"github.com/stretchr/testify/require"
+	"github.com/tinh-tinh/queue/v2"
+	"github.com/tinh-tinh/tinhtinh/v2/core"
+)
+
+func Test_Proccessor(t *testing.T) {
+	service := func(module core.Module) core.Provider {
+		processor := queue.NewProcessor("test", module)
+
+		processor.Process(func(job *queue.Job) {
+			job.Process(func() error {
+				num, err := strconv.Atoi(job.Id)
+				require.Nil(t, err)
+				if num%3 == 0 {
+					return errors.New("error by test")
+				}
+
+				key, err := json.Marshal(job.Data)
+				require.Nil(t, err)
+				_, err = HeaveTask(string(key))
+				require.Nil(t, err)
+				return nil
+			})
+		})
+
+		return processor
+	}
+
+	controller := func(module core.Module) core.Controller {
+		ctrl := module.NewController("test")
+
+		testQueue := queue.Inject(module, "test")
+		ctrl.Post("", func(ctx core.Ctx) error {
+			testQueue.AddJob(queue.AddJobOptions{
+				Id:   "1",
+				Data: "value 1",
+			})
+			return ctx.JSON(core.Map{
+				"data": "ok",
+			})
+		})
+
+		return ctrl
+	}
+
+	appMoule := func() core.Module {
+		return core.NewModule(core.NewModuleOptions{
+			Imports: []core.Modules{
+				queue.ForRootFactory(func(ref core.RefProvider) *queue.Options {
+					return &queue.Options{
+						Connect: &redis.Options{
+							Addr:     "localhost:6379",
+							DB:       0,
+							Password: "",
+						},
+						Workers:       6,
+						RetryFailures: 3,
+					}
+				}),
+				queue.Register("test"),
+			},
+			Controllers: []core.Controllers{controller},
+			Providers:   []core.Providers{service},
+		})
+	}
+
+	server := core.CreateFactory(appMoule)
+	server.SetGlobalPrefix("api")
+
+	testServer := httptest.NewServer(server.PrepareBeforeListen())
+	defer testServer.Close()
+
+	testClient := testServer.Client()
+	resp, err := testClient.Post(testServer.URL+"/api/test", "application/json", nil)
+	require.Nil(t, err)
+	require.Equal(t, http.StatusOK, resp.StatusCode)
+
+	time.Sleep(1 * time.Second)
+}

queue.go

@@ -13,6 +13,7 @@ import (
 	"github.com/go-redsync/redsync/v4/redis/goredis/v9"
 	"github.com/redis/go-redis/v9"
 	"github.com/robfig/cron/v3"
+	"github.com/tinh-tinh/tinhtinh/v2/common"
 )
 
 type JobFnc func(job *Job)
@@ -362,15 +363,15 @@ func (q *Queue) Resume() {
 
 func (q *Queue) RemoveCompleted() {
 	if q.config.RemoveOnComplete {
-		q.jobs = Remove(q.jobs, func(j Job) bool {
+		q.jobs = common.Remove(q.jobs, func(j Job) bool {
 			return j.Status == CompletedStatus
 		})
 	}
 }
 
 func (q *Queue) RemoveFailed() {
 	if q.config.RemoveOnFail {
-		q.jobs = Remove(q.jobs, func(j Job) bool {
+		q.jobs = common.Remove(q.jobs, func(j Job) bool {
 			return j.Status == FailedStatus
 		})
 	}