Redesign readers to caller-owned channels + error return

README.md

@@ -5,10 +5,15 @@
 
 Go does not let you find the _matched data_ of a regex from a stream of data.  They let you find [the first position of a match](https://godoc.org/regexp#Regexp.FindReaderIndex), but not the data itself.
 
-Streamregex allows you to get a channel of the _matched data_ of a regex on a io.Reader stream.
+Streamregex sends you the _matched data_ of a regex on a io.Reader stream over a channel.
 
 ## Usage
 
+`FindReader` sends each match on a channel you own. It runs synchronously and returns an
+error (`nil` on clean EOF, otherwise a read or context error), so the caller owns error
+handling, concurrency, and the channel's lifecycle. Run it in a goroutine and close the
+channel once it returns:
+
 ```go
 // Create string
 data := `0123456789this is a stream    of data with lots of trailing information`
@@ -17,15 +22,42 @@ stream := strings.NewReader(data)
 // Build regex
 regex := regexp.MustCompile(`stream\s+of`)
 
-// Find matches
-matchedData := FindReader(context.Background(), regex, 100, stream)
-for match := range matchedData {
+// The caller owns the channel: run FindReader in a goroutine and close it when done.
+matches := make(chan string)
+go func() {
+    FindReader(context.Background(), regex, 100, stream, matches)
+    close(matches)
+}()
+
+for match := range matches {
     fmt.Println(match)
 }
 
 // Output: stream    of
 ```
 
+### With match locations
+
+`FindReaderIndex` additionally reports each match's `[start, end]` byte offsets on a second
+channel. The index is sent just before its match, so the caller MUST drain both channels;
+buffer the index channel by 1 (or receive the index before the match) to avoid a deadlock:
+
+```go
+matches := make(chan string)
+indexes := make(chan []int, 1) // buffered: index is sent just before its match
+go func() {
+    FindReaderIndex(context.Background(), regex, 100, stream, matches, indexes)
+    close(matches)
+    close(indexes)
+}()
+
+for match := range matches {
+    fmt.Println(match, <-indexes)
+}
+
+// Output: stream    of [20 32]
+```
+
 ## How it works
 
 We use a custom [SplitFunc](https://golang.org/pkg/bufio/#SplitFunc) to split the reader into each regex match.  Normally for a SplitFunc it will keep reading more and more data into the buffer until it finds a match. To avoid pulling all the reader data into memory, the function accepts a `maxMatchLength` if you know the maximum match length of a match.

go.mod

@@ -1,3 +1,3 @@
 module github.com/vertoforce/streamregex
 
-go 1.12
+go 1.24

streamregex.go

@@ -27,27 +27,30 @@ func SplitRegex(re *regexp.Regexp, maxMatchLength int) bufio.SplitFunc {
 	}
 }
 
-// FindReader return channel of matched []byte from reader.
-// This function will allocate maxMatchLength*2 bytes of memory
-func FindReader(ctx context.Context, r *regexp.Regexp, maxMatchLength int, reader io.Reader) chan string {
-	allMatches := make(chan string)
-
+// FindReader scans reader for matches of re and sends each match on the caller-owned
+// matches channel. It runs synchronously (no internal goroutine) and returns when the
+// reader is exhausted, ctx is canceled, or a read error occurs.
+//
+// The caller owns matches: FindReader never closes it. Typically you launch FindReader
+// in a goroutine, then close matches once it returns and range over it from another
+// goroutine (see ExampleFindReader).
+//
+// If ctx is canceled, FindReader returns ctx.Err(). Otherwise it returns any error from
+// the underlying reader (a nil return means a clean EOF). This function will allocate
+// maxMatchLength*2 bytes of memory.
+func FindReader(ctx context.Context, re *regexp.Regexp, maxMatchLength int, reader io.Reader, matches chan<- string) error {
 	buf := make([]byte, maxMatchLength*2)
 
-	go func() {
-		defer close(allMatches)
-
-		scanner := bufio.NewScanner(reader)
-		scanner.Buffer(buf, maxMatchLength)
-		scanner.Split(SplitRegex(r, maxMatchLength))
-		for scanner.Scan() {
-			select {
-			case <-ctx.Done():
-				return
-			case allMatches <- scanner.Text():
-			}
+	scanner := bufio.NewScanner(reader)
+	scanner.Buffer(buf, maxMatchLength)
+	scanner.Split(SplitRegex(re, maxMatchLength))
+	for scanner.Scan() {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case matches <- scanner.Text():
 		}
-	}()
+	}
 
-	return allMatches
+	return scanner.Err()
 }

streamregex_test.go

@@ -39,8 +39,16 @@ func TestFindReader(t *testing.T) {
 	}
 
 	for _, test := range tests {
-		// Find matches
-		matchedData := FindReader(context.Background(), test.regex, test.maxMatchLen, strings.NewReader(test.input))
+		// Find matches. The caller owns the channel: run FindReader in a goroutine
+		// and close the channel once it returns.
+		matchedData := make(chan string)
+		go func() {
+			if err := FindReader(context.Background(), test.regex, test.maxMatchLen, strings.NewReader(test.input), matchedData); err != nil {
+				t.Errorf("FindReader returned error: %v", err)
+			}
+			close(matchedData)
+		}()
+
 		matches := 0
 		for match := range matchedData {
 			matches++

streamregexdoc_test.go

@@ -15,11 +15,16 @@ func ExampleFindReader() {
 	// Build regex
 	regex := regexp.MustCompile(`stream\s+of`)
 
+	// The caller owns the channel. Run FindReader in a goroutine and close the channel
+	// once it returns so the range below terminates.
+	matches := make(chan string)
+	go func() {
+		FindReader(context.Background(), regex, 100, stream, matches)
+		close(matches)
+	}()
+
 	// Find matches
-	matchedData := FindReader(context.Background(), regex, 100, stream)
-	matches := 0
-	for match := range matchedData {
-		matches++
+	for match := range matches {
 		fmt.Println(match)
 	}
 
@@ -34,14 +39,21 @@ func ExampleFindReaderIndex() {
 	// Build regex
 	regex := regexp.MustCompile(`stream\s+of`)
 
+	// The caller owns both channels. The index is sent just before its match, so the
+	// index channel is buffered by 1 to avoid a deadlock. Run FindReaderIndex in a
+	// goroutine and close both channels once it returns.
+	matches := make(chan string)
+	indexes := make(chan []int, 1)
+	go func() {
+		FindReaderIndex(context.Background(), regex, 100, stream, matches, indexes)
+		close(matches)
+		close(indexes)
+	}()
+
 	// Find matches and indexes
-	matchedData, matchedIndexes := FindReaderIndex(context.Background(), regex, 100, stream)
-	matches := 0
-	for match := range matchedData {
-		index := <-matchedIndexes
-		matches++
-		fmt.Println(match, index)
+	for match := range matches {
+		fmt.Println(match, <-indexes)
 	}
 
-	// Output: stream    of
+	// Output: stream    of [20 32]
 }

streamregexindex.go

@@ -0,0 +1,69 @@
+package streamregex
+
+import (
+	"bufio"
+	"context"
+	"io"
+	"regexp"
+)
+
+// SplitRegexIndex takes a regex and a channel for sending the locations at which the matches are found; and returns a split function that will find that regex in a byte slice
+func SplitRegexIndex(re *regexp.Regexp, maxMatchLength int, indexChannel chan<- []int) bufio.SplitFunc {
+	var byteCount int
+
+	return func(data []byte, atEOF bool) (advance int, token []byte, err error) {
+		if atEOF && len(data) == 0 {
+			return 0, nil, io.EOF
+		}
+
+		if loc := re.FindIndex(data); loc != nil {
+			absLoc := []int{loc[0] + byteCount, loc[1] + byteCount}
+			indexChannel <- absLoc
+			byteCount = byteCount + loc[1]
+			return loc[1], data[loc[0]:loc[1]], nil
+		}
+		if atEOF {
+			return 0, nil, io.EOF
+		}
+		if len(data) >= maxMatchLength {
+			var advance = len(data) - maxMatchLength
+			byteCount = byteCount + advance
+			return advance, nil, nil
+		}
+		return 0, nil, nil
+	}
+}
+
+// FindReaderIndex scans reader for matches of re. It sends each match on the caller-owned
+// matches channel and its location ([start, end] absolute byte offsets) on the caller-owned
+// indexes channel. It runs synchronously (no internal goroutine) and returns when the reader
+// is exhausted, ctx is canceled, or a read error occurs.
+//
+// The caller owns both channels: FindReaderIndex never closes them. Typically you launch
+// FindReaderIndex in a goroutine, then close both channels once it returns and range over
+// them from another goroutine (see ExampleFindReaderIndex).
+//
+// The caller MUST drain BOTH channels. Each index is sent immediately before its
+// corresponding match, so give the index channel a buffer of 1 (or receive the index
+// before the match) — otherwise the scanner blocks sending the index while the caller
+// waits for the match and the whole thing deadlocks.
+//
+// If ctx is canceled, FindReaderIndex returns ctx.Err(). Otherwise it returns any error
+// from the underlying reader (a nil return means a clean EOF). This function will allocate
+// maxMatchLength*2 bytes of memory.
+func FindReaderIndex(ctx context.Context, re *regexp.Regexp, maxMatchLength int, reader io.Reader, matches chan<- string, indexes chan<- []int) error {
+	buf := make([]byte, maxMatchLength*2)
+
+	scanner := bufio.NewScanner(reader)
+	scanner.Buffer(buf, maxMatchLength)
+	scanner.Split(SplitRegexIndex(re, maxMatchLength, indexes))
+	for scanner.Scan() {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case matches <- scanner.Text():
+		}
+	}
+
+	return scanner.Err()
+}

steamregexindex_test.go → streamregexindex_test.go

@@ -43,8 +43,19 @@ func TestFindReaderIndex(t *testing.T) {
 	}
 
 	for _, test := range tests {
-		// Find matches
-		matchedData, locationsData := FindReaderIndex(context.Background(), test.regex, test.maxMatchLen, strings.NewReader(test.input))
+		// Find matches. The caller owns both channels. The index is sent just before its
+		// match, so the index channel is buffered by 1 to avoid a deadlock. Run
+		// FindReaderIndex in a goroutine and close both channels once it returns.
+		matchedData := make(chan string)
+		locationsData := make(chan []int, 1)
+		go func() {
+			if err := FindReaderIndex(context.Background(), test.regex, test.maxMatchLen, strings.NewReader(test.input), matchedData, locationsData); err != nil {
+				t.Errorf("FindReaderIndex returned error: %v", err)
+			}
+			close(matchedData)
+			close(locationsData)
+		}()
+
 		matches := 0
 		for match := range matchedData {
 			location := <-locationsData