feat: add unmarshal to callback function

README.md

@@ -12,11 +12,14 @@ type Record struct {
     Height float32   `flat:"-"` // ignored
 }
 
-ch := make(chan Record)
+...
+
+goflat.MarshalSliceToWriter[Record](ctx,inputCh,csvWriter,options)
 
 ...
 
-goflat.MarshalSliceToWriter[Record](ctx,ch,csvWriter,options)
+goflat.UnmarshalToChan[Record](ctx,csvReader,options,outputCh)
+
 ```
 
 Will result in:

helper_test.go

@@ -0,0 +1,41 @@
+package goflat_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+)
+
+func assertChannel[T any](t *testing.T, ch <-chan T, expected []T, cmpOpts ...cmp.Option) {
+	t.Helper()
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+	var got []T
+
+	go func() {
+		defer cancel()
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case v, ok := <-ch:
+				if !ok {
+					return
+				}
+
+				got = append(got, v)
+			}
+		}
+	}()
+
+	t.Cleanup(func() {
+		<-ctx.Done()
+
+		if diff := cmp.Diff(expected, got, cmpOpts...); diff != "" {
+			t.Errorf("(-expected,+got):\n%s", diff)
+		}
+	})
+}

marshal.go

@@ -53,7 +53,7 @@ func MarshalChannelToWriter[T any](ctx context.Context, inputCh <-chan T, writer
 
 		select {
 		case <-ctx.Done():
-			return ctx.Err() //nolint:wrapcheck // No need here.
+			return context.Cause(ctx) //nolint:wrapcheck // Fine here.
 		case value, channelHasValue = <-inputCh:
 		}
 

reflect.go

@@ -7,6 +7,25 @@ import (
 	"strings"
 )
 
+// Options is used to configure the marshalling and unmarshalling processes.
+type Options struct {
+	headersFromStruct bool
+	// ErrorIfTaglessField causes goflat to error out if any struct field is
+	// missing the `flat` tag.
+	ErrorIfTaglessField bool
+	// ErrorIfDuplicateHeaders causes goflat to error out if two struct fields
+	// share the same `flat` tag value.
+	ErrorIfDuplicateHeaders bool
+	// ErrorIfMissingHeaders causes goflat to error out at unmarshalling time if
+	// a header has no struct field with a corresponding `flat` tag.
+	ErrorIfMissingHeaders bool
+	// UnmarshalIgnoreEmpty causes the unmarshaller to skip any column which is
+	// an empty string. This is useful for instance if you have integer values
+	// and you are okay with empty string mapping to the zero value (0). For the
+	// same reason this will cause booleans to be false if the column is empty.
+	UnmarshalIgnoreEmpty bool
+}
+
 type structFactory[T any] struct {
 	structType   reflect.Type
 	pointer      bool

unmarshal.go

@@ -10,25 +10,6 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
-// Options is used to configure the marshalling and unmarshalling processes.
-type Options struct {
-	headersFromStruct bool
-	// ErrorIfTaglessField causes goflat to error out if any struct field is
-	// missing the `flat` tag.
-	ErrorIfTaglessField bool
-	// ErrorIfDuplicateHeaders causes goflat to error out if two struct fields
-	// share the same `flat` tag value.
-	ErrorIfDuplicateHeaders bool
-	// ErrorIfMissingHeaders causes goflat to error out at unmarshalling time if
-	// a header has no struct field with a corresponding `flat` tag.
-	ErrorIfMissingHeaders bool
-	// UnmarshalIgnoreEmpty causes the unmarshaller to skip any column which is
-	// an empty string. This is useful for instance if you have integer values
-	// and you are okay with empty string mapping to the zero value (0). For the
-	// same reason this will cause booleans to be false if the column is empty.
-	UnmarshalIgnoreEmpty bool
-}
-
 // Unmarshaller can be used to tell goflat to use custom logic to convert the
 // input string into the type itself.
 type Unmarshaller interface {
@@ -103,3 +84,31 @@ func UnmarshalToSlice[T any](ctx context.Context, reader *csv.Reader, opts Optio
 
 	return slice, nil
 }
+
+// UnmarshalToCallback unamrshals a CSV file invoking a callback function on
+// each row.
+func UnmarshalToCallback[T any](ctx context.Context, reader *csv.Reader, opts Options, callback func(T) error) error {
+	g, ctx := errgroup.WithContext(ctx) //nolint:varnamelen // Fine here.
+
+	ch := make(chan T) //nolint:varnamelen // Fine here.
+
+	g.Go(func() error {
+		for v := range ch {
+			if err := callback(v); err != nil {
+				return fmt.Errorf("callback: %w", err)
+			}
+		}
+
+		return nil
+	})
+
+	g.Go(func() error {
+		return UnmarshalToChannel(ctx, reader, opts, ch)
+	})
+
+	if err := g.Wait(); err != nil {
+		return fmt.Errorf("wait: %w", err)
+	}
+
+	return nil
+}

unmarshal_test.go

@@ -4,23 +4,24 @@ import (
 	"context"
 	"embed"
 	"testing"
-	"time"
 
 	"github.com/google/go-cmp/cmp"
 
 	"github.com/lzambarda/goflat"
 )
 
 func TestUnmarshal(t *testing.T) {
-	t.Run("error empty", testUnmarshalErrorEmpty)
+	t.Run("error", testUnmarshalError)
 	t.Run("success", testUnmarshalSuccess)
-	t.Run("success ignore empty", testUnmarshalSuccessIgnoreEmpty)
-	t.Run("success pointer", testUnmarshalSuccessPointer)
 }
 
 //go:embed testdata
 var testdata embed.FS
 
+func testUnmarshalError(t *testing.T) {
+	t.Run("empty", testUnmarshalErrorEmpty)
+}
+
 func testUnmarshalErrorEmpty(t *testing.T) {
 	file, err := testdata.Open("testdata/unmarshal/success empty.csv")
 	if err != nil {
@@ -57,6 +58,14 @@ func testUnmarshalErrorEmpty(t *testing.T) {
 }
 
 func testUnmarshalSuccess(t *testing.T) {
+	t.Run("full", testUnmarshalSuccessFull)
+	t.Run("ignore empty", testUnmarshalSuccessIgnoreEmpty)
+	t.Run("pointer", testUnmarshalSuccessPointer)
+	t.Run("slice", testUnmarshalSuccessSlice)
+	t.Run("callback", testUnmarshalSuccessCallback)
+}
+
+func testUnmarshalSuccessFull(t *testing.T) {
 	file, err := testdata.Open("testdata/unmarshal/success.csv")
 	if err != nil {
 		t.Fatalf("open test file: %v", err)
@@ -225,34 +234,122 @@ func testUnmarshalSuccessPointer(t *testing.T) {
 	}
 }
 
-func assertChannel[T any](t *testing.T, ch <-chan T, expected []T, cmpOpts ...cmp.Option) {
-	t.Helper()
+func testUnmarshalSuccessSlice(t *testing.T) {
+	file, err := testdata.Open("testdata/unmarshal/success.csv")
+	if err != nil {
+		t.Fatalf("open test file: %v", err)
+	}
+
+	type record struct {
+		FirstName string  `flat:"first_name"`
+		LastName  string  `flat:"last_name"`
+		Age       int     `flat:"age"`
+		Height    float32 `flat:"height"`
+	}
+
+	expected := []record{
+		{
+			FirstName: "Guybrush",
+			LastName:  "Threepwood",
+			Age:       28,
+			Height:    1.78,
+		},
+		{
+			FirstName: "Elaine",
+			LastName:  "Marley",
+			Age:       20,
+			Height:    1.6,
+		},
+		{
+			FirstName: "LeChuck",
+			LastName:  "",
+			Age:       100,
+			Height:    2.01,
+		},
+	}
+
+	ctx := context.Background()
+
+	csvReader, err := goflat.DetectReader(file)
+	if err != nil {
+		t.Fatalf("detect reader: %v", err)
+	}
+
+	options := goflat.Options{
+		ErrorIfTaglessField:     true,
+		ErrorIfDuplicateHeaders: true,
+		ErrorIfMissingHeaders:   true,
+	}
+
+	got, err := goflat.UnmarshalToSlice[record](ctx, csvReader, options)
+	if err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+
+	if diff := cmp.Diff(expected, got, cmp.AllowUnexported(record{})); diff != "" {
+		t.Errorf("(-expected,+got):\n%s", diff)
+	}
+}
+
+func testUnmarshalSuccessCallback(t *testing.T) {
+	file, err := testdata.Open("testdata/unmarshal/success.csv")
+	if err != nil {
+		t.Fatalf("open test file: %v", err)
+	}
+
+	type record struct {
+		FirstName string  `flat:"first_name"`
+		LastName  string  `flat:"last_name"`
+		Age       int     `flat:"age"`
+		Height    float32 `flat:"height"`
+	}
+
+	expected := []record{
+		{
+			FirstName: "Guybrush",
+			LastName:  "Threepwood",
+			Age:       28,
+			Height:    1.78,
+		},
+		{
+			FirstName: "Elaine",
+			LastName:  "Marley",
+			Age:       20,
+			Height:    1.6,
+		},
+		{
+			FirstName: "LeChuck",
+			LastName:  "",
+			Age:       100,
+			Height:    2.01,
+		},
+	}
 
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
-	var got []T
+	ctx := context.Background()
 
-	go func() {
-		defer cancel()
+	csvReader, err := goflat.DetectReader(file)
+	if err != nil {
+		t.Fatalf("detect reader: %v", err)
+	}
 
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case v, ok := <-ch:
-				if !ok {
-					return
-				}
+	options := goflat.Options{
+		ErrorIfTaglessField:     true,
+		ErrorIfDuplicateHeaders: true,
+		ErrorIfMissingHeaders:   true,
+	}
 
-				got = append(got, v)
-			}
-		}
-	}()
+	var got []record
 
-	t.Cleanup(func() {
-		<-ctx.Done()
+	err = goflat.UnmarshalToCallback(ctx, csvReader, options, func(r record) error {
+		got = append(got, r)
 
-		if diff := cmp.Diff(expected, got, cmpOpts...); diff != "" {
-			t.Errorf("(-expected,+got):\n%s", diff)
-		}
+		return nil
 	})
+	if err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+
+	if diff := cmp.Diff(expected, got, cmp.AllowUnexported(record{})); diff != "" {
+		t.Errorf("(-expected,+got):\n%s", diff)
+	}
 }