Merge pull request #28 from enriquebris/ff_enqueue_invalidWaitForNextElement

FixedFixo.Enqueue verifies the waitForNextElementchan

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:

readme.md

@@ -1,4 +1,4 @@
-[![go.dev reference](https://img.shields.io/badge/go.dev-reference-007d9c?logo=go&logoColor=white)](https://pkg.go.dev/mod/github.com/enriquebris/goconcurrentqueue) [![godoc reference](https://img.shields.io/badge/godoc-reference-blue.svg)](https://godoc.org/github.com/enriquebris/goconcurrentqueue) ![version](https://img.shields.io/badge/version-v0.6.0-yellowgreen.svg?style=flat "goconcurrentqueue v0.6.0") [![Build Status](https://api.travis-ci.org/enriquebris/goconcurrentqueue.svg?branch=master)](https://travis-ci.org/enriquebris/goconcurrentqueue) [![Go Report Card](https://goreportcard.com/badge/github.com/enriquebris/goconcurrentqueue)](https://goreportcard.com/report/github.com/enriquebris/goconcurrentqueue)  [![codecov](https://codecov.io/gh/enriquebris/goconcurrentqueue/branch/master/graph/badge.svg)](https://codecov.io/gh/enriquebris/goconcurrentqueue) [![CodeFactor](https://www.codefactor.io/repository/github/enriquebris/goconcurrentqueue/badge)](https://www.codefactor.io/repository/github/enriquebris/goconcurrentqueue) [![Mentioned in Awesome Go](https://awesome.re/mentioned-badge.svg)](https://github.com/avelino/awesome-go)
+[![go.dev reference](https://img.shields.io/badge/go.dev-reference-007d9c?logo=go&logoColor=white)](https://pkg.go.dev/mod/github.com/enriquebris/goconcurrentqueue) [![godoc reference](https://img.shields.io/badge/godoc-reference-blue.svg)](https://godoc.org/github.com/enriquebris/goconcurrentqueue) ![version](https://img.shields.io/badge/version-v0.6.1-yellowgreen.svg?style=flat "goconcurrentqueue v0.6.1") [![Build Status](https://api.travis-ci.org/enriquebris/goconcurrentqueue.svg?branch=master)](https://travis-ci.org/enriquebris/goconcurrentqueue) [![Go Report Card](https://goreportcard.com/badge/github.com/enriquebris/goconcurrentqueue)](https://goreportcard.com/report/github.com/enriquebris/goconcurrentqueue)  [![codecov](https://codecov.io/gh/enriquebris/goconcurrentqueue/branch/master/graph/badge.svg)](https://codecov.io/gh/enriquebris/goconcurrentqueue) [![CodeFactor](https://www.codefactor.io/repository/github/enriquebris/goconcurrentqueue/badge)](https://www.codefactor.io/repository/github/enriquebris/goconcurrentqueue) [![Mentioned in Awesome Go](https://awesome.re/mentioned-badge.svg)](https://github.com/avelino/awesome-go)
 
 # goconcurrentqueue - Concurrent safe queues
 The package goconcurrentqueue offers a public interface Queue with methods for a [queue](https://en.wikipedia.org/wiki/Queue_(abstract_data_type)).
@@ -242,6 +242,10 @@ func workWithQueue(queue goconcurrentqueue.Queue) error {
 
 ## History
 
+### v0.6.1
+
+- FixedFifo.Enqueue prevents to gets blocked trying to send the item over an invalid waitForNextElementChan channel
+
 ### v0.6.0
 
 - Added DequeueOrWaitForNextElementContext()