Merge branch 'master' into bobg/fromseq

go.mod

@@ -6,3 +6,5 @@ require (
 	github.com/mattn/go-sqlite3 v1.14.12
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
 )
+
+require github.com/google/go-cmp v0.6.0 // indirect

go.sum

@@ -1,3 +1,5 @@
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
+github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/mattn/go-sqlite3 v1.14.12 h1:TJ1bhYJPV44phC+IMu1u2K/i5RriLTPe+yc68XDJ1Z0=
 github.com/mattn/go-sqlite3 v1.14.12/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=

iter/gen.go

@@ -1,10 +1,5 @@
 package iter
 
-import (
-	"bufio"
-	"io"
-)
-
 // Gen produces an iterator of values obtained by repeatedly calling f.
 // If f returns an error,
 // iteration stops and the error is available via the iterator's Err method.
@@ -58,17 +53,3 @@ func Ints(start, delta int) Of[int] {
 func Repeat[T any](val T) Of[T] {
 	return Gen(func() (T, bool, error) { return val, true, nil })
 }
-
-// Lines produces an iterator over the lines of text in r.
-// This uses a bufio.Scanner
-// and is subject to its default line-length limit
-// (see https://pkg.go.dev/bufio#pkg-constants).
-func Lines(r io.Reader) Of[string] {
-	sc := bufio.NewScanner(r)
-	return Gen(func() (string, bool, error) {
-		if sc.Scan() {
-			return sc.Text(), true, nil
-		}
-		return "", false, sc.Err()
-	})
-}

iter/lines.go

@@ -0,0 +1,128 @@
+package iter
+
+import (
+	"bufio"
+	"context"
+	"errors"
+	"io"
+)
+
+// Lines produces an iterator over the lines of text in r.
+// This uses a bufio.Scanner
+// and is subject to its default line-length limit
+// (see https://pkg.go.dev/bufio#pkg-constants).
+func Lines(r io.Reader) Of[string] {
+	sc := bufio.NewScanner(r)
+	return Gen(func() (string, bool, error) {
+		if sc.Scan() {
+			return sc.Text(), true, nil
+		}
+		return "", false, sc.Err()
+	})
+}
+
+// LongLines produces an iterator of readers,
+// each delivering a single line of text from r.
+// Unlike [Lines],
+// this does not use a [bufio.Scanner]
+// and is not subject to its default line-length limit.
+// Each reader must be fully consumed before the next one is available.
+// If not consuming all readers from the iterator,
+// the caller should cancel the context to reclaim resources.
+func LongLines(ctx context.Context, r io.Reader) Of[io.Reader] {
+	br, ok := r.(io.ByteReader)
+	if !ok {
+		br = bufio.NewReader(r)
+	}
+
+	return Go(func(ch chan<- io.Reader) error {
+		var (
+			pr    io.Reader
+			pw    io.WriteCloser
+			sawCR bool
+		)
+
+		// Defer closing only the final value of pw.
+		defer func() {
+			if pw != nil {
+				pw.Close()
+			}
+		}()
+
+		newpipe := func() error {
+			pr, pw = io.Pipe()
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case ch <- pr:
+			}
+			return nil
+		}
+
+		newline := func() error {
+			if pw == nil {
+				if err := newpipe(); err != nil {
+					return err
+				}
+			}
+
+			if err := pw.Close(); err != nil {
+				return err
+			}
+
+			pw = nil
+			sawCR = false
+
+			return nil
+		}
+
+		write := func(b byte) error {
+			if pw == nil {
+				if err := newpipe(); err != nil {
+					return err
+				}
+			}
+			_, err := pw.Write([]byte{b})
+			return err
+		}
+
+		if err := newpipe(); err != nil {
+			return err
+		}
+
+		for {
+			b, err := br.ReadByte()
+			if errors.Is(err, io.EOF) {
+				if pw != nil {
+					return pw.Close()
+				}
+				return nil
+			}
+			if err != nil {
+				return err
+			}
+
+			if b == '\n' {
+				if err := newline(); err != nil {
+					return err
+				}
+				continue
+			}
+
+			if sawCR {
+				if err := write('\r'); err != nil {
+					return err
+				}
+				sawCR = false
+			}
+			if b == '\r' {
+				sawCR = true
+				continue
+			}
+
+			if err := write(b); err != nil {
+				return err
+			}
+		}
+	})
+}

iter/lines_test.go

@@ -0,0 +1,62 @@
+package iter
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+)
+
+func TestLines(t *testing.T) {
+	indep, err := os.ReadFile("testdata/indep.txt")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var (
+		lines = Lines(bytes.NewReader(indep))
+		got   = new(bytes.Buffer)
+	)
+	for lines.Next() {
+		line := lines.Val()
+		fmt.Fprintln(got, line)
+	}
+	if err := lines.Err(); err != nil {
+		t.Fatal(err)
+	}
+
+	if diff := cmp.Diff(string(indep), got.String()); diff != "" {
+		t.Errorf("mismatch (-want +got):\n%s", diff)
+	}
+}
+
+func TestLongLines(t *testing.T) {
+	indep, err := os.ReadFile("testdata/indep.txt")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var (
+		lines = LongLines(context.Background(), bytes.NewReader(indep))
+		got   = new(bytes.Buffer)
+	)
+	for lines.Next() {
+		r := lines.Val()
+		line, err := io.ReadAll(r)
+		if err != nil {
+			t.Fatal(err)
+		}
+		fmt.Fprintln(got, string(line))
+	}
+	if err := lines.Err(); err != nil {
+		t.Fatal(err)
+	}
+
+	if diff := cmp.Diff(string(indep), got.String()); diff != "" {
+		t.Errorf("mismatch (-want +got):\n%s", diff)
+	}
+}