2 files changed, 61 insertions, 0 deletions
diff --git a/writer/linesplitwriter.go b/writer/linesplitwriter.go
new file mode 100644
index 0000000..ebb90c4
--- /dev/null
+++ b/writer/linesplitwriter.go
@@ -0,0 +1,29 @@
+package writer
+
+import (
+	"bytes"
+	"io"
+)
+
+func LineSplitting(sink io.Writer) io.Writer {
+	return lineSplittingWriter{writer: sink}
+}
+
+type lineSplittingWriter struct {
+	writer io.Writer
+}
+
+func (lsw lineSplittingWriter) Write(p []byte) (int, error) {
+	for _, line := range bytes.Split(p, []byte("\n")) {
+		if len(line) == 0 {
+			continue
+		}
+
+		_, err := lsw.writer.Write(append(line, '\n'))
+		if err != nil {
+			return -1, err
+		}
+	}
+
+	return len(p), nil
+}
diff --git a/writer/prefixwriter.go b/writer/prefixwriter.go
new file mode 100644
index 0000000..c019c3a
--- /dev/null
+++ b/writer/prefixwriter.go
@@ -0,0 +1,32 @@
+package writer
+
+import "io"
+
+type prefixingWriter struct {
+	prefix []byte
+	writer io.Writer
+}
+
+func (pw *prefixingWriter) Write(p []byte) (int, error) {
+	if len(p) == 0 {
+		return 0, nil
+	}
+
+	// Perform an "atomic" write of a prefix and p to make sure that it doesn't interleave
+	// sub-line when used concurrently with io.PipeWrite.
+	n, err := pw.writer.Write(append(pw.prefix, p...))
+	if n > len(p) {
+		// To comply with the io.Writer interface requirements we must
+		// return a number of bytes written from p (0 <= n <= len(p)),
+		// so we are ignoring the length of the prefix here.
+		return len(p), err
+	}
+	return n, err
+}
+
+func PrefixWriter(prefix string, sink io.Writer) io.Writer {
+	return &prefixingWriter{
+		prefix: []byte(prefix),
+		writer: sink,
+	}
+}