feat: support Stop() in queue

Author: PalanQu

queue/queue.go

@@ -30,7 +30,8 @@ type Queue struct {
 	Threads int
 	storage Storage
 	wake    chan struct{}
-	mut     sync.Mutex // guards wake
+	mut     sync.Mutex // guards wake and running
+	running bool
 }
 
 // InMemoryQueueStorage is the default implementation of the Storage interface.
@@ -62,6 +63,7 @@ func New(threads int, s Storage) (*Queue, error) {
 	return &Queue{
 		Threads: threads,
 		storage: s,
+		running: true,
 	}, nil
 }
 
@@ -139,6 +141,13 @@ func (q *Queue) Run(c *colly.Collector) error {
 	return <-errc
 }
 
+func (q *Queue) Stop() error {
+	q.mut.Lock()
+	q.running = false
+	q.mut.Unlock()
+	return nil
+}
+
 func (q *Queue) loop(c *colly.Collector, requestc chan<- *colly.Request, complete <-chan struct{}, errc chan<- error) {
 	var active int
 	for {
@@ -147,7 +156,7 @@ func (q *Queue) loop(c *colly.Collector, requestc chan<- *colly.Request, complet
 			errc <- err
 			break
 		}
-		if size == 0 && active == 0 {
+		if size == 0 && active == 0 || !q.running {
 			// Terminate when
 			//   1. No active requests
 			//   2. Emtpy queue