Simplified New* functions and added close to channel piper

README.md

@@ -35,7 +35,7 @@ func runMetricsEndpoint() {
 }
 
 func main() {
-	pv := peevee.NewPeeVee("myWorkChannel", WithPromMetrics[int]())
+	pv := peevee.New("myWorkChannel", WithPromMetrics[int]())
 
 	// mimic an asynchronous channel writer
 	go func() {

pkg/pv/channelpiper.go

@@ -19,7 +19,7 @@ type CallbackChannelPiper[T any] struct {
 	callback func(T)
 }
 
-// Pipe pipe items between channels
+// Pipe pipes items between the read and write channels
 func (ccp *CallbackChannelPiper[T]) Pipe() {
 	for item := range ccp.WriteChan {
 		if ccp.callback != nil {
@@ -28,4 +28,6 @@ func (ccp *CallbackChannelPiper[T]) Pipe() {
 
 		ccp.ReadChan <- item
 	}
+
+	close(ccp.ReadChan)
 }

pkg/pv/channelpiper_test.go

@@ -6,7 +6,7 @@ import (
 )
 
 func TestCallbackChannelPiper(t *testing.T) {
-	pv := NewPeeVee("myintpeevee", WithCallback(func(i int) {}))
+	pv := New("myintpeevee", WithCallback(func(i int) {}))
 
 	go func() {
 		pv.GetWritableChan() <- 1234
@@ -28,7 +28,7 @@ func TestCallbackChannelPiper(t *testing.T) {
 }
 
 func TestDefaultChannelPiper(t *testing.T) {
-	pv := NewPeeVee[int]("myintpeevee")
+	pv := New[int]("myintpeevee")
 
 	go func() {
 		pv.GetWritableChan() <- 1234

pkg/pv/metrics_test.go

@@ -6,7 +6,7 @@ package peevee
 // )
 
 // func TestWithMetricsWriter(t *testing.T) {
-// 	NewPeeVee("myintpeevee", WithMetricsWriter[int](WithMetricsWriterConfig{
+// 	New("myintpeevee", WithMetricsWriter[int](WithMetricsWriterConfig{
 // 		Writer: os.Stdout,
 // 	}))
 // }

pkg/pv/prom_test.go

@@ -8,13 +8,13 @@ import (
 )
 
 func TestWithPromMetricsSimple(t *testing.T) {
-	NewPeeVee("myintpeevee", WithPromMetrics[int]())
+	New("myintpeevee", WithPromMetrics[int]())
 }
 
 func TestWithPromMetrics(t *testing.T) {
 	r := prometheus.NewRegistry()
 
-	pv := NewPeeVee("myintpeevee", WithPromMetrics[int](WithPromMetricsConfig{
+	pv := New("myintpeevee", WithPromMetrics[int](WithPromMetricsConfig{
 		UseRegisterer: r,
 	}))
 

pkg/pv/pv.go

@@ -25,8 +25,8 @@ func (pv *PeeVee[T]) GetWritableChan() chan<- T {
 	return pv.writeChan
 }
 
-//NewPeeVee Configures and returns a new PeeVee
-func NewPeeVee[T any](PVname string, ops ...PVOptions[T]) PeeVee[T] {
+//New Configures and returns a new PeeVee
+func New[T any](PVname string, ops ...PVOptions[T]) PeeVee[T] {
 	pv := PeeVee[T]{
 		Name:      PVname,
 		readChan:  make(chan T),

pkg/pv/pv_test.go

@@ -33,3 +33,44 @@ func TestNewPeeVeeReaderWrap(t *testing.T) {
 		}
 	}
 }
+
+func TestNewPeeVeeReaderWrapWithClose(t *testing.T) {
+	// this is a channel that gets returned
+	// from an SDK or a third party lib for example
+	channelYouDontControl := make(chan bool)
+
+	go func() {
+		// this would be the third party code sending items
+		// to the channel
+		channelYouDontControl <- true
+
+		// and this is the channel being closed by the same third party code
+		close(channelYouDontControl)
+	}()
+
+	pv := NewReaderWrap(
+		"boolwrap",
+		channelYouDontControl,
+	)
+
+	for {
+		select {
+		case i, more := <-pv.GetReadableChan():
+			// now, we purposedly don't return
+			// and instead stay in the loop waiting for the channel
+			// to be closed
+			if !more {
+				return
+			}
+
+			if i != true {
+				t.Error("got wrong item from channel:", i)
+				t.FailNow()
+			}
+
+		case <-time.After(time.Second * 5):
+			t.Error("timeout reading from channel")
+			t.FailNow()
+		}
+	}
+}