chore(queue): update to latest version (#24)

Author: appleboy

go.mod

@@ -3,8 +3,8 @@ module github.com/golang-queue/redisdb
 go 1.18
 
 require (
-	github.com/go-redis/redis/v8 v8.11.5
-	github.com/golang-queue/queue v0.1.3
+	github.com/go-redis/redis/v9 v9.0.0-rc.2
+	github.com/golang-queue/queue v0.1.4-0.20221210024521-cb8720b0c721
 	github.com/stretchr/testify v1.8.1
 	go.uber.org/goleak v1.2.0
 )
@@ -13,8 +13,7 @@ require (
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
-	github.com/goccy/go-json v0.9.7 // indirect
-	github.com/kr/text v0.2.0 // indirect
+	github.com/goccy/go-json v0.10.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -1,25 +1,24 @@
 github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
 github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
-github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 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/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
-github.com/go-redis/redis/v8 v8.11.5 h1:AcZZR7igkdvfVmQTPnu9WE37LRrO/YrBH5zWyjDC0oI=
-github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo=
-github.com/goccy/go-json v0.9.7 h1:IcB+Aqpx/iMHu5Yooh7jEzJk1JZ7Pjtmys2ukPr7EeM=
-github.com/goccy/go-json v0.9.7/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I=
-github.com/golang-queue/queue v0.1.3 h1:FGIrn8e0fN8EmL3glP0rFEcYVtWUGMEeqX4h4nnzh40=
-github.com/golang-queue/queue v0.1.3/go.mod h1:h/PhaoMwT5Jc4sQNus7APgWBUItm6QC9k6JtmwrsRos=
+github.com/go-redis/redis/v9 v9.0.0-rc.2 h1:IN1eI8AvJJeWHjMW/hlFAv2sAfvTun2DVksDDJ3a6a0=
+github.com/go-redis/redis/v9 v9.0.0-rc.2/go.mod h1:cgBknjwcBJa2prbnuHH/4k/Mlj4r0pWNV2HBanHujfY=
+github.com/goccy/go-json v0.10.0 h1:mXKd9Qw4NuzShiRlOXKews24ufknHO7gx30lsDyokKA=
+github.com/goccy/go-json v0.10.0/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I=
+github.com/golang-queue/queue v0.1.4-0.20221210024521-cb8720b0c721 h1:sYfb8dgd4PQkbDNUFAeVI/RkxCgFJPx49hTILog/CNU=
+github.com/golang-queue/queue v0.1.4-0.20221210024521-cb8720b0c721/go.mod h1:8P7IgwdxwKh0/W1I9yCuQQGI8OHIuc7fIHi4OYr1COU=
 github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc=
+github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
 github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
-github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
 github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
-github.com/onsi/gomega v1.18.1 h1:M1GfJqGRrBrrGGsbxzV5dqM2U2ApXefZCQpkukxYRLE=
+github.com/onsi/gomega v1.24.1 h1:KORJXNNTzJXzu4ScJWssJfJMnJ+2QJqhoQSRwNlze9E=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
@@ -32,14 +31,13 @@ github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o
 go.uber.org/goleak v1.2.0 h1:xqgm/S+aQvhWFTtR0XK3Jvg7z8kGV8P4X14IzwN3Eqk=
 go.uber.org/goleak v1.2.0/go.mod h1:XJYK+MuIchqpmGmUSAzotztawfKvYLUIgg7guXrwVUo=
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de h1:5hukYrvBGR8/eNkX5mdUezrA6JiaEZDtJb9Ei+1LlBs=
-golang.org/x/net v0.0.0-20210428140749-89ef3d95e781 h1:DzZ89McO9/gWPsQXS/FVKAlG02ZjaQ6AlZRBimEYOd0=
-golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e h1:fLOSk5Q00efkSvAm+4xcoXD+RRmLmmulPn5I3Y9F2EM=
-golang.org/x/text v0.3.6 h1:aRYxNxv6iGQlyVaZmk6ZgYEDa+Jg18DxebPSrd6bg1M=
+golang.org/x/net v0.2.0 h1:sZfSu1wtKLGlWI4ZZayP0ck9Y73K1ynO6gqzTdBVdPU=
+golang.org/x/sys v0.2.0 h1:ljd4t30dBnAvMZaQCevtY0xLLD0A+bRZXbgLMLU1F/A=
+golang.org/x/text v0.4.0 h1:BrVqGRd7+k1DiOgtnFvAkoQEWQvBc25ouMJM6429SFg=
 golang.org/x/tools v0.1.5 h1:ouewzE6p+/VEB31YYnTbEJdi8pFqKp4P4n85vwo3DHA=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
-gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

redis.go

@@ -8,9 +8,11 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/go-redis/redis/v8"
 	"github.com/golang-queue/queue"
 	"github.com/golang-queue/queue/core"
+	"github.com/golang-queue/queue/job"
+
+	"github.com/go-redis/redis/v9"
 )
 
 var _ core.Worker = (*Worker)(nil)
@@ -86,51 +88,9 @@ func NewWorker(opts ...Option) *Worker {
 	return w
 }
 
-func (w *Worker) handle(job *queue.Job) error {
-	// create channel with buffer size 1 to avoid goroutine leak
-	done := make(chan error, 1)
-	panicChan := make(chan interface{}, 1)
-	startTime := time.Now()
-	ctx, cancel := context.WithTimeout(context.Background(), job.Timeout)
-	defer func() {
-		cancel()
-	}()
-
-	// run the job
-	go func() {
-		// handle panic issue
-		defer func() {
-			if p := recover(); p != nil {
-				panicChan <- p
-			}
-		}()
-
-		// run custom process function
-		done <- w.opts.runFunc(ctx, job)
-	}()
-
-	select {
-	case p := <-panicChan:
-		panic(p)
-	case <-ctx.Done(): // timeout reached
-		return ctx.Err()
-	case <-w.stop: // shutdown service
-		// cancel job
-		cancel()
-
-		leftTime := job.Timeout - time.Since(startTime)
-		// wait job
-		select {
-		case <-time.After(leftTime):
-			return context.DeadlineExceeded
-		case err := <-done: // job finish
-			return err
-		case p := <-panicChan:
-			panic(p)
-		}
-	case err := <-done: // job finish
-		return err
-	}
+// Run to execute new task
+func (w *Worker) Run(ctx context.Context, task core.QueuedMessage) error {
+	return w.opts.runFunc(ctx, task)
 }
 
 // Shutdown worker
@@ -169,17 +129,6 @@ func (w *Worker) Queue(job core.QueuedMessage) error {
 	return nil
 }
 
-// Run start the worker
-func (w *Worker) Run(task core.QueuedMessage) error {
-	data, _ := task.(*queue.Job)
-
-	if err := w.handle(data); err != nil {
-		return err
-	}
-
-	return nil
-}
-
 // Request a new task
 func (w *Worker) Request() (core.QueuedMessage, error) {
 	clock := 0
@@ -190,7 +139,7 @@ loop:
 			if !ok {
 				return nil, queue.ErrQueueHasBeenClosed
 			}
-			var data queue.Job
+			var data job.Message
 			_ = json.Unmarshal([]byte(task.Payload), &data)
 			return &data, nil
 		case <-time.After(1 * time.Second):

redis_test.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/golang-queue/queue"
 	"github.com/golang-queue/queue/core"
+	"github.com/golang-queue/queue/job"
 
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/goleak"
@@ -182,7 +183,7 @@ func TestJobReachTimeout(t *testing.T) {
 	assert.NoError(t, err)
 	q.Start()
 	time.Sleep(50 * time.Millisecond)
-	assert.NoError(t, q.QueueWithTimeout(20*time.Millisecond, m))
+	assert.NoError(t, q.Queue(m, job.WithTimeout(20*time.Millisecond)))
 	time.Sleep(2 * time.Second)
 	q.Shutdown()
 	q.Wait()
@@ -220,7 +221,7 @@ func TestCancelJobAfterShutdown(t *testing.T) {
 	assert.NoError(t, err)
 	q.Start()
 	time.Sleep(50 * time.Millisecond)
-	assert.NoError(t, q.QueueWithTimeout(3*time.Second, m))
+	assert.NoError(t, q.Queue(m, job.WithTimeout(3*time.Second)))
 	time.Sleep(2 * time.Second)
 	q.Shutdown()
 	q.Wait()
@@ -296,84 +297,3 @@ func TestGoroutinePanic(t *testing.T) {
 	assert.Error(t, q.Queue(m))
 	q.Wait()
 }
-
-func TestHandleTimeout(t *testing.T) {
-	job := &queue.Job{
-		Timeout: 100 * time.Millisecond,
-		Payload: []byte("foo"),
-	}
-	w := NewWorker(
-		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
-			time.Sleep(200 * time.Millisecond)
-			return nil
-		}),
-	)
-
-	err := w.handle(job)
-	assert.Error(t, err)
-	assert.Equal(t, context.DeadlineExceeded, err)
-	assert.NoError(t, w.Shutdown())
-
-	job = &queue.Job{
-		Timeout: 150 * time.Millisecond,
-		Payload: []byte("foo"),
-	}
-
-	w = NewWorker(
-		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
-			time.Sleep(200 * time.Millisecond)
-			return nil
-		}),
-	)
-
-	done := make(chan error)
-	go func() {
-		done <- w.handle(job)
-	}()
-
-	assert.NoError(t, w.Shutdown())
-
-	err = <-done
-	assert.Error(t, err)
-	assert.Equal(t, context.DeadlineExceeded, err)
-}
-
-func TestJobComplete(t *testing.T) {
-	job := &queue.Job{
-		Timeout: 100 * time.Millisecond,
-		Payload: []byte("foo"),
-	}
-	w := NewWorker(
-		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
-			return errors.New("job completed")
-		}),
-	)
-
-	err := w.handle(job)
-	assert.Error(t, err)
-	assert.Equal(t, errors.New("job completed"), err)
-	assert.NoError(t, w.Shutdown())
-
-	job = &queue.Job{
-		Timeout: 250 * time.Millisecond,
-		Payload: []byte("foo"),
-	}
-
-	w = NewWorker(
-		WithRunFunc(func(ctx context.Context, m core.QueuedMessage) error {
-			time.Sleep(200 * time.Millisecond)
-			return errors.New("job completed")
-		}),
-	)
-
-	done := make(chan error)
-	go func() {
-		done <- w.handle(job)
-	}()
-
-	assert.NoError(t, w.Shutdown())
-
-	err = <-done
-	assert.Error(t, err)
-	assert.Equal(t, errors.New("job completed"), err)
-}