Change the Transform coding pattern from Next() -> Read() to just Read() (#110)

Found this `Next()->Read()` pattern pretty silly and unnecessary. Also current `Next()`
has a useless `for {}` wrapping loop. Probably from long time ago, but it's not needed.

Author: jf-tech

README.md

@@ -80,10 +80,12 @@ Take a detailed look at samples here:
     if err != nil { ... }
     transform, err := schema.NewTransform("input-name", strings.NewReader("..."), &transformctx.Ctx{})
     if err != nil { ... }
-    if !transform.Next() { ... }  
-    b, err := transform.Read()
-    if err != nil { ... }
-    fmt.Println(string(b))
+    for {
+        b, err := transform.Read()
+        if err == io.EOF { break }
+        if err != nil { ... }
+        fmt.Println(string(b))
+    }
     ```
 - Output:
     ```
@@ -115,7 +117,7 @@ Take a detailed look at samples here:
 ## Requirements
 - Golang 1.14
 
-## Recent Feature Additions
+## Recent Major Feature Additions/Changes
 - Added EDI file format support in omniv2 handler.
 - Major restructure/refactoring
     - Upgrade omni schema version to `omni.2.1` due a number of incompatible schema changes:

cli/cmd/serverCmd.go

@@ -3,6 +3,7 @@ package cmd
 import (
 	"encoding/json"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"log"
 	"net/http"
@@ -136,8 +137,11 @@ func httpPostTransform(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	var records []string
-	for t.Next() {
+	for {
 		b, err := t.Read()
+		if err == io.EOF {
+			break
+		}
 		if err != nil {
 			writeBadRequest(w, fmt.Sprintf("bad request: transform failed. err: %s", err))
 			return

cli/cmd/transformCmd.go

@@ -77,30 +77,36 @@ func doTransform() error {
 		return err
 	}
 
-	fmt.Println("[")
-	doOne := func() error {
+	doOne := func() (string, error) {
 		b, err := transform.Read()
 		if err != nil {
-			return err
+			return "", err
 		}
-		fmt.Print(
-			strings.Join(
-				strs.NoErrMapSlice(
-					strings.Split(jsons.BPJ(string(b)), "\n"),
-					func(s string) string { return "\t" + s }),
-				"\n"))
+		return strings.Join(
+			strs.NoErrMapSlice(
+				strings.Split(jsons.BPJ(string(b)), "\n"),
+				func(s string) string { return "\t" + s }),
+			"\n"), nil
+	}
+
+	record, err := doOne()
+	if err == io.EOF {
+		fmt.Println("[]")
 		return nil
 	}
-	if transform.Next() {
-		if err := doOne(); err != nil {
-			return err
+	if err != nil {
+		return err
+	}
+	fmt.Printf("[\n%s", record)
+	for {
+		record, err = doOne()
+		if err == io.EOF {
+			break
 		}
-		for transform.Next() {
-			fmt.Print(",\n")
-			if err := doOne(); err != nil {
-				return err
-			}
+		if err != nil {
+			return err
 		}
+		fmt.Printf(",\n%s", record)
 	}
 	fmt.Println("\n]")
 	return nil

extensions/omniv21/samples/csv/csv_test.go

@@ -2,6 +2,7 @@ package csv
 
 import (
 	"bytes"
+	"io"
 	"io/ioutil"
 	"testing"
 
@@ -48,8 +49,11 @@ func Benchmark1_Weather_Data_CSV(b *testing.B) {
 		if err != nil {
 			b.FailNow()
 		}
-		for transform.Next() {
+		for {
 			_, err := transform.Read()
+			if err == io.EOF {
+				break
+			}
 			if err != nil {
 				b.FailNow()
 			}

extensions/omniv21/samples/customfileformats/jsonlog/sample_test.go

@@ -1,6 +1,7 @@
 package jsonlog
 
 import (
+	"io"
 	"os"
 	"path/filepath"
 	"strings"
@@ -76,8 +77,11 @@ func TestSample(t *testing.T) {
 	assert.NoError(t, err)
 
 	var records []string
-	for transform.Next() {
+	for {
 		recordBytes, err := transform.Read()
+		if err == io.EOF {
+			break
+		}
 		assert.NoError(t, err)
 		records = append(records, string(recordBytes))
 	}

extensions/omniv21/samples/customparse/sample_test.go

@@ -1,6 +1,7 @@
 package customparse
 
 import (
+	"io"
 	"os"
 	"path/filepath"
 	"strconv"
@@ -45,12 +46,15 @@ func TestSample(t *testing.T) {
 			},
 		})
 	assert.NoError(t, err)
-	tfm, err := schema.NewTransform(inputFileBaseName, inputFileReader, &transformctx.Ctx{})
+	transform, err := schema.NewTransform(inputFileBaseName, inputFileReader, &transformctx.Ctx{})
 	assert.NoError(t, err)
 
 	var records []string
-	for tfm.Next() {
-		recordBytes, err := tfm.Read()
+	for {
+		recordBytes, err := transform.Read()
+		if err == io.EOF {
+			break
+		}
 		assert.NoError(t, err)
 		records = append(records, string(recordBytes))
 	}

extensions/omniv21/samples/edi/edi_test.go

@@ -2,6 +2,7 @@ package edi
 
 import (
 	"bytes"
+	"io"
 	"io/ioutil"
 	"testing"
 
@@ -45,8 +46,11 @@ func Benchmark1_CanadaPost_EDI_214(b *testing.B) {
 		if err != nil {
 			b.FailNow()
 		}
-		for transform.Next() {
+		for {
 			_, err := transform.Read()
+			if err == io.EOF {
+				break
+			}
 			if err != nil {
 				b.FailNow()
 			}

extensions/omniv21/samples/json/json_test.go

@@ -2,6 +2,7 @@ package json
 
 import (
 	"bytes"
+	"io"
 	"io/ioutil"
 	"testing"
 
@@ -58,8 +59,11 @@ func Benchmark2_Multiple_Objects(b *testing.B) {
 		if err != nil {
 			b.FailNow()
 		}
-		for transform.Next() {
+		for {
 			_, err := transform.Read()
+			if err == io.EOF {
+				break
+			}
 			if err != nil {
 				b.FailNow()
 			}

extensions/omniv21/samples/testCommon.go

@@ -1,6 +1,7 @@
 package samples
 
 import (
+	"io"
 	"os"
 	"path/filepath"
 	"strings"
@@ -30,11 +31,13 @@ func SampleTestCommon(t *testing.T, schemaFile, inputFile string) string {
 	assert.NoError(t, err)
 
 	var records []string
-	for transform.Next() {
+	for {
 		recordBytes, err := transform.Read()
+		if err == io.EOF {
+			break
+		}
 		assert.NoError(t, err)
 		records = append(records, string(recordBytes))
 	}
-
 	return "[" + strings.Join(records, ",") + "]"
 }

transform.go

@@ -1,8 +1,6 @@
 package omniparser
 
 import (
-	"io"
-
 	"github.com/jf-tech/omniparser/errs"
 	"github.com/jf-tech/omniparser/schemahandler"
 )
@@ -11,56 +9,50 @@ import (
 // operation. An instance of a Transform must not be shared and reused among different
 // input streams. An instance of a Transform must not be used across multiple goroutines.
 type Transform interface {
-	// Next indicates whether the ingestion/transform is completed or not.
-	Next() bool
 	// Read returns a JSON byte slice representing one ingested and transformed record.
+	// io.EOF should be returned when input stream is completely consumed and future calls
+	// to Read should always return io.EOF.
+	// errs.ErrTransformFailed should be returned when a record ingestion and transformation
+	// failed and such failure isn't considered fatal. Future calls to Read will attempt
+	// new record ingestion and transformations.
+	// Any other error returned is considered fatal and future calls to Read will always
+	// return the same error.
+	// Note if returned error isn't nil, then returned []byte will be nil.
 	Read() ([]byte, error)
 }
 
 type transform struct {
-	ingester  schemahandler.Ingester
-	curErr    error
-	curRecord []byte
+	ingester schemahandler.Ingester
+	lastErr  error
 }
 
-// Next calls the underlying schema handler's ingester to do the ingestion and transformation, saves
-// the resulting record and/or error, and returns whether the entire operation is completed or not.
-func (o *transform) Next() bool {
-	// ErrTransformFailed is a generic wrapping error around all handlers' ingesters'
+// Read returns a JSON byte slice representing one ingested and transformed record.
+// io.EOF should be returned when input stream is completely consumed and future calls
+// to Read should always return io.EOF.
+// errs.ErrTransformFailed should be returned when a record ingestion and transformation
+// failed and such failure isn't considered fatal. Future calls to Read will attempt
+// new record ingestion and transformations.
+// Any other error returned is considered fatal and future calls to Read will always
+// return the same error.
+// Note if returned error isn't nil, then returned []byte will be nil.
+func (o *transform) Read() ([]byte, error) {
+	// errs.ErrTransformFailed is a generic wrapping error around all handlers' ingesters'
 	// **continuable** errors (so client side doesn't have to deal with myriad of different
-	// types of benign continuable errors. All other errors: non-continuable errors or ErrEOF
+	// types of benign continuable errors. All other errors: non-continuable errors or io.EOF
 	// should cause the operation to cease.
-	if o.curErr != nil && !errs.IsErrTransformFailed(o.curErr) {
-		return false
+	if o.lastErr != nil && !errs.IsErrTransformFailed(o.lastErr) {
+		return nil, o.lastErr
 	}
-
-	for {
-		record, err := o.ingester.Read()
-		switch {
-		case err == nil:
-			o.curRecord = record
-			o.curErr = nil
-			return true
-		case err == io.EOF:
-			o.curErr = err
-			o.curRecord = nil
-			// No more processing needed.
-			return false
-		default:
-			o.curErr = err
+	record, err := o.ingester.Read()
+	if err != nil {
+		if o.ingester.IsContinuableError(err) {
 			// If ingester error is continuable, wrap it into a standard generic ErrTransformFailed
 			// so caller has an easier time to deal with it. If fatal error, then leave it raw to the
 			// caller so they can decide what it is and how to proceed.
-			if o.ingester.IsContinuableError(err) {
-				o.curErr = errs.ErrTransformFailed(err.Error())
-			}
-			o.curRecord = nil
-			return true
+			err = errs.ErrTransformFailed(err.Error())
 		}
+		record = nil
 	}
-}
-
-// Read returns the current ingested and transformed record and/or the current error.
-func (o *transform) Read() ([]byte, error) {
-	return o.curRecord, o.curErr
+	o.lastErr = err
+	return record, err
 }