diff --git a/README.md b/README.md index 7998f7b..f798ba2 100644 --- a/README.md +++ b/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. diff --git a/go.mod b/go.mod index 77adead..f364f8f 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,3 @@ module github.com/vertoforce/streamregex -go 1.12 +go 1.24 diff --git a/steamregexindex.go b/steamregexindex.go deleted file mode 100644 index 7d40df1..0000000 --- a/steamregexindex.go +++ /dev/null @@ -1,63 +0,0 @@ -package streamregex - -import ( - "bufio" - "context" - "io" - "regexp" -) - -// SplitRegexIndex takes a regex and a channel for sending the locations at wich 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 := make([]int, 2) - absLoc[0] = loc[0] + byteCount - absLoc[1] = 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 returns a channel of matched []byte from the reader and a channel of the locations at wich the matches where found -// This function will allocate maxMatchLength*2 bytes of memory -func FindReaderIndex(ctx context.Context, r *regexp.Regexp, maxMatchLength int, reader io.Reader) (chan string, chan []int) { - allMatches := make(chan string) - allIndexes := make(chan []int, 1) // Avoid deadlock because if location is not read it blocks SplitRegexIndex from returning the match - - buf := make([]byte, maxMatchLength*2) - - go func() { - defer close(allMatches) - - scanner := bufio.NewScanner(reader) - scanner.Buffer(buf, maxMatchLength) - scanner.Split(SplitRegexIndex(r, maxMatchLength, allIndexes)) - for scanner.Scan() { - select { - case <-ctx.Done(): - return - case allMatches <- scanner.Text(): - } - } - }() - - return allMatches, allIndexes -} diff --git a/streamregex.go b/streamregex.go index f8c0b50..6fca6af 100644 --- a/streamregex.go +++ b/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() } diff --git a/streamregex_test.go b/streamregex_test.go index 14d6198..edaad43 100644 --- a/streamregex_test.go +++ b/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++ diff --git a/streamregexdoc_test.go b/streamregexdoc_test.go index b09c76e..43daf3a 100644 --- a/streamregexdoc_test.go +++ b/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] } diff --git a/streamregexindex.go b/streamregexindex.go new file mode 100644 index 0000000..0cc6455 --- /dev/null +++ b/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() +} diff --git a/steamregexindex_test.go b/streamregexindex_test.go similarity index 67% rename from steamregexindex_test.go rename to streamregexindex_test.go index d5bcde7..5cd4171 100644 --- a/steamregexindex_test.go +++ b/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