Repository: moov-io/ach
Pull request #1304 (2023-10-16 21:27:02)

feat: add a File iterator
This iterator will help us process files entry-by-entry which should have a large reduction of memory usage.

adamdecaf: cc @wadearnold 

Patch:
diff --git a/batchHeader.go b/batchHeader.go
index 602bbfd3..1defe63f 100644
--- a/batchHeader.go
+++ b/batchHeader.go
@@ -213,6 +213,10 @@ func (bh *BatchHeader) String() string {
 // Equal returns true only if two BatchHeaders are equal.
 // Equality is determined by the Nacha defined fields of each record.
 func (bh *BatchHeader) Equal(other *BatchHeader) bool {
+	if bh == nil || other == nil {
+		return false
+	}
+
 	if bh.ServiceClassCode != other.ServiceClassCode {
 		return false
 	}
diff --git a/iterator.go b/iterator.go
new file mode 100644
index 00000000..f9519ce3
--- /dev/null
+++ b/iterator.go
@@ -0,0 +1,162 @@
+// Licensed to The Moov Authors under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. The Moov Authors licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package ach
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"strings"
+	"unicode"
+)
+
+// Iterator is a data structure for processing an ACH file one entry at a time.
+type Iterator struct {
+	reader     *Reader
+	scanner    *bufio.Scanner
+	cachedLine string
+}
+
+// NewIterator returns an Iterator
+func NewIterator(r io.Reader) *Iterator {
+	reader := NewReader(strings.NewReader("")) // the input is not used, we rely on .readLine()
+	reader.skipBatchAccumulation = true        // don't call .AddBatch(..)
+
+	out := &Iterator{
+		reader:  reader,
+		scanner: bufio.NewScanner(r),
+	}
+	return out
+}
+
+func (i *Iterator) SetValidation(opts *ValidateOpts) {
+	if i.reader != nil {
+		i.reader.SetValidation(opts)
+	}
+}
+
+// GetHeader will return the FileHeader once encountered by the iterator.
+// Call NextEntry() at least once to populate the header.
+func (i *Iterator) GetHeader() *FileHeader {
+	if i.reader != nil {
+		return &i.reader.File.Header
+	}
+	return nil
+}
+
+// GetControl will return the FileControl once encountered by the iterator.
+// Call NextEntry() at least once to populate the control.
+func (i *Iterator) GetControl() *FileControl {
+	if i.reader != nil {
+		return &i.reader.File.Control
+	}
+	return nil
+}
+
+// NextEntry will return the next available EntryDetail record and the BatchHeader the entry belongs to.
+//
+// IAT entries are not currently supported.
+func (i *Iterator) NextEntry() (*BatchHeader, *EntryDetail, error) {
+	// Clear the reader's File
+	defer func() {
+		i.reader.File = File{}
+	}()
+
+	// Read the file one line at a time
+	line := i.cachedLine
+	if line != "" {
+		i.cachedLine = "" // clear cache
+	} else {
+		// Consume lines until we reach a non-empty line
+		for i.scanner.Scan() {
+			line = i.scanner.Text()
+			i.reader.lineNum++
+			if allSpaces(line) {
+				continue
+			}
+			if line != "" {
+				break
+			}
+		}
+		// If we've exhausted all lines in the reader then quit
+		if line == "" || allSpaces(line) {
+			return nil, nil, nil
+		}
+	}
+
+	if err := i.reader.readLine(line); err != nil {
+		return nil, nil, fmt.Errorf("reading line %d failed: %w", i.reader.lineNum, err)
+	}
+
+	if i.reader.currentBatch != nil {
+		bh := i.reader.currentBatch.GetHeader()
+		entries := i.reader.currentBatch.GetEntries()
+		if len(entries) > 0 {
+			// Find the next entry to return and consume the file until we run out of
+			// addenda records or encounter a batch control/header record.
+			returnableEntry := entries[len(entries)-1]
+
+			// Read lines so long as we encounter an addenda or batch control record
+			for {
+				if i.scanner.Scan() {
+					foundLine := i.scanner.Text()
+					i.reader.lineNum++
+					if foundLine == "" {
+						break
+					}
+					switch {
+					case strings.HasPrefix(foundLine, entryAddendaPos):
+						i.reader.line = foundLine
+						if err := i.reader.parseEDAddenda(); err != nil {
+							return nil, nil, fmt.Errorf("reading addenda on line %d failed: %w", i.reader.lineNum, err)
+						}
+
+						entries := i.reader.currentBatch.GetEntries()
+						ed := entries[len(entries)-1]
+
+						return bh, ed, nil
+
+					case strings.HasPrefix(foundLine, batchControlPos):
+						// Do nothing with the Batch Control record
+						return bh, returnableEntry, nil
+
+					default:
+						i.cachedLine = foundLine
+						return bh, returnableEntry, nil
+					}
+				}
+			}
+
+			return bh, returnableEntry, nil
+		} else {
+			// We processed the BatchHeader, but need to find an Entry Detail record
+			return i.NextEntry()
+		}
+	}
+
+	return i.NextEntry()
+}
+
+func allSpaces(input string) bool {
+	for _, r := range input {
+		if !unicode.IsSpace(r) {
+			return false
+		}
+	}
+	return len(input) > 0
+}
diff --git a/reader.go b/reader.go
index 9ac05077..6c26b7c9 100644
--- a/reader.go
+++ b/reader.go
@@ -68,6 +68,9 @@ type Reader struct {
 
 	// errors holds each error encountered when attempting to parse the file
 	errors base.ErrorList
+
+	// skipBatchAccumulation is a flag to skip .AddBatch
+	skipBatchAccumulation bool
 }
 
 // error returns a new ParseError based on err
@@ -183,51 +186,18 @@ func (r *Reader) Read() (File, error) {
 			r.errors.Add(ErrFileTooLong)
 			return r.File, r.errors
 		}
-
-		lineLength := utf8.RuneCountInString(line)
-
-		switch {
-		case r.lineNum == 1 && lineLength > RecordLength:
-			extraChars := lineLength % RecordLength
-			if extraChars != 0 {
-				err := fmt.Errorf(
-					"%d extra character(s) in ACH file: must be %d but found %d",
-					extraChars,
-					lineLength-extraChars,
-					lineLength,
-				)
-				r.errors.Add(r.parseError(err))
-			} else if err := r.processFixedWidthFile(&line); err != nil {
-				r.errors.Add(err)
-			}
-		case lineLength != RecordLength:
-			if lineLength > RecordLength {
-				line = trimSpacesFromLongLine(line)
-			}
-			// right-pad the line with spaces
-			line, err := rightPadShortLine(line)
-			if err != nil {
-				r.errors.Add(r.parseError(err))
-			}
-			r.line = line
-			// parse the line
-			if err := r.parseLine(); err != nil {
-				r.errors.Add(r.parseError(NewRecordWrongLengthErr(lineLength)))
-				r.errors.Add(err)
-			}
-
-		default:
-			r.line = line
-			if err := r.parseLine(); err != nil {
-				r.errors.Add(err)
-			}
+		err := r.readLine(line)
+		if err != nil {
+			r.errors.Add(err)
 		}
 	}
 
 	// Add a lingering Batch to the file if there was no BatchControl record.
 	// This is common when files just contain a BatchHeader and EntryDetail records.
 	if r.currentBatch != nil {
-		r.File.AddBatch(r.currentBatch)
+		if !r.skipBatchAccumulation {
+			r.File.AddBatch(r.currentBatch)
+		}
 		r.currentBatch = nil
 	}
 
@@ -257,6 +227,47 @@ func (r *Reader) Read() (File, error) {
 	return r.File, r.errors
 }
 
+func (r *Reader) readLine(line string) error {
+	lineLength := utf8.RuneCountInString(line)
+	switch {
+	case r.lineNum == 1 && lineLength > RecordLength:
+		extraChars := lineLength % RecordLength
+		if extraChars != 0 {
+			err := fmt.Errorf(
+				"%d extra character(s) in ACH file: must be %d but found %d",
+				extraChars,
+				lineLength-extraChars,
+				lineLength,
+			)
+			return r.parseError(err)
+		} else if err := r.processFixedWidthFile(&line); err != nil {
+			return err
+		}
+	case lineLength != RecordLength:
+		if lineLength > RecordLength {
+			line = trimSpacesFromLongLine(line)
+		}
+		// right-pad the line with spaces
+		line, err := rightPadShortLine(line)
+		if err != nil {
+			return r.parseError(err)
+		}
+		r.line = line
+		// parse the line
+		if err := r.parseLine(); err != nil {
+			r.errors.Add(r.parseError(NewRecordWrongLengthErr(lineLength)))
+			return err
+		}
+
+	default:
+		r.line = line
+		if err := r.parseLine(); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func trimSpacesFromLongLine(s string) string {
 	return strings.TrimSuffix(s[:94], " ")
 }
@@ -301,7 +312,9 @@ func (r *Reader) parseLine() error {
 			batch := r.currentBatch
 			r.currentBatch = nil
 			batch.SetValidation(r.File.validateOpts)
-			r.File.AddBatch(batch)
+			if !r.skipBatchAccumulation {
+				r.File.AddBatch(batch)
+			}
 		}
 		if err := r.parseBH(); err != nil {
 			return err
@@ -322,7 +335,9 @@ func (r *Reader) parseLine() error {
 			batch := r.currentBatch
 			r.currentBatch = nil
 			batch.SetValidation(r.File.validateOpts)
-			r.File.AddBatch(batch)
+			if !r.skipBatchAccumulation {
+				r.File.AddBatch(batch)
+			}
 			if err := maybeValidate(batch, r.File.validateOpts); err != nil {
 				r.recordName = "Batches"
 				return r.parseError(err)
@@ -331,7 +346,9 @@ func (r *Reader) parseLine() error {
 			batch := r.IATCurrentBatch
 			r.IATCurrentBatch = IATBatch{}
 			batch.SetValidation(r.File.validateOpts)
-			r.File.AddIATBatch(batch)
+			if !r.skipBatchAccumulation {
+				r.File.AddIATBatch(batch)
+			}
 			if err := maybeValidate(&batch, r.File.validateOpts); err != nil {
 				r.recordName = "Batches"
 				return r.parseError(err)


Test patch:
diff --git a/iterator_test.go b/iterator_test.go
new file mode 100644
index 00000000..45a6273a
--- /dev/null
+++ b/iterator_test.go
@@ -0,0 +1,192 @@
+// Licensed to The Moov Authors under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. The Moov Authors licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package ach
+
+import (
+	"bytes"
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/moov-io/base"
+	"github.com/stretchr/testify/require"
+)
+
+func TestIterator(t *testing.T) {
+	t.Run("valid PPD file", func(t *testing.T) {
+		where := filepath.Join("test", "testdata", "ppd-mixedDebitCredit.ach")
+
+		file := openFile(t, where, nil)
+		iter := iteratorFromFile(t, where, nil)
+
+		ensureFileEqualsIterator(t, file, iter)
+	})
+
+	t.Run("more valid files", func(t *testing.T) {
+		paths := []string{
+			filepath.Join("test", "testdata", "two-micro-deposits.ach"),
+			filepath.Join("test", "testdata", "web-debit.ach"),
+		}
+		for i := range paths {
+			t.Logf("checking %s", paths[i])
+
+			file := openFile(t, paths[i], nil)
+			iter := iteratorFromFile(t, paths[i], nil)
+			ensureFileEqualsIterator(t, file, iter)
+		}
+	})
+
+	t.Run("bh-ed-ad-bh-ed-ad-ed-ad", func(t *testing.T) {
+		paths := []string{
+			filepath.Join("test", "testdata", "bh-ed-ad-bh-ed-ad-ed-ad.ach"),
+		}
+		for i := range paths {
+			t.Logf("checking %s", paths[i])
+
+			file := openFile(t, paths[i], nil)
+			iter := iteratorFromFile(t, paths[i], nil)
+			ensureFileEqualsIterator(t, file, iter)
+		}
+	})
+
+	t.Run("return examples", func(t *testing.T) {
+		paths := []string{
+			filepath.Join("test", "testdata", "return-WEB.ach"),
+			filepath.Join("test", "testdata", "return-no-batch-controls.ach"),
+			filepath.Join("test", "testdata", "return-no-file-header-control.ach"),
+		}
+
+		for i := range paths {
+			t.Logf("checking %s", paths[i])
+
+			file := openFile(t, paths[i], nil)
+			iter := iteratorFromFile(t, paths[i], nil)
+			ensureFileEqualsIterator(t, file, iter)
+		}
+	})
+
+	t.Run("custom return codes", func(t *testing.T) {
+		where := filepath.Join("test", "testdata", "return-PPD-custom-reason-code.ach")
+
+		opts := &ValidateOpts{
+			CustomReturnCodes: true,
+		}
+		file := openFile(t, where, opts)
+		iter := iteratorFromFile(t, where, opts)
+
+		ensureFileEqualsIterator(t, file, iter)
+	})
+
+	t.Run("whitespace in file", func(t *testing.T) {
+		firstBytes, err := os.ReadFile(filepath.Join(filepath.Join("test", "testdata", "bh-ed-ad-bh-ed-ad-ed-ad.ach")))
+		require.NoError(t, err)
+
+		secondBytes, err := os.ReadFile(filepath.Join(filepath.Join("test", "testdata", "return-no-file-header-control.ach")))
+		require.NoError(t, err)
+
+		checkBytes := func(t *testing.T, data []byte) {
+			t.Helper()
+
+			iter := NewIterator(bytes.NewReader(data))
+			var entries int
+			for {
+				bh, ed, err := iter.NextEntry()
+				if err != nil {
+					t.Fatal(err)
+				}
+				if bh != nil && ed != nil {
+					entries += 1
+				} else {
+					break
+				}
+			}
+			require.Equal(t, 4, entries)
+		}
+
+		t.Run("first then second", func(t *testing.T) {
+			data := append(firstBytes, append([]byte("\n  \n"), secondBytes...)...)
+			checkBytes(t, data)
+		})
+
+		t.Run("second then first", func(t *testing.T) {
+			data := append(secondBytes, append([]byte("\n  \n"), firstBytes...)...)
+			checkBytes(t, data)
+		})
+	})
+
+	t.Run("allSpaces", func(t *testing.T) {
+		require.True(t, allSpaces("\n"))
+		require.True(t, allSpaces("\n \r\n"))
+
+		require.False(t, allSpaces(""))
+		require.False(t, allSpaces("abc"))
+	})
+}
+
+func openFile(t *testing.T, where string, opts *ValidateOpts) *File {
+	t.Helper()
+
+	fd, err := os.Open(where)
+	require.NoError(t, err)
+
+	t.Cleanup(func() { fd.Close() })
+
+	r := NewReader(fd)
+	r.SetValidation(opts)
+
+	file, err := r.Read()
+	if err != nil {
+		if !base.Has(err, ErrFileHeader) {
+			require.NoError(t, err)
+			return nil
+		}
+	}
+	return &file
+}
+
+func iteratorFromFile(t *testing.T, where string, opts *ValidateOpts) *Iterator {
+	t.Helper()
+
+	fd, err := os.Open(where)
+	require.NoError(t, err)
+
+	t.Cleanup(func() { fd.Close() })
+
+	iter := NewIterator(fd)
+	iter.SetValidation(opts)
+	return iter
+}
+
+func ensureFileEqualsIterator(t *testing.T, file *File, iter *Iterator) {
+	t.Helper()
+
+	for i := range file.Batches {
+		bh := file.Batches[i].GetHeader()
+		entries := file.Batches[i].GetEntries()
+		for j := range entries {
+			ed := entries[j]
+
+			ibh, ied, err := iter.NextEntry()
+			require.NoError(t, err, "iterator batch[%d], entry[%d] error: %v", i, j, err)
+
+			require.True(t, bh.Equal(ibh), fmt.Sprintf("batch[%d] headers", i))
+			require.Equal(t, ed, ied, fmt.Sprintf("batch[%d] entry[%d] details", i, j))
+		}
+	}
+}