chann: support to consume all data after Close the channel.

chann.go

@@ -79,6 +79,7 @@ func Cap(n int) Opt {
 // one, and use Cap to configure the capacity of the channel.
 type Chann[T any] struct {
 	in, out chan T
+	backlog chan T
 	close   chan struct{}
 	cfg     *config
 	q       []T
@@ -129,6 +130,7 @@ func New[T any](opts ...Opt) *Chann[T] {
 	case unbounded:
 		ch.in = make(chan T, 16)
 		ch.out = make(chan T, 16)
+		ch.backlog = make(chan T, 1024)
 		go ch.unboundedProcessing()
 	}
 	return ch
@@ -208,14 +210,13 @@ func (ch *Chann[T]) unboundedTerminate() {
 	}
 	for len(ch.q) > 0 {
 		select {
+		// Note if receiver doesn't consume all data that has been sent to input
+		// channel, the `unboundedProcessing` goroutine will leak forever.
+		// Ref: https://github.com/golang-design/chann/issues/3
 		case ch.out <- ch.q[0]:
-		// The default branch exists because we need guarantee
-		// the loop can terminate. If there is a receiver, the
-		// first case will ways be selected. See #3.
-		default:
+			ch.q[0] = nilT // de-reference earlier to help GC
+			ch.q = ch.q[1:]
 		}
-		ch.q[0] = nilT // de-reference earlier to help GC
-		ch.q = ch.q[1:]
 	}
 	close(ch.out)
 	close(ch.close)

chann_test.go

@@ -207,7 +207,6 @@ func TestNonblockRecvRace(t *testing.T) {
 	for i := 0; i < n; i++ {
 		c := chann.New[int](chann.Cap(1))
 		c.In() <- 1
-		t.Log(i)
 		go func() {
 			select {
 			case <-c.Out():
@@ -433,11 +432,8 @@ func TestUnboundedChannClose(t *testing.T) {
 		}
 		ch.Close()
 
-		// Theoretically, this is not a dead loop. If the channel
-		// is closed, then this loop must terminate at somepoint.
-		// If not, we will meet timeout in the test.
-		for !chann.IsClosed(ch) {
-			t.Log("unbounded channel is still not entirely closed")
+		if chann.IsClosed(ch) {
+			t.Fatal("channel should not be closed if data is not consumed")
 		}
 	})
 
@@ -492,6 +488,31 @@ func TestUnboundedChannClose(t *testing.T) {
 			t.Fatalf("After close, not all elements are received, got %v, want %v", n, N)
 		}
 	})
+
+	// ref: https://github.com/golang-design/chann/issues/3#issuecomment-1150189421
+	t.Run("consume-data-after-close", func(t *testing.T) {
+		var wg sync.WaitGroup
+		ch := chann.New[int]()
+
+		wg.Add(1)
+		c := 0
+		go func() {
+			for range ch.Out() {
+				c++
+			}
+			wg.Done()
+		}()
+
+		for i := 0; i < 2048; i++ {
+			ch.In() <- 42
+		}
+		ch.Close()
+
+		wg.Wait()
+		if c != 2048 {
+			t.Fatalf("not all elements are received after channel being closed, want %v got %v", 2048, c)
+		}
+	})
 }
 
 func BenchmarkUnboundedChann(b *testing.B) {