add a custom TextFormatter

use logrus' formatter instead of the built in one for compatibility reasons (closes #19)
only write a new line if the log entry doesn't have one

glide.yaml

@@ -9,7 +9,7 @@ import:
   - ssh/terminal
 - package: github.com/araddon/dateparse
 - package: github.com/namsral/flag
-  version: v1.7.4-pre
+  version: master
 - package: github.com/sirupsen/logrus
   version: master
 testImport:

reader/factory.go

@@ -16,7 +16,11 @@ import (
 
 const (
 	// INFO is a log level.
-	INFO = "INFO"
+	INFO = "info"
+	// ERROR is a log level.
+	ERROR = "error"
+	// WARN is a log level.
+	WARN = "warning"
 )
 
 // GetReader tries to guess an appropriate reader from the input byte slice

reader/factory_benchmark_test.go

@@ -18,7 +18,7 @@ func BenchmarkGetReaderPlain(b *testing.B) {
 }
 
 func BenchmarkGetReaderPlainMedium(b *testing.B) {
-	benchmarkGetReaderPlain(b, "adkjh kjhasdkjh kjhkjahsd", "12:01:10", 100)
+	benchmarkGetReaderPlain(b, "adkjh kjhasdkjh kjhkjahsd", "2013-01-10", 100)
 }
 
 func BenchmarkGetReaderPlainLarge(b *testing.B) {

reader/plain.go

@@ -7,16 +7,20 @@ package reader
 import (
 	"bytes"
 	"io"
-	"strings"
 	"sync"
 	"time"
 
+	"github.com/araddon/dateparse"
 	"github.com/arsham/logpipe/internal"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
 )
 
 // TimestampFormat is the default formatting defined for logs.
 var TimestampFormat = time.RFC3339
 
+// var TimestampFormat = "2006-01-02 15:04:05"
+
 // Plain implements the io.Reader interface and can read a json object and
 // output a one line error message. For example:
 //
@@ -40,9 +44,36 @@ type Plain struct {
 	current  int //current position on reading the message
 }
 
-func (p *Plain) Read(b []byte) (int, error) {
-	var n int
+// TextFormatter is used for rendering a custom format. We need to put the time
+// at the very beginning of the line.
+type TextFormatter struct {
+	logrus.TextFormatter
+}
+
+// Format will use the timestamp passed by the payload and injects it in the
+// entry itself.
+func (f *TextFormatter) Format(entry *logrus.Entry) ([]byte, error) {
+	ts, ok := entry.Data["time"].(string)
+	if !ok {
+		return nil, errors.New("no time in the log entry")
+	}
+	t, err := dateparse.ParseAny(ts)
+	if err != nil {
+		return nil, errors.Wrap(err, "parsing datetime")
+	}
+	e := &logrus.Entry{
+		Logger:  entry.Logger,
+		Time:    t,
+		Level:   entry.Level,
+		Message: entry.Message,
+		Buffer:  entry.Buffer,
+	}
+	delete(entry.Data, "time")
+	e.Data = entry.Data
+	return f.TextFormatter.Format(e)
+}
 
+func (p *Plain) Read(b []byte) (int, error) {
 	if p.Timestamp.Equal(time.Time{}) {
 		p.Logger.Error(ErrNilTimestamp)
 		return 0, ErrNilTimestamp
@@ -63,24 +94,24 @@ func (p *Plain) Read(b []byte) (int, error) {
 	}
 
 	p.once.Do(func() {
-		t := p.Timestamp.Format(TimestampFormat)
-		l := 6 + len(p.Kind) + len(p.Message) + len(t)
-		buf := bytes.NewBuffer(make([]byte, l))
-		buf.Reset()
-
-		inputs := []string{
-			"[",
-			t,
-			"] [",
-			strings.ToUpper(p.Kind),
-			"] ",
-			p.Message,
-		}
-		for _, in := range inputs {
-			nn, _ := buf.WriteString(in)
-			n += nn
+		logger := logrus.New()
+		customFormatter := new(TextFormatter)
+		customFormatter.DisableColors = true
+		logger.Formatter = customFormatter
+
+		buf := new(bytes.Buffer)
+		logger.Out = buf
+		ll := logger.WithField("time", p.Timestamp.Format(TimestampFormat))
+		switch p.Kind {
+		case INFO:
+			ll.Info(p.Message)
+		case WARN:
+			ll.Warn(p.Message)
+		case ERROR:
+			ll.Error(p.Message)
 		}
 		p.compiled = buf.Bytes()
+
 	})
 
 	end := len(p.compiled)

reader/plain_test.go

@@ -17,6 +17,7 @@ import (
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
 	"github.com/sirupsen/logrus/hooks/test"
 )
 
@@ -28,8 +29,8 @@ var _ = Describe("Plain", func() {
 
 	BeforeEach(func() {
 		logger, hook = test.NewNullLogger()
-
 	})
+
 	AfterEach(func() {
 		hook.Reset()
 	})
@@ -105,10 +106,15 @@ var _ = Describe("Plain", func() {
 				var (
 					rp      *reader.Plain
 					kind    string
-					message = strings.Repeat("this is a long message", 150)
+					message = strings.Repeat("this is a long message ", 150)
 					now     = time.Now()
 					nowStr  = now.Format(reader.TimestampFormat)
+					format  = `time="%s" level=%s msg="%s"`
 				)
+				b := new(bytes.Buffer)
+				b.WriteString(format)
+				b.WriteByte('\n')
+				format = b.String()
 
 				JustBeforeEach(func() {
 					rp = &reader.Plain{
@@ -129,8 +135,8 @@ var _ = Describe("Plain", func() {
 					)
 
 					BeforeEach(func() {
-						expected = fmt.Sprintf("[%s] [%s] %s", nowStr, "ERROR", message)
-						kind = "error"
+						expected = fmt.Sprintf(format, nowStr, reader.ERROR, message)
+						kind = reader.ERROR
 						b = make([]byte, len(expected))
 					})
 					It("should not error or return io.EOF", func() {
@@ -151,8 +157,8 @@ var _ = Describe("Plain", func() {
 					)
 
 					BeforeEach(func() {
-						expected = fmt.Sprintf("[%s] [%s] %s", nowStr, "ERROR", message)
-						kind = "error"
+						expected = fmt.Sprintf(format, nowStr, reader.WARN, message)
+						kind = reader.WARN
 						buf = &bytes.Buffer{}
 					})
 					It("should not error or return io.EOF", func() {
@@ -171,20 +177,109 @@ var _ = Describe("Plain", func() {
 					var expected string
 
 					BeforeEach(func() {
-						expected = fmt.Sprintf("[%s] [%s] %s", nowStr, "ERROR", message)
-						kind = "error"
+						expected = fmt.Sprintf(format, nowStr, reader.INFO, message)
+						kind = reader.INFO
 					})
 					It("should not error or return io.EOF", func() {
 						_, err := ioutil.ReadAll(rp)
 						Expect(err).To(Or(BeNil(), Equal(io.EOF)))
 					})
 					Specify("length and return value to be as expected", func() {
 						b, _ := ioutil.ReadAll(rp)
+						Expect(string(b)).To(BeEquivalentTo(expected))
 						Expect(len(b)).To(Equal(len(expected)))
-						Expect(b).To(BeEquivalentTo(expected))
 					})
 				})
 			})
 		})
 	})
 })
+
+var _ = Describe("TextFormatter", func() {
+	Describe("Format", func() {
+		var (
+			t         time.Time
+			timeStr   string
+			entry     logrus.Entry
+			formatter *reader.TextFormatter
+		)
+		JustBeforeEach(func() {
+			data := logrus.Fields{
+				"msg":   "this is a message",
+				"level": "error",
+				"time":  timeStr,
+			}
+			entry = logrus.Entry{
+				Logger:  internal.DiscardLogger().Logger,
+				Time:    t,
+				Level:   internal.WarnLevel,
+				Message: "this is a message",
+				Data:    data,
+			}
+			formatter = new(reader.TextFormatter)
+			formatter.DisableColors = true
+		})
+		AfterEach(func() {
+			timeStr = ""
+		})
+
+		Context("having an entry without a timestamp", func() {
+			var (
+				b   []byte
+				err error
+			)
+			BeforeEach(func() {
+				t = time.Time{}
+			})
+			JustBeforeEach(func() {
+				b, err = formatter.Format(&entry)
+			})
+
+			It("should return error", func() {
+				Expect(err).To(HaveOccurred())
+				Expect(b).To(BeEmpty())
+			})
+		})
+
+		Context("having Data without the time key", func() {
+			var (
+				b   []byte
+				err error
+			)
+
+			JustBeforeEach(func() {
+				delete(entry.Data, "time")
+				b, err = formatter.Format(&entry)
+			})
+
+			It("should return error", func() {
+				Expect(err).To(HaveOccurred())
+				Expect(b).To(BeEmpty())
+			})
+		})
+
+		Context("having a ready to use entry", func() {
+			var (
+				b   []byte
+				err error
+			)
+			BeforeEach(func() {
+				t = time.Now()
+				timeStr = t.Format(reader.TimestampFormat)
+			})
+			JustBeforeEach(func() {
+				b, err = formatter.Format(&entry)
+			})
+
+			It("should not return error", func() {
+				Expect(err).NotTo(HaveOccurred())
+			})
+			It("should remove the time from its Data slice", func() {
+				Expect(entry.Data).NotTo(HaveKey("time"))
+			})
+			Specify("the time should appear at the beginning of the line", func() {
+				Expect(string(b)).To(HavePrefix("time="))
+			})
+		})
+	})
+})

writer/file.go

@@ -6,6 +6,7 @@ package writer
 
 import (
 	"bufio"
+	"bytes"
 	"fmt"
 	"io"
 	"os"
@@ -81,9 +82,12 @@ func (f *File) Write(p []byte) (int, error) {
 		return n1, errors.Wrap(err, "writing the bytes")
 	}
 
-	n2, err := f.buf.Write([]byte("\n")) // required for creating a new line
+	if !bytes.HasSuffix(p, []byte("\n")) {
+		err = f.buf.WriteByte('\n') // required for creating a new line
+	}
+
 	if err != nil {
-		return n2, errors.Wrap(err, "writing new line")
+		return 0, errors.Wrap(err, "writing new line")
 	}
 	return n1, nil
 }