prevents FixedFifo.Enqueue to gets blocked trying to send the item over an invalid waitForNextElementChan channel

Author: enriquebris

fixed_fifo_queue.go

@@ -32,16 +32,30 @@ func (st *FixedFIFO) Enqueue(value interface{}) error {
 	// check if there is a listener waiting for the next element (this element)
 	select {
 	case listener := <-st.waitForNextElementChan:
-		// send the element through the listener's channel instead of enqueue it
-		listener <- value
-
-	default:
-		// enqueue the element following the "normal way"
+		// verify whether it is possible to notify the listener (it could be the listener is no longer
+		// available because the context expired: DequeueOrWaitForNextElementContext)
 		select {
-		case st.queue <- value:
-		default:
-			return NewQueueError(QueueErrorCodeFullCapacity, "FixedFIFO queue is at full capacity")
+			// 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:
+		// enqueue the element into the queue
+		return st.enqueueIntoQueue(value)
+	}
+
+	return nil
+}
+
+// enqueueIntoQueue enqueues the given item directly into the regular queue
+func (st *FixedFIFO) enqueueIntoQueue(value interface{}) error {
+	select {
+	case st.queue <- value:
+	default:
+		return NewQueueError(QueueErrorCodeFullCapacity, "FixedFIFO queue is at full capacity")
 	}
 
 	return nil

fixed_fifo_queue_test.go

@@ -78,6 +78,23 @@ func (suite *FixedFIFOTestSuite) TestEnqueueFullCapacitySingleGR() {
 	suite.Equalf(QueueErrorCodeFullCapacity, customError.Code(), "Expected code: '%v'", QueueErrorCodeFullCapacity)
 }
 
+func (suite *FixedFIFOTestSuite) TestEnqueueListenerToExpireSingleGR() {
+	var (
+		uselessChan = make(chan interface{})
+		value = "my-test-value"
+	)
+
+	// let Enqueue knows there is a channel to send the next item instead of enqueueing it into the queue
+	suite.fifo.waitForNextElementChan <- uselessChan
+
+	// enqueues an item having an waiting channel but without a valid listener, so the item should be enqueued into the queue
+	suite.fifo.Enqueue(value)
+	// dequeues the item directly from the queue
+	dequeuedValue, _ := suite.fifo.Dequeue()
+
+	suite.Equal(value, dequeuedValue)
+}
+
 // TestEnqueueLenMultipleGR enqueues elements concurrently
 //
 // Detailed steps: