issue-34

Author: enriquebris

fixed_fifo_queue.go

@@ -1,6 +1,8 @@
 package goconcurrentqueue
 
-import "context"
+import (
+	"context"
+)
 
 // Fixed capacity FIFO (First In First Out) concurrent queue
 type FixedFIFO struct {
@@ -35,11 +37,11 @@ func (st *FixedFIFO) Enqueue(value interface{}) error {
 		// verify whether it is possible to notify the listener (it could be the listener is no longer
 		// available because the context expired: DequeueOrWaitForNextElementContext)
 		select {
-			// sends the element through the listener's channel instead of enqueueing it
-			case listener <- value:
-			default:
-				// push the element into the queue instead of sending it through the listener's channel (which is not available at this moment)
-				return st.enqueueIntoQueue(value)
+		// sends the element through the listener's channel instead of enqueueing it
+		case listener <- value:
+		default:
+			// push the element into the queue instead of sending it through the listener's channel (which is not available at this moment)
+			return st.enqueueIntoQueue(value)
 		}
 
 	default:
@@ -114,6 +116,12 @@ func (st *FixedFIFO) DequeueOrWaitForNextElementContext(ctx context.Context) (in
 				return item, nil
 			case <-ctx.Done():
 				return nil, ctx.Err()
+			// try again to get the element from the regular queue (in case waitChan doesn't provide any item)
+			case value, ok := <-st.queue:
+				if ok {
+					return value, nil
+				}
+				return nil, NewQueueError(QueueErrorCodeInternalChannelClosed, "internal channel is closed")
 			}
 		default:
 			// too many watchers (waitForNextElementChanCapacity) enqueued waiting for next elements

fixed_fifo_queue_test.go

@@ -81,7 +81,7 @@ func (suite *FixedFIFOTestSuite) TestEnqueueFullCapacitySingleGR() {
 func (suite *FixedFIFOTestSuite) TestEnqueueListenerToExpireSingleGR() {
 	var (
 		uselessChan = make(chan interface{})
-		value = "my-test-value"
+		value       = "my-test-value"
 	)
 
 	// let Enqueue knows there is a channel to send the next item instead of enqueueing it into the queue
@@ -98,6 +98,7 @@ func (suite *FixedFIFOTestSuite) TestEnqueueListenerToExpireSingleGR() {
 // TestEnqueueLenMultipleGR enqueues elements concurrently
 //
 // Detailed steps:
+//
 //	1 - Enqueue totalGRs concurrently (from totalGRs different GRs)
 //	2 - Verifies the len, it should be equal to totalGRs
 //	3 - Verifies that all elements from 0 to totalGRs were enqueued
@@ -269,6 +270,7 @@ func (suite *FixedFIFOTestSuite) TestDequeueClosedChannelSingleGR() {
 // TestDequeueMultipleGRs dequeues elements concurrently
 //
 // Detailed steps:
+//
 //	1 - Enqueues totalElementsToEnqueue consecutive integers
 //	2 - Dequeues totalElementsToDequeue concurrently from totalElementsToDequeue GRs
 //	3 - Verifies the final len, should be equal to totalElementsToEnqueue - totalElementsToDequeue
@@ -376,6 +378,39 @@ func (suite *FixedFIFOTestSuite) TestDequeueOrWaitForNextElementWithEmptyQueue()
 	}
 }
 
+// calling DequeueOrWaitForNextElement with empty queue, then adding an item directly into queue's internal channel
+func (suite *FixedFIFOTestSuite) TestDequeueOrWaitForNextElementWithStuckWaitChan() {
+	var (
+		dummyValue = "dummyValue"
+		doneChan   = make(chan struct{})
+	)
+
+	// consumer
+	go func(queue *FixedFIFO, expectedValue interface{}, done chan struct{}) {
+		item, err := queue.DequeueOrWaitForNextElement()
+		suite.NoError(err)
+		suite.Equal(expectedValue, item)
+
+		done <- struct{}{}
+	}(suite.fifo, dummyValue, doneChan)
+
+	// a second should be enough for the consumer to start consuming ...
+	time.Sleep(time.Second)
+
+	// add an item (enqueue) directly into queue's internal channel
+	suite.fifo.queue <- dummyValue
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	select {
+	case <-doneChan:
+
+	case <-ctx.Done():
+		suite.Fail("too much time waiting ...")
+	}
+}
+
 // single GR calling DequeueOrWaitForNextElement (WaitForNextElementChanCapacity + 1) times, last one should return error
 func (suite *FixedFIFOTestSuite) TestDequeueOrWaitForNextElementWithFullWaitingChannel() {
 	// enqueue WaitForNextElementChanCapacity listeners to future enqueued elements
@@ -554,4 +589,4 @@ func (suite *FixedFIFOTestSuite) TestContextAlreadyCanceled() {
 	case <-time.After(2 * time.Second):
 		suite.Fail("DequeueOrWaitForNextElementContext did not return immediately after context was canceled")
 	}
-}
+}