separate line reader and output channel

Author: fstab

tailer/fileTailer.go

@@ -107,15 +107,23 @@ func runFileTailer(path string, readall bool, failOnMissingFile bool, logger sim
 		if eventLoop != nil {
 			defer eventLoop.Close()
 		}
-		reader := NewBufferedLineReader(lines, done)
+		reader := NewLineReader()
 		if file != nil {
 			// process all pre-existing lines
-			finished, err := reader.ReadAvailableLines(file)
-			if err != nil {
-				writeError(errors, done, err, "failed to initialize file system watcher for %v", path)
-				return
-			} else if finished {
-				return
+			for {
+				line, eof, err := reader.ReadLine(file)
+				if err != nil {
+					writeError(errors, done, err, "failed to initialize file system watcher for %v", path)
+					return
+				}
+				if eof {
+					break
+				}
+				select {
+				case <-done:
+					return
+				case lines <- line:
+				}
 			}
 		}
 
@@ -148,11 +156,19 @@ func runFileTailer(path string, readall bool, failOnMissingFile bool, logger sim
 					}
 					return
 				}
-				file, err = evnts.Process(file, reader, abspath, logger)
+				var freshLines []string
+				file, freshLines, err = evnts.Process(file, reader, abspath, logger)
 				if err != nil {
 					writeError(errors, done, err, "failed to watch %v", abspath)
 					return
 				}
+				for _, line := range freshLines {
+					select {
+					case <-done:
+						return
+					case lines <- line:
+					}
+				}
 			}
 		}
 	}()

tailer/fileTailer_darwin.go

@@ -139,9 +139,13 @@ func (l *eventLoop) Events() chan Events {
 	return l.events
 }
 
-func (events *eventList) Process(fileBefore *File, reader *bufferedLineReader, abspath string, logger simpleLogger) (file *File, err error) {
+func (events *eventList) Process(fileBefore *File, reader *lineReader, abspath string, logger simpleLogger) (file *File, lines []string, err error) {
 	file = fileBefore
-	var truncated bool
+	lines = []string{}
+	var (
+		line           string
+		truncated, eof bool
+	)
 	logger.Debug("File system watcher received %v event(s):\n", len(events.events))
 	for i, event := range events.events {
 		logger.Debug("%v/%v: %v.\n", i+1, len(events.events), event2string(events.watcher.dir, file, event))
@@ -159,17 +163,23 @@ func (events *eventList) Process(fileBefore *File, reader *bufferedLineReader, a
 				if err != nil {
 					return
 				}
+				reader.Clear()
 			}
 		}
 	}
 
 	// Handle write event.
 	for _, event := range events.events {
 		if file != nil && event.Ident == fdToInt(file.Fd()) && event.Fflags&syscall.NOTE_WRITE == syscall.NOTE_WRITE {
-			var finished bool
-			finished, err = reader.ReadAvailableLines(file)
-			if finished || err != nil {
-				return
+			for {
+				line, eof, err = reader.ReadLine(file)
+				if err != nil {
+					return
+				}
+				if eof {
+					break
+				}
+				lines = append(lines, line)
 			}
 		}
 	}
@@ -194,10 +204,15 @@ func (events *eventList) Process(fileBefore *File, reader *bufferedLineReader, a
 					return
 				}
 				reader.Clear()
-				var finished bool
-				finished, err = reader.ReadAvailableLines(file)
-				if finished || err != nil {
-					return
+				for {
+					line, eof, err = reader.ReadLine(file)
+					if err != nil {
+						return
+					}
+					if eof {
+						break
+					}
+					lines = append(lines, line)
 				}
 			} else {
 				// If file could not be opened, the CREATE event was for another file, we ignore this.

tailer/fileTailer_linux.go

@@ -141,10 +141,15 @@ func (l *eventLoop) Events() chan Events {
 	return l.events
 }
 
-func (events eventList) Process(fileBefore *File, reader *bufferedLineReader, abspath string, logger simpleLogger) (file *File, err error) {
+func (events eventList) Process(fileBefore *File, reader *lineReader, abspath string, logger simpleLogger) (file *File, lines []string, err error) {
 	file = fileBefore
-	filename := filepath.Base(abspath)
-	var truncated bool
+	lines = []string{}
+	var (
+		filename  = filepath.Base(abspath)
+		truncated bool
+		line      string
+		eof       bool
+	)
 	for _, event := range events {
 		logger.Debug("File system watcher received %v.\n", event2string(event))
 	}
@@ -161,11 +166,17 @@ func (events eventList) Process(fileBefore *File, reader *bufferedLineReader, ab
 				if err != nil {
 					return
 				}
+				reader.Clear()
 			}
-			var finished bool
-			finished, err = reader.ReadAvailableLines(file)
-			if finished || err != nil {
-				return
+			for {
+				line, eof, err = reader.ReadLine(file)
+				if err != nil {
+					return
+				}
+				if eof {
+					break
+				}
+				lines = append(lines, line)
 			}
 		}
 	}
@@ -187,10 +198,15 @@ func (events eventList) Process(fileBefore *File, reader *bufferedLineReader, ab
 				return
 			}
 			reader.Clear()
-			var finished bool
-			finished, err = reader.ReadAvailableLines(file)
-			if finished || err != nil {
-				return
+			for {
+				line, eof, err = reader.ReadLine(file)
+				if err != nil {
+					return
+				}
+				if eof {
+					break
+				}
+				lines = append(lines, line)
 			}
 		}
 	}

tailer/fileTailer_windows.go

@@ -105,9 +105,13 @@ func (l *eventLoop) Events() chan Events {
 	return l.events
 }
 
-func (event *event) Process(fileBefore *File, reader *bufferedLineReader, abspath string, logger simpleLogger) (file *File, err error) {
+func (event *event) Process(fileBefore *File, reader *lineReader, abspath string, logger simpleLogger) (file *File, lines []string, err error) {
 	file = fileBefore
-	var truncated bool
+	lines = []string{}
+	var (
+		truncated, eof bool
+		line           string
+	)
 	logger.Debug("File system watcher received %v.\n", event.String())
 
 	// WRITE or TRUNCATE
@@ -121,11 +125,17 @@ func (event *event) Process(fileBefore *File, reader *bufferedLineReader, abspat
 			if err != nil {
 				return
 			}
+			reader.Clear()
 		}
-		var finished bool
-		finished, err = reader.ReadAvailableLines(file)
-		if finished || err != nil {
-			return
+		for {
+			line, eof, err = reader.ReadLine(file)
+			if err != nil {
+				return
+			}
+			if eof {
+				break
+			}
+			lines = append(lines, line)
 		}
 	}
 
@@ -142,10 +152,15 @@ func (event *event) Process(fileBefore *File, reader *bufferedLineReader, abspat
 			return
 		}
 		reader.Clear()
-		var finished bool
-		finished, err = reader.ReadAvailableLines(file)
-		if finished || err != nil {
-			return
+		for {
+			line, eof, err = reader.ReadLine(file)
+			if err != nil {
+				return
+			}
+			if eof {
+				break
+			}
+			lines = append(lines, line)
 		}
 	}
 	return

tailer/linereader.go

@@ -16,72 +16,65 @@ package tailer
 
 import (
 	"bytes"
-	"fmt"
 	"io"
 )
 
-type bufferedLineReader struct {
+type lineReader struct {
 	remainingBytesFromLastRead []byte
-
-	// channels are used to stream the results out
-	lines chan<- string
-	done  <-chan struct{}
 }
 
-func NewBufferedLineReader(lines chan<- string, done <-chan struct{}) *bufferedLineReader {
-	return &bufferedLineReader{
+func NewLineReader() *lineReader {
+	return &lineReader{
 		remainingBytesFromLastRead: []byte{},
-		lines: lines,
-		done:  done,
 	}
 }
 
-func (r *bufferedLineReader) ReadAvailableLines(file io.Reader) (bool, error) {
-
-	// for each buffer, split lines and stream
-	buf := make([]byte, 512)
-	var done bool
-
+// read the next line from the file.
+// return values are (line, eof, err).
+// * line is the line read.
+// * eof is a boolean indicating if the end of file was reached before getting to the next '\n'.
+// * err is set if an error other than io.EOF has occurred. err is never io.EOF.
+// if eof is true, line is always "" and err always is nil.
+// if eof is false and err is nil, an empty line means that there actually was an empty line in the file.
+func (r *lineReader) ReadLine(file io.Reader) (string, bool, error) {
+	var (
+		err error
+		buf = make([]byte, 512)
+		n   = 0
+	)
 	for {
-		n, err := file.Read(buf)
-		if n > 0 {
-			// Callers should always process the n > 0 bytes returned before considering the error err.
-			result := append(r.remainingBytesFromLastRead, buf[0:n]...)
-			done, r.remainingBytesFromLastRead = r.processLines(result)
-			if done {
-				return true, nil
-			}
-		}
-		if err != nil {
+		newlinePos := bytes.IndexByte(r.remainingBytesFromLastRead, '\n')
+		if newlinePos >= 0 {
+			l := len(r.remainingBytesFromLastRead)
+			result := make([]byte, newlinePos)
+			copy(result, r.remainingBytesFromLastRead[:newlinePos])
+			copy(r.remainingBytesFromLastRead, r.remainingBytesFromLastRead[newlinePos+1:])
+			r.remainingBytesFromLastRead = r.remainingBytesFromLastRead[:l-(newlinePos+1)]
+			return string(stripWindowsLineEnding(result)), false, nil
+		} else if err != nil {
 			if err == io.EOF {
-				return false, nil
+				return "", true, nil
 			} else {
-				return false, fmt.Errorf("read error: %v", err.Error())
+				return "", false, err
+			}
+		} else {
+			n, err = file.Read(buf)
+			if n > 0 {
+				// io.Reader: Callers should always process the n > 0 bytes returned before considering the error err.
+				r.remainingBytesFromLastRead = append(r.remainingBytesFromLastRead, buf[0:n]...)
 			}
 		}
 	}
 }
 
-func (r *bufferedLineReader) Clear() {
-	r.remainingBytesFromLastRead = []byte{}
+func stripWindowsLineEnding(s []byte) []byte {
+	if len(s) > 0 && s[len(s)-1] == '\r' {
+		return s[:len(s)-1]
+	} else {
+		return s
+	}
 }
 
-func (r *bufferedLineReader) processLines(data []byte) (finished bool, remainingBytes []byte) {
-	newline := []byte("\n")
-	for _, line := range bytes.SplitAfter(data, newline) {
-		if bytes.HasSuffix(line, newline) {
-			line = bytes.TrimSuffix(line, newline)
-			line = bytes.TrimSuffix(line, []byte("\r")) // Needed for CRLF line endings?
-			select {
-			case r.lines <- string(line):
-			case <-r.done:
-				finished = true
-				return
-			}
-		} else {
-			// This is the last (incomplete) line returned by SplitAfter(). We will exit the for loop here.
-			remainingBytes = line
-		}
-	}
-	return
+func (r *lineReader) Clear() {
+	r.remainingBytesFromLastRead = r.remainingBytesFromLastRead[:0]
 }