Function TestPipelineWithLinewiseFunction

internal/pipe/pipeline_test.go:444–491

(t *testing.T)

}

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

	p := pipe.New()
	// Print the numbers from 1 to 20 (generated from scratch):
	p.Add(
		seqFunction(20),
		// Discard all but the multiples of 5, and emit the results
		// separated by spaces on one line:
		pipe.LinewiseFunction(
			"multiples-of-5",
			func(_ context.Context, _ pipe.Env, line []byte, w *bufio.Writer) error {
				n, err := strconv.Atoi(string(line))
				if err != nil {
					return err
				}
				if n%5 != 0 {
					return nil
				}
				_, err = fmt.Fprintf(w, " %d", n)
				return err
			},
		),
		// Read the words and square them, emitting the results one per
		// line:
		pipe.ScannerFunction(
			"square-multiples-of-5",
			func(r io.Reader) (pipe.Scanner, error) {
				scanner := bufio.NewScanner(r)
				scanner.Split(bufio.ScanWords)
				return scanner, nil
			},
			func(_ context.Context, _ pipe.Env, line []byte, w *bufio.Writer) error {
				n, err := strconv.Atoi(string(line))
				if err != nil {
					return err
				}
				_, err = fmt.Fprintf(w, "%d\n", n*n)
				return err
			},
		),
	)

	out, err := p.Output(ctx)
	assert.NoError(t, err)
	assert.EqualValues(t, "25\n100\n225\n400\n", out)
}

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