chore: Add job timeout in body with bytes format. (#23)

Author: appleboy

queue.go

@@ -1,6 +1,7 @@
 package queue
 
 import (
+	"encoding/json"
 	"errors"
 	"runtime"
 	"sync"
@@ -37,6 +38,11 @@ func (j Job) Bytes() []byte {
 	return j.Body
 }
 
+func (j Job) Encode() []byte {
+	b, _ := json.Marshal(j)
+	return b
+}
+
 // Option for queue system
 type Option func(*Queue)
 
@@ -126,28 +132,29 @@ func (q *Queue) Wait() {
 	q.routineGroup.Wait()
 }
 
-// Queue to queue all job
-func (q *Queue) Queue(job QueuedMessage) error {
+func (q *Queue) handleQueue(timeout time.Duration, job QueuedMessage) error {
 	if atomic.LoadInt32(&q.stopFlag) == 1 {
 		return ErrQueueShutdown
 	}
 
-	return q.worker.Queue(Job{
-		Timeout: q.timeout,
+	data := Job{
+		Timeout: timeout,
 		Body:    job.Bytes(),
+	}
+
+	return q.worker.Queue(Job{
+		Body: data.Encode(),
 	})
 }
 
 // Queue to queue all job
-func (q *Queue) QueueWithTimeout(timeout time.Duration, job QueuedMessage) error {
-	if atomic.LoadInt32(&q.stopFlag) == 1 {
-		return ErrQueueShutdown
-	}
+func (q *Queue) Queue(job QueuedMessage) error {
+	return q.handleQueue(q.timeout, job)
+}
 
-	return q.worker.Queue(Job{
-		Timeout: timeout,
-		Body:    job.Bytes(),
-	})
+// Queue to queue all job
+func (q *Queue) QueueWithTimeout(timeout time.Duration, job QueuedMessage) error {
+	return q.handleQueue(q.timeout, job)
 }
 
 func (q *Queue) work() {

simple/simple.go

@@ -2,6 +2,7 @@ package simple
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"sync"
 	"sync/atomic"
@@ -39,11 +40,10 @@ func (s *Worker) AfterRun() error {
 	return nil
 }
 
-func (s *Worker) handle(m interface{}) error {
+func (s *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)
-	job, _ := m.(queue.Job)
 	startTime := time.Now()
 	ctx, cancel := context.WithTimeout(context.Background(), job.Timeout)
 	defer cancel()
@@ -95,7 +95,9 @@ func (s *Worker) Run() error {
 	}
 
 	for task := range s.taskQueue {
-		if err := s.handle(task); err != nil {
+		var data queue.Job
+		_ = json.Unmarshal(task.Bytes(), &data)
+		if err := s.handle(data); err != nil {
 			s.logger.Error(err.Error())
 		}
 	}