use mutex to synchronize

Author: swinkelhofer

encoder.go

@@ -117,11 +117,13 @@ func (enc *Encoder) Stream() chan<- interface{} {
 				value interface{}
 			)
 
+			enc.mutex.Lock()
 			for value = range enc.s {
 				if enc.err = enc.write(value); enc.err != nil {
 					continue
 				}
 			}
+			enc.mutex.Unlock()
 		}()
 	})
 
@@ -141,12 +143,14 @@ func (enc *Encoder) Close() {
 			time.Sleep(100 * time.Microsecond)
 		}
 		close(enc.s)
+		enc.mutex.Lock()
 
 		enc.buf.Flush()
 
 		if _, ok = enc.w.(io.Closer); ok {
 			enc.w.(io.Closer).Close()
 		}
+		enc.mutex.Unlock()
 	})
 
 }

linebuf.go

@@ -27,6 +27,7 @@ type LinebufJSONConverter struct {
 	r          *bufio.Reader
 	err        error
 	finished   bool
+	mutex      sync.Mutex
 }
 
 /*
@@ -43,6 +44,7 @@ type Encoder struct {
 	closeOnce   sync.Once
 	unbuffered  bool
 	bufBytes    uint64
+	mutex       sync.Mutex
 }
 
 /*

linebuf2json.go

@@ -3,7 +3,6 @@ package linebuf
 import (
 	"bufio"
 	"io"
-	"time"
 )
 
 /*
@@ -12,9 +11,15 @@ NewLinebufJSONConverter returns a writer that converts line-buffered JSON back t
 func NewLinebufJSONConverter(w io.WriteCloser) *LinebufJSONConverter {
 	var (
 		r, pipeWriter   = io.Pipe()
-		bufReader       = bufio.NewReaderSize(r, 4096)
-		bufWriter       = bufio.NewWriterSize(pipeWriter, 4096)
-		sanitizedWriter = &LinebufJSONConverter{bufWriter, w, pipeWriter, bufReader, nil, false}
+		bufReader       = bufio.NewReaderSize(r, 16)
+		bufWriter       = bufio.NewWriterSize(pipeWriter, 16)
+		sanitizedWriter = &LinebufJSONConverter{
+			Writer:     bufWriter,
+			w:          w,
+			pipeWriter: pipeWriter,
+			r:          bufReader,
+			err:        nil,
+		}
 	)
 
 	go func() {
@@ -24,8 +29,9 @@ func NewLinebufJSONConverter(w io.WriteCloser) *LinebufJSONConverter {
 			isArray   = false
 		)
 
+		sanitizedWriter.mutex.Lock()
 		defer func() {
-			sanitizedWriter.finished = true
+			sanitizedWriter.mutex.Unlock()
 		}()
 
 		if firstline, sanitizedWriter.err = sanitizedWriter.r.ReadBytes('\n'); sanitizedWriter.err != nil {
@@ -68,12 +74,20 @@ func NewLinebufJSONConverter(w io.WriteCloser) *LinebufJSONConverter {
 Close waits until the last writes have finished and gracefully closes the underlaying writers
 */
 func (l *LinebufJSONConverter) Close() error {
-	l.Writer.Flush()
-	l.pipeWriter.Close()
-	for !l.finished {
-		time.Sleep(100 * time.Microsecond)
+	var (
+		err error
+	)
+
+	if err = l.Writer.Flush(); err != nil {
+		return err
+	}
+	if err = l.pipeWriter.Close(); err != nil {
+		return err
 	}
-	return l.w.Close()
+	l.mutex.Lock()
+	err = l.w.Close()
+	l.mutex.Unlock()
+	return err
 }
 
 /*