Fix memory queue not blocking on Next

memory/memory.go

@@ -1,6 +1,7 @@
 package memory
 
 import (
+"io"
 "sync"
 "time"
 
@@ -161,7 +162,7 @@ func (i *JobIter) next() (*queue.Job, error) {
 i.Lock()
 defer i.Unlock()
 if len(i.q.jobs) <= i.q.idx {
-return nil, nil
+return nil, io.EOF
 }
 
 j := i.q.jobs[i.q.idx]

memory/memory_test.go

@@ -20,7 +20,6 @@ type MemorySuite struct {
 
 func (s *MemorySuite) SetupSuite() {
 s.BrokerURI = "memory://"
-s.AdvWindowNotSupported = true
 }
 
 func (s *MemorySuite) TestIntegration() {

test/suite.go

@@ -27,9 +27,8 @@ type QueueSuite struct {
 suite.Suite
 r rand.Rand
 
-TxNotSupported bool
-AdvWindowNotSupported bool
-BrokerURI string
+TxNotSupported bool
+BrokerURI string
 
 Broker queue.Broker
 }
@@ -63,7 +62,7 @@ func (s *QueueSuite) TestConsume_empty() {
 assert.NoError(iter.Close())
 }
 
-func (s *QueueSuite) TestJobIter_Next_empty() {
+func (s *QueueSuite) TestJobIter_Next_closed() {
 assert := assert.New(s.T())
 
 qName := NewName()
@@ -81,6 +80,50 @@ func (s *QueueSuite) TestJobIter_Next_empty() {
 <-done
 }
 
+func (s *QueueSuite) TestJobIter_Next_empty() {
+assert := assert.New(s.T())
+
+qName := NewName()
+q, err := s.Broker.Queue(qName)
+assert.NoError(err)
+assert.NotNil(q)
+
+advertisedWindow := 1
+iter, err := q.Consume(advertisedWindow)
+assert.NoError(err)
+assert.NotNil(iter)
+
+nJobs := 0
+
+done := make(chan struct{})
+go func() {
+j, err := iter.Next()
+assert.NoError(err)
+assert.NotNil(j)
+
+nJobs += 1
+done <- struct{}{}
+}()
+
+time.Sleep(50 * time.Millisecond)
+
+assert.Equal(0, nJobs)
+
+j, err := queue.NewJob()
+assert.NoError(err)
+
+err = j.Encode(1)
+assert.NoError(err)
+
+err = q.Publish(j)
+assert.NoError(err)
+
+<-done
+
+assert.Equal(1, nJobs)
+assert.NoError(iter.Close())
+}
+
 func (s *QueueSuite) TestJob_Reject_no_requeue() {
 assert := assert.New(s.T())
 
@@ -110,17 +153,10 @@ func (s *QueueSuite) TestJob_Reject_no_requeue() {
 err = j.Reject(false)
 assert.NoError(err)
 
-if s.AdvWindowNotSupported {
-j, err := iter.Next()
-assert.Nil(j)
-assert.NoError(err)
-assert.NoError(iter.Close())
-} else {
-done := s.checkNextClosed(iter)
-time.Sleep(50 * time.Millisecond)
-assert.NoError(iter.Close())
-<-done
-}
+done := s.checkNextClosed(iter)
+time.Sleep(50 * time.Millisecond)
+assert.NoError(iter.Close())
+<-done
 }
 
 func (s *QueueSuite) TestJob_Reject_requeue() {
@@ -385,17 +421,11 @@ func (s *QueueSuite) TestTransaction_Error() {
 advertisedWindow := 1
 i, err := q.Consume(advertisedWindow)
 assert.NoError(err)
-if s.AdvWindowNotSupported {
-j, err := i.Next()
-assert.Nil(j)
-assert.NoError(err)
-assert.NoError(i.Close())
-} else {
-done := s.checkNextClosed(i)
-time.Sleep(50 * time.Millisecond)
-assert.NoError(i.Close())
-<-done
-}
+
+done := s.checkNextClosed(i)
+time.Sleep(50 * time.Millisecond)
+assert.NoError(i.Close())
+<-done
 }
 
 func (s *QueueSuite) TestTransaction() {