feat(worker): add timer in Request method (#12)

Author: appleboy

_example/single/go.mod

@@ -4,7 +4,7 @@ go 1.16
 
 require (
 	github.com/golang-queue/nsq v0.0.0-00010101000000-000000000000 // indirect
-	github.com/golang-queue/queue v0.0.12-0.20220122071422-2c41650f045a
+	github.com/golang-queue/queue v0.0.13-0.20220330060848-d1a0d31ce747
 )
 
 replace github.com/golang-queue/nsq => ../../

_example/single/go.sum

@@ -2,6 +2,8 @@ github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/golang-queue/queue v0.0.12-0.20220122071422-2c41650f045a h1:olgbdu53u5A6Babr1w+CUO19UdZASYPPqqIBfdFWrWE=
 github.com/golang-queue/queue v0.0.12-0.20220122071422-2c41650f045a/go.mod h1:ku8iyjYffqYY6Duts+xl+QYfN3/KDK4MEvXMZUkHyio=
+github.com/golang-queue/queue v0.0.13-0.20220330060848-d1a0d31ce747 h1:uNTbCoWORAcna89KcKgP22WFGv5fsij05e70DCnLrUU=
+github.com/golang-queue/queue v0.0.13-0.20220330060848-d1a0d31ce747/go.mod h1:ku8iyjYffqYY6Duts+xl+QYfN3/KDK4MEvXMZUkHyio=
 github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
 github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=

_example/single/main.go

@@ -65,6 +65,6 @@ func main() {
 	// wait until all tasks done
 	for i := 0; i < taskN; i++ {
 		fmt.Println("message:", <-rets)
-		time.Sleep(50 * time.Millisecond)
+		time.Sleep(10 * time.Millisecond)
 	}
 }

nsq.go

@@ -198,15 +198,24 @@ func (w *Worker) Queue(job queue.QueuedMessage) error {
 
 // Request fetch new task from queue
 func (w *Worker) Request() (queue.QueuedMessage, error) {
-	select {
-	case task, ok := <-w.tasks:
-		if !ok {
-			return nil, queue.ErrQueueHasBeenClosed
+	clock := 0
+loop:
+	for {
+		select {
+		case task, ok := <-w.tasks:
+			if !ok {
+				return nil, queue.ErrQueueHasBeenClosed
+			}
+			var data queue.Job
+			_ = json.Unmarshal(task.Body, &data)
+			return data, nil
+		case <-time.After(1 * time.Second):
+			if clock == 5 {
+				break loop
+			}
+			clock += 1
 		}
-		var data queue.Job
-		_ = json.Unmarshal(task.Body, &data)
-		return data, nil
-	default:
-		return nil, queue.ErrNoTaskInQueue
 	}
+
+	return nil, queue.ErrNoTaskInQueue
 }