Chore: review changes; constructor func update

Author: vladimirvivien

sinks/chansink.go

@@ -9,39 +9,43 @@ import (
 	"github.com/vladimirvivien/automi/log"
 )
 
-// ChanSink sends streamed items to an output channel.
-type ChanSink[T any] struct {
-	output chan T
+// ChanSink represents a sink backed by a Go channel.
+type ChanSink[IN any, CHAN <-chan IN] struct {
+	chansink chan IN
 	input  <-chan any
 	logf   api.StreamLogFunc
 }
 
-// Chan is the constructor function which returns a new ChanSink.
-func Chan[T any](outputChan chan T) *ChanSink[T] {
-	return &ChanSink[T]{
-		output: outputChan,
+// BufferedChan returns a new ChanSink backed by a buffered channel.
+func BufferedChan[IN any, CHAN <-chan IN](bufferSize int) *ChanSink[IN,CHAN] {
+	return &ChanSink[IN,CHAN]{
+		chansink: make(chan IN, bufferSize),
 		logf:   log.NoLogFunc,
 	}
 }
 
-// SetInput sets the source for the sink.
-func (s *ChanSink[T]) SetInput(in <-chan any) {
+// Chan returns a new ChanSink backed by an unbuffered channel.
+func Chan[IN any, CHAN <-chan IN]() *ChanSink[IN,CHAN] {
+	return BufferedChan[IN,CHAN](0)
+}
+
+// SetInput sets the input for the sink.
+func (s *ChanSink[IN,CHAN]) SetInput(in <-chan any) {
 	s.input = in
 }
 
-// Get returns the output channel used by the sink.
-func (s *ChanSink[T]) Get() <-chan T {
-	return s.output
+// Get returns a receive-only channel used by the sink.
+func (s *ChanSink[IN,CHAN]) Get() CHAN {
+	return s.chansink
 }
 
 // SetLogFunc sets a logging func for the component.
-func (s *ChanSink[T]) SetLogFunc(f api.StreamLogFunc) {
+func (s *ChanSink[IN,CHAN]) SetLogFunc(f api.StreamLogFunc) {
 	s.logf = f
 }
 
 // Open starts the sink and returns and waits on the returned
-// channel for the sink to be done or an error to be received.
-func (s *ChanSink[T]) Open(ctx context.Context) <-chan error {
+func (s *ChanSink[IN,CHAN]) Open(ctx context.Context) <-chan error {
 	result := make(chan error)
 
 	s.logf(ctx, log.LogInfo(
@@ -56,7 +60,7 @@ func (s *ChanSink[T]) Open(ctx context.Context) <-chan error {
 				"Component closing",
 				slog.String("sink", "Chan"),
 			))
-			close(s.output) // Ensure output channel is closed
+			close(s.chansink) // Ensure output channel is closed
 		}()
 
 		for {
@@ -65,7 +69,7 @@ func (s *ChanSink[T]) Open(ctx context.Context) <-chan error {
 				if !opened {
 					return
 				}
-				data, ok := item.(T)
+				data, ok := item.(IN)
 				if !ok {
 					s.logf(ctx, log.LogDebug(
 						"Error: unexpected data type",
@@ -74,7 +78,7 @@ func (s *ChanSink[T]) Open(ctx context.Context) <-chan error {
 					))
 					continue
 				}
-				s.output <- data
+				s.chansink <- data
 			case <-ctx.Done():
 				return
 			}

sinks/chansink_test.go

@@ -8,10 +8,9 @@ import (
 )
 
 func TestChanSink_Open(t *testing.T) {
-	outputChan := make(chan string)
-	cs := Chan[string](outputChan)
+	cs := Chan[string]()
 
-	inputChan := make(chan interface{})
+	inputChan := make(chan any)
 	go func() {
 		inputChan <- "A"
 		inputChan <- "B"
@@ -23,7 +22,7 @@ func TestChanSink_Open(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
 	defer cancel()
 
-	sinkErr := cs.Open(ctx)
+	sink := cs.Open(ctx)
 
 	var receivedData []string
 	doneReceiving := make(chan struct{})
@@ -36,7 +35,7 @@ func TestChanSink_Open(t *testing.T) {
 	}()
 
 	select {
-	case err := <-sinkErr:
+	case err := <-sink:
 		if err != nil {
 			t.Fatalf("ChanSink.Open() returned an error: %v", err)
 		}
@@ -54,18 +53,11 @@ func TestChanSink_Open(t *testing.T) {
 	if !reflect.DeepEqual(expectedData, receivedData) {
 		t.Errorf("Received data did not match expected data. Got %v, expected %v", receivedData, expectedData)
 	}
-
-	// Check if the sink's output channel was closed by the sink (indirectly via the receiving goroutine)
-	// This is a bit tricky to test directly for a send-only channel from the sink's perspective.
-	// However, if the receiving goroutine exited, it implies the channel was closed.
-	// A more direct test might involve trying to send to cs.Get() after it's supposed to be closed,
-	// but that's not how this sink is designed to be used.
 }
 
 func TestChanSink_Open_ContextCancel(t *testing.T) {
-	outputChan := make(chan int)
-	cs := Chan[int](outputChan)
-	inputChan := make(chan interface{})
+	cs := Chan[int]()
+	inputChan := make(chan any)
 
 	cs.SetInput(inputChan)
 
@@ -98,9 +90,8 @@ func TestChanSink_Open_ContextCancel(t *testing.T) {
 }
 
 func TestChanSink_Open_WrongDataType(t *testing.T) {
-	outputChan := make(chan string)
-	cs := Chan[string](outputChan) // Expecting string
-	inputChan := make(chan interface{})
+	cs := Chan[string]() // Expecting string
+	inputChan := make(chan any)
 
 	go func() {
 		inputChan <- "CorrectType"

sinks/slice.go

@@ -39,8 +39,8 @@ func (s *SliceSink[IN, SLICE]) SetLogFunc(f api.StreamLogFunc) {
 	s.logf = f
 }
 
-// Open starts the collector and returns and waits on the returned
-// channel for the collector to be done or an error to be received.
+// Open opens the sink to start collecting items. It returns a channel
+// that will receive an error if the sink encounters an error during setup operation.
 func (s *SliceSink[IN, SLICE]) Open(ctx context.Context) <-chan error {
 	result := make(chan error)