Ensure full redirection occurs

Signed-off-by: Andrew Thornton <[email protected]>

Author: zeripath

modules/queue/queue_disk_channel.go

@@ -313,14 +313,13 @@ func (q *PersistableChannelQueue) Shutdown() {
 	q.channelQueue.Wait()
 	q.internal.(*LevelQueue).Wait()
 	// Redirect all remaining data in the chan to the internal channel
-	go func() {
-		log.Trace("PersistableChannelQueue: %s Redirecting remaining data", q.delayedStarter.name)
-		for data := range q.channelQueue.dataChan {
-			_ = q.internal.Push(data)
-			atomic.AddInt64(&q.channelQueue.numInQueue, -1)
-		}
-		log.Trace("PersistableChannelQueue: %s Done Redirecting remaining data", q.delayedStarter.name)
-	}()
+	log.Trace("PersistableChannelQueue: %s Redirecting remaining data", q.delayedStarter.name)
+	close(q.channelQueue.dataChan)
+	for data := range q.channelQueue.dataChan {
+		_ = q.internal.Push(data)
+		atomic.AddInt64(&q.channelQueue.numInQueue, -1)
+	}
+	log.Trace("PersistableChannelQueue: %s Done Redirecting remaining data", q.delayedStarter.name)
 
 	log.Debug("PersistableChannelQueue: %s Shutdown", q.delayedStarter.name)
 }

modules/queue/queue_disk_channel_test.go

@@ -497,6 +497,7 @@ func TestPersistableChannelQueue_Pause(t *testing.T) {
 		callback()
 	}
 	lock.Lock()
+	log.Info("Finally terminating")
 	callbacks = make([]func(), len(queueTerminate))
 	copy(callbacks, queueTerminate)
 	lock.Unlock()