Make sure 'Fetcher' doesn't have a lingering job upon exit (#79)

Author: lxfontes

fetcher.go

@@ -54,40 +54,32 @@ func (f *fetch) processOldMessages() {
 }
 
 func (f *fetch) Fetch() {
-	messages := make(chan string)
-
 	f.processOldMessages()
 
-	go func(c chan string) {
+	go func() {
 		for {
 			// f.Close() has been called
 			if f.Closed() {
 				break
 			}
 			<-f.Ready()
-			f.tryFetchMessage(c)
+			f.tryFetchMessage()
 		}
-	}(messages)
-
-	f.handleMessages(messages)
-}
+	}()
 
-func (f *fetch) handleMessages(messages chan string) {
 	for {
 		select {
-		case message := <-messages:
-			f.sendMessage(message)
 		case <-f.stop:
 			// Stop the redis-polling goroutine
 			close(f.closed)
 			// Signal to Close() that the fetcher has stopped
 			close(f.exit)
-			return
+			break
 		}
 	}
 }
 
-func (f *fetch) tryFetchMessage(messages chan string) {
+func (f *fetch) tryFetchMessage() {
 	conn := Config.Pool.Get()
 	defer conn.Close()
 
@@ -100,7 +92,7 @@ func (f *fetch) tryFetchMessage(messages chan string) {
 			time.Sleep(1 * time.Second)
 		}
 	} else {
-		messages <- message
+		f.sendMessage(message)
 	}
 }
 

manager_test.go

@@ -3,6 +3,7 @@ package workers
 import (
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/customerio/gospec"
 	. "github.com/customerio/gospec"
@@ -99,6 +100,38 @@ func ManagerSpec(c gospec.Context) {
 			c.Expect(len, Equals, 0)
 		})
 
+		c.Specify("drain queue completely on exit", func() {
+			sentinel, _ := NewMsg("{\"foo\":\"bar2\",\"args\":\"sentinel\"}")
+
+			drained := false
+
+			slowJob := (func(message *Msg) {
+				if message.ToJson() == sentinel.ToJson() {
+					drained = true
+				} else {
+					processed <- message.Args()
+				}
+
+				time.Sleep(1 * time.Second)
+			})
+			manager := newManager("manager1", slowJob, 10)
+
+			for i := 0; i < 9; i++ {
+				conn.Do("lpush", "prod:queue:manager1", message.ToJson())
+			}
+			conn.Do("lpush", "prod:queue:manager1", sentinel.ToJson())
+
+			manager.start()
+			for i := 0; i < 9; i++ {
+				<-processed
+			}
+			manager.quit()
+
+			len, _ := redis.Int(conn.Do("llen", "prod:queue:manager1"))
+			c.Expect(len, Equals, 0)
+			c.Expect(drained, Equals, true)
+		})
+
 		c.Specify("per-manager middlwares are called separately, global middleware is called in each manager", func() {
 			mid1 := customMid{Base: "1"}
 			mid2 := customMid{Base: "2"}