func NewBytesScanner(r io.Reader) *BytesScanner {
	return &BytesScanner{Scanner: bufio.NewScanner(r)}
}
NewBytesScanner returns a new Scanner to read from r. The split function
defaults to bufio.ScanLines.
package pipeline
import (
	"bufio"
	"context"
	"io"
	"reflect"
)
type Source interface {
	Err() error
	Prepare(ctx context.Context) (size int)
	Fetch(size int) (fetched int)
	Data() interface{}
}
type sliceSource struct {
	value reflect.Value
	size  int
	data  interface{}
}
func newSliceSource(value reflect.Value) *sliceSource {
	size := value.Len()
	return &sliceSource{value: value.Slice(0, size), size: size}
}
func (src *sliceSource) Err() error {
	return nil
}
func (src *sliceSource) Prepare(_ context.Context) int {
	return src.size
}
func (src *sliceSource) Fetch(n int) (fetched int) {
	switch {
	case src.size == 0:
		src.data = nil
	case n >= src.size:
		fetched = src.size
		src.data = src.value.Interface()
		src.value = reflect.ValueOf(nil)
		src.size = 0
	default:
		fetched = n
		src.data = src.value.Slice(0, n).Interface()
		src.value = src.value.Slice(n, src.size)
		src.size -= n
	}
	return
}
func (src *sliceSource) Data() interface{} {
	return src.data
}
type chanSource struct {
	cases []reflect.SelectCase
	zero  reflect.Value
	data  interface{}
}
func newChanSource(value reflect.Value) *chanSource {
	zeroElem := value.Type().Elem()
	return &chanSource{
		cases: []reflect.SelectCase{{Dir: reflect.SelectRecv, Chan: value}},
		zero:  reflect.Zero(reflect.SliceOf(zeroElem)),
	}
}
func (src *chanSource) Err() error {
	return nil
}
func (src *chanSource) Prepare(ctx context.Context) (size int) {
	src.cases = append(src.cases, reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(ctx.Done())})
	return -1
}
func (src *chanSource) Fetch(n int) (fetched int) {
	data := src.zero
	for fetched = 0; fetched < n; fetched++ {
		if chosen, element, ok := reflect.Select(src.cases); (chosen == 0) && ok {
			data = reflect.Append(data, element)
		} else {
			break
		}
	}
	src.data = data.Interface()
	return
}
func (src *chanSource) Data() interface{} {
	return src.data
}
func reflectSource(source interface{}) Source {
	switch value := reflect.ValueOf(source); value.Kind() {
	case reflect.Array, reflect.Slice, reflect.String:
		return newSliceSource(value)
	case reflect.Chan:
		return newChanSource(value)
	default:
		panic("A default pipeline source is not of kind Array, Slice, String, or Chan.")
	}
}
type Scanner struct {
	*bufio.Scanner
	data interface{}
}
func NewScanner(r io.Reader) *Scanner {
	return &Scanner{Scanner: bufio.NewScanner(r)}
}
func (src *Scanner) Prepare(_ context.Context) (size int) {
	return -1
}
func (src *Scanner) Fetch(n int) (fetched int) {
	var data []string
	for fetched = 0; fetched < n; fetched++ {
		if src.Scan() {
			data = append(data, src.Text())
		} else {
			break
		}
	}
	src.data = data
	return
}
func (src *Scanner) Data() interface{} {
	return src.data
}
type BytesScanner struct {
	*bufio.Scanner
	data interface{}
}