feat: optimize message handling and improve tests

appleboy · appleboy · commit dcbcb6a745b4 · 2023-06-25T00:22:52.000+08:00
- Change message references to pointers in various functions
- Add dependency: github.com/appleboy/com v0.1.7
- Add new benchmark test: BenchmarkNewMessage
- Update string to byte conversion using bytesconv.StrToBytes from appleboy/com package
- Update comments in queue.go for better clarity

Signed-off-by: appleboy &lt;appleboy.tw@gmail.com&gt;
diff --git a/benchmark_test.go b/benchmark_test.go
@@ -31,7 +31,7 @@ func testQueue(b *testing.B, pool testqueue) {
 	b.ResetTimer()
 	for n := 0; n < b.N; n++ {
 		for i := 0; i < count; i++ {
-			_ = pool.Queue(message)
+			_ = pool.Queue(&message)
 			_, _ = pool.Request()
 		}
 	}
@@ -60,7 +60,7 @@ func BenchmarkQueueTask(b *testing.B) {
 	})
 
 	for n := 0; n < b.N; n++ {
-		if err := q.queue(m); err != nil {
+		if err := q.queue(&m); err != nil {
 			b.Fatal(err)
 		}
 	}
@@ -81,7 +81,7 @@ func BenchmarkQueue(b *testing.B) {
 	m.Encode()
 
 	for n := 0; n < b.N; n++ {
-		if err := q.queue(m); err != nil {
+		if err := q.queue(&m); err != nil {
 			b.Fatal(err)
 		}
 	}
@@ -113,7 +113,7 @@ func BenchmarkQueue(b *testing.B) {
 func BenchmarkRingWithTask(b *testing.B) {
 	b.ReportAllocs()
 
-	task := &job.Message{
+	task := job.Message{
 		Timeout: 100 * time.Millisecond,
 		Task: func(_ context.Context) error {
 			return nil
@@ -131,6 +131,6 @@ func BenchmarkRingWithTask(b *testing.B) {
 	)
 
 	for n := 0; n < b.N; n++ {
-		_ = q.run(task)
+		_ = q.run(&task)
 	}
 }
diff --git a/go.mod b/go.mod
@@ -3,6 +3,7 @@ module github.com/golang-queue/queue
 go 1.18
 
 require (
+	github.com/appleboy/com v0.1.7
 	github.com/golang/mock v1.6.0
 	github.com/jpillora/backoff v1.0.0
 	github.com/stretchr/testify v1.8.4
diff --git a/go.sum b/go.sum
@@ -1,3 +1,5 @@
+github.com/appleboy/com v0.1.7 h1:4lYTFNoMAAXGGIC8lDxVg/NY+1aXbYqfAWN05cZhd0M=
+github.com/appleboy/com v0.1.7/go.mod h1:JUK+oH0SXCLRH57pDMJx6VWVsm8CPdajalmRSWwamBE=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 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=
diff --git a/job/benchmark_test.go b/job/benchmark_test.go
@@ -21,6 +21,21 @@ func BenchmarkNewTask(b *testing.B) {
 	}
 }
 
+func BenchmarkNewMessage(b *testing.B) {
+	b.ReportAllocs()
+	for i := 0; i < b.N; i++ {
+		_ = NewMessage(mockMessage{
+			message: "foo",
+		},
+			AllowOption{
+				RetryCount: Int64(100),
+				RetryDelay: Time(30 * time.Millisecond),
+				Timeout:    Time(3 * time.Millisecond),
+			},
+		)
+	}
+}
+
 func BenchmarkNewOption(b *testing.B) {
 	b.ReportAllocs()
 	for i := 0; i < b.N; i++ {
diff --git a/job/job.go b/job/job.go
@@ -68,10 +68,10 @@ func (m *Message) Encode() {
 }
 
 // NewMessage create new message
-func NewMessage(m core.QueuedMessage, opts ...AllowOption) *Message {
+func NewMessage(m core.QueuedMessage, opts ...AllowOption) Message {
 	o := NewOptions(opts...)
 
-	return &Message{
+	return Message{
 		RetryCount:  o.retryCount,
 		RetryDelay:  o.retryDelay,
 		RetryFactor: o.retryFactor,
@@ -83,10 +83,10 @@ func NewMessage(m core.QueuedMessage, opts ...AllowOption) *Message {
 }
 
 // NewTask create new task
-func NewTask(task TaskFunc, opts ...AllowOption) *Message {
+func NewTask(task TaskFunc, opts ...AllowOption) Message {
 	o := NewOptions(opts...)
 
-	return &Message{
+	return Message{
 		Timeout:     o.timeout,
 		RetryCount:  o.retryCount,
 		RetryDelay:  o.retryDelay,
diff --git a/job/job_test.go b/job/job_test.go
@@ -4,6 +4,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/appleboy/com/bytesconv"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -12,7 +13,7 @@ type mockMessage struct {
 }
 
 func (m mockMessage) Bytes() []byte {
-	return []byte(m.message)
+	return bytesconv.StrToBytes(m.message)
 }
 
 func TestMessageEncodeDecode(t *testing.T) {
diff --git a/queue.go b/queue.go
@@ -119,13 +119,13 @@ func (q *Queue) Queue(message core.QueuedMessage, opts ...job.AllowOption) error
 	data := job.NewMessage(message, opts...)
 	data.Encode()
 
-	return q.queue(data)
+	return q.queue(&data)
 }
 
 // QueueTask to queue single task
 func (q *Queue) QueueTask(task job.TaskFunc, opts ...job.AllowOption) error {
 	data := job.NewTask(task, opts...)
-	return q.queue(data)
+	return q.queue(&data)
 }
 
 func (q *Queue) queue(m *job.Message) error {
@@ -266,6 +266,7 @@ func (q *Queue) UpdateWorkerCount(num int) {
 	q.schedule()
 }
 
+// schedule to check worker number
 func (q *Queue) schedule() {
 	q.Lock()
 	defer q.Unlock()
@@ -279,7 +280,7 @@ func (q *Queue) schedule() {
 	}
 }
 
-// start handle job
+// start to start all worker
 func (q *Queue) start() {
 	tasks := make(chan core.QueuedMessage, 1)
 
diff --git a/queue_test.go b/queue_test.go
@@ -6,6 +6,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/appleboy/com/bytesconv"
 	"github.com/golang-queue/queue/core"
 	"github.com/golang-queue/queue/job"
 	"github.com/golang-queue/queue/mocks"
@@ -24,7 +25,7 @@ type mockMessage struct {
 }
 
 func (m mockMessage) Bytes() []byte {
-	return []byte(m.message)
+	return bytesconv.StrToBytes(m.message)
 }
 
 func TestNewQueueWithZeroWorker(t *testing.T) {

Original file line number	Diff line number	Diff line change
`@@ -31,7 +31,7 @@ func testQueue(b *testing.B, pool testqueue) {`
`31`	`31`	`b.ResetTimer()`
`32`	`32`	`for n := 0; n < b.N; n++ {`
`33`	`33`	`for i := 0; i < count; i++ {`
`34`		`- _ = pool.Queue(message)`
	`34`	`+ _ = pool.Queue(&message)`
`35`	`35`	`_, _ = pool.Request()`
`36`	`36`	`}`
`37`	`37`	`}`
`@@ -60,7 +60,7 @@ func BenchmarkQueueTask(b *testing.B) {`
`60`	`60`	`})`
`61`	`61`
`62`	`62`	`for n := 0; n < b.N; n++ {`
`63`		`- if err := q.queue(m); err != nil {`
	`63`	`+ if err := q.queue(&m); err != nil {`
`64`	`64`	`b.Fatal(err)`
`65`	`65`	`}`
`66`	`66`	`}`
`@@ -81,7 +81,7 @@ func BenchmarkQueue(b *testing.B) {`
`81`	`81`	`m.Encode()`
`82`	`82`
`83`	`83`	`for n := 0; n < b.N; n++ {`
`84`		`- if err := q.queue(m); err != nil {`
	`84`	`+ if err := q.queue(&m); err != nil {`
`85`	`85`	`b.Fatal(err)`
`86`	`86`	`}`
`87`	`87`	`}`
`@@ -113,7 +113,7 @@ func BenchmarkQueue(b *testing.B) {`
`113`	`113`	`func BenchmarkRingWithTask(b *testing.B) {`
`114`	`114`	`b.ReportAllocs()`
`115`	`115`
`116`		`- task := &job.Message{`
	`116`	`+ task := job.Message{`
`117`	`117`	`Timeout: 100 * time.Millisecond,`
`118`	`118`	`Task: func(_ context.Context) error {`
`119`	`119`	`return nil`
`@@ -131,6 +131,6 @@ func BenchmarkRingWithTask(b *testing.B) {`
`131`	`131`	`)`
`132`	`132`
`133`	`133`	`for n := 0; n < b.N; n++ {`
`134`		`- _ = q.run(task)`
	`134`	`+ _ = q.run(&task)`
`135`	`135`	`}`
`136`	`136`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,3 +1,5 @@`
	`1`	`+github.com/appleboy/com v0.1.7 h1:4lYTFNoMAAXGGIC8lDxVg/NY+1aXbYqfAWN05cZhd0M=`
	`2`	`+github.com/appleboy/com v0.1.7/go.mod h1:JUK+oH0SXCLRH57pDMJx6VWVsm8CPdajalmRSWwamBE=`
`1`	`3`	`github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=`
`2`	`4`	`github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=`
`3`	`5`	`github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=`
Original file line number	Diff line number	Diff line change
`@@ -4,6 +4,7 @@ import (`
`4`	`4`	`"testing"`
`5`	`5`	`"time"`
`6`	`6`
	`7`	`+ "github.com/appleboy/com/bytesconv"`
`7`	`8`	`"github.com/stretchr/testify/assert"`
`8`	`9`	`)`
`9`	`10`
`@@ -12,7 +13,7 @@ type mockMessage struct {`
`12`	`13`	`}`
`13`	`14`
`14`	`15`	`func (m mockMessage) Bytes() []byte {`
`15`		`- return []byte(m.message)`
	`16`	`+ return bytesconv.StrToBytes(m.message)`
`16`	`17`	`}`
`17`	`18`
`18`	`19`	`func TestMessageEncodeDecode(t *testing.T) {`
Original file line number	Diff line number	Diff line change
`@@ -119,13 +119,13 @@ func (q *Queue) Queue(message core.QueuedMessage, opts ...job.AllowOption) error`
`119`	`119`	`data := job.NewMessage(message, opts...)`
`120`	`120`	`data.Encode()`
`121`	`121`
`122`		`- return q.queue(data)`
	`122`	`+ return q.queue(&data)`
`123`	`123`	`}`
`124`	`124`
`125`	`125`	`// QueueTask to queue single task`
`126`	`126`	`func (q *Queue) QueueTask(task job.TaskFunc, opts ...job.AllowOption) error {`
`127`	`127`	`data := job.NewTask(task, opts...)`
`128`		`- return q.queue(data)`
	`128`	`+ return q.queue(&data)`
`129`	`129`	`}`
`130`	`130`
`131`	`131`	`func (q Queue) queue(m job.Message) error {`
`@@ -266,6 +266,7 @@ func (q *Queue) UpdateWorkerCount(num int) {`
`266`	`266`	`q.schedule()`
`267`	`267`	`}`
`268`	`268`
	`269`	`+// schedule to check worker number`
`269`	`270`	`func (q *Queue) schedule() {`
`270`	`271`	`q.Lock()`
`271`	`272`	`defer q.Unlock()`
`@@ -279,7 +280,7 @@ func (q *Queue) schedule() {`
`279`	`280`	`}`
`280`	`281`	`}`
`281`	`282`
`282`		`-// start handle job`
	`283`	`+// start to start all worker`
`283`	`284`	`func (q *Queue) start() {`
`284`	`285`	`tasks := make(chan core.QueuedMessage, 1)`
`285`	`286`
Original file line number	Diff line number	Diff line change
`@@ -6,6 +6,7 @@ import (`
`6`	`6`	`"testing"`
`7`	`7`	`"time"`
`8`	`8`
	`9`	`+ "github.com/appleboy/com/bytesconv"`
`9`	`10`	`"github.com/golang-queue/queue/core"`
`10`	`11`	`"github.com/golang-queue/queue/job"`
`11`	`12`	`"github.com/golang-queue/queue/mocks"`
`@@ -24,7 +25,7 @@ type mockMessage struct {`
`24`	`25`	`}`
`25`	`26`
`26`	`27`	`func (m mockMessage) Bytes() []byte {`
`27`		`- return []byte(m.message)`
	`28`	`+ return bytesconv.StrToBytes(m.message)`
`28`	`29`	`}`
`29`	`30`
`30`	`31`	`func TestNewQueueWithZeroWorker(t *testing.T) {`