line_pipe.go

package util

import (
	"io"
)

type LinePipe struct {
	leftover     []byte
	scanner      TextScanner
	bytesWritten int
	newLineChar  string
}

type TextScanner interface {
	Text() string
	Err() error
	Scan() bool
}

func NewLinePipe(scanner TextScanner) *LinePipe {
	lp := &LinePipe{scanner: scanner}
	lp.newLineChar = "\n"
	return lp
}

func (lp *LinePipe) NewLineChar(newlineChar string) {
	lp.newLineChar = newlineChar
}

// Write as much as possible from data to p, save leftover bytes for next round
func (lp *LinePipe) write(p []byte, data []byte) int {
	end := lp.bytesWritten + len(data)
	end = min(end, len(p))
	written := copy(p[lp.bytesWritten:end], data)
	lp.bytesWritten += written
	lp.leftover = data[written:]
	return written
}

func (lp *LinePipe) Read(p []byte) (n int, err error) {
	lp.bytesWritten = 0

	// First empty local buffer
	if len(lp.leftover) > 0 {
		toWrite := len(lp.leftover)
		written := lp.write(p, lp.leftover)
		if written < toWrite {
			return lp.bytesWritten, nil
		}
	}

	// If there's still room in p, write from scanner
	for {
		if !lp.scanner.Scan() {
			if err := lp.scanner.Err(); err != nil {
				return lp.bytesWritten, err
			} else {
				return lp.bytesWritten, io.EOF
			}
		}
		line := lp.scanner.Text() + lp.newLineChar
		data := []byte(line)

		written := lp.write(p, data)
		if written < len(data) {
			break
		}
	}

	return lp.bytesWritten, nil
}

func min(a, b int) int {
	if a < b {
		return a
	}
	return b
}