Function TestScannerAlwaysFlushes

internal/pipe/pipeline_test.go:493–539

(t *testing.T)

}

func TestScannerAlwaysFlushes(t *testing.T) {
	t.Parallel()
	ctx := context.Background()

	var length int64

	p := pipe.New()
	// Print the numbers from 1 to 20 (generated from scratch):
	p.Add(
		pipe.IgnoreError(
			seqFunction(20),
			pipe.IsPipeError,
		),
		// Pass the numbers through up to 7, then exit with an
		// ignored error:
		pipe.IgnoreError(
			pipe.LinewiseFunction(
				"error-after-7",
				func(_ context.Context, _ pipe.Env, line []byte, w *bufio.Writer) error {
					fmt.Fprintf(w, "%s\n", line)
					if string(line) == "7" {
						return errors.New("ignore")
					}
					return nil
				},
			),
			func(err error) bool {
				return err.Error() == "ignore"
			},
		),
		// Read the numbers and add them into the sum:
		pipe.Function(
			"compute-length",
			func(_ context.Context, _ pipe.Env, stdin io.Reader, _ io.Writer) error {
				var err error
				length, err = io.Copy(io.Discard, stdin)
				return err
			},
		),
	)

	err := p.Run(ctx)
	assert.NoError(t, err)
	// Make sure that all of the bytes emitted before the second
	// stage's error were received by the third stage:
	assert.EqualValues(t, 14, length)
}

func TestScannerFinishEarly(t *testing.T) {
	t.Parallel()