refactored constructors

Author: swinkelhofer

decoder.go

@@ -3,49 +3,81 @@ package linebuf
 import (
 	"bufio"
 	"io"
+	"math"
 	"reflect"
 
 	"code.cloudfoundry.org/bytefmt"
 	"github.com/segmentio/encoding/json"
 )
 
+type DecoderOption (func(*Decoder) error)
+
 /*
-NewDecoder returns a linebuf decoder from an io.Reader. The buffersize will be 4kB
+WithDecoderUnbuffered can be provided as option to NewDecoder, thus turning off buffer mode
 */
-func NewDecoder(r io.Reader) *Decoder {
-	d, _ := NewDecoderWithBuffersize(r, "4k")
-	return d
+func WithDecoderUnbuffered() DecoderOption {
+	return func(dec *Decoder) error {
+		dec.unbuffered = true
+		return nil
+	}
 }
 
 /*
-NewDecoderWithBuffersize returns a linebuf decoder from an io.Reader while allowing to specify a custom buffersize.
+WithDecoderBuffersize can be used as option to NewDecoder, thus providing another buffersize than the default 4k
+*/
+func WithDecoderBuffersize(bufSize string) DecoderOption {
+	return func(dec *Decoder) error {
+		var (
+			bufBytes uint64
+			err      error
+		)
+		if bufBytes, err = bytefmt.ToBytes(bufSize); err != nil {
+			return err
+		}
+		dec.bufBytes = bufBytes
+		return nil
+	}
+}
 
-An error will be returned if parsing of the bufsize fails
+/*
+NewDecoder returns a linebuf decoder from an io.Reader while allowing to specify custom options like unbuffered mode and/or setting the buffer size.
+
+The default buffer size is 4k
+
+An error will be returned e.g. if parsing of the bufsize fails
 */
-func NewDecoderWithBuffersize(r io.Reader, bufSize string) (*Decoder, error) {
+func NewDecoder(r io.Reader, options ...DecoderOption) (*Decoder, error) {
 	var (
-		err      error
-		bufBytes uint64
-		jsonDec  *json.Decoder
-		buf      *bufio.Reader
+		err     error
+		jsonDec *json.Decoder
+		option  DecoderOption
+		dec     = &Decoder{
+			err: nil,
+			r:   r,
+			s:   make(chan interface{}),
+		}
+		buf *bufio.Reader
 	)
 
-	if bufBytes, err = bytefmt.ToBytes(bufSize); err != nil {
-		return nil, err
+	for _, option = range options {
+		if err = option(dec); err != nil {
+			return nil, err
+		}
 	}
 
-	buf = bufio.NewReaderSize(r, int(bufBytes))
+	buf = bufio.NewReaderSize(r, int(math.Max(float64(dec.bufBytes), 4<<10)))
 	jsonDec = json.NewDecoder(buf)
 	jsonDec.ZeroCopy()
 	jsonDec.UseNumber()
 	jsonDec.DontMatchCaseInsensitiveStructFields()
-	return &Decoder{
-		buf:     buf,
-		err:     nil,
-		r:       r,
-		s:       make(chan interface{}, 10),
-		jsonDec: jsonDec,
-	}, nil
+
+	dec.buf = buf
+	dec.jsonDec = jsonDec
+	if !dec.unbuffered {
+		dec.s = make(chan interface{}, 10)
+	}
+
+	return dec, nil
 }
 
 /*

encoder.go

@@ -3,44 +3,71 @@ package linebuf
 import (
 	"bufio"
 	"io"
+	"math"
 	"time"
 
 	"code.cloudfoundry.org/bytefmt"
 	"github.com/segmentio/encoding/json"
 )
 
+type EncoderOption (func(*Encoder) error)
+
 /*
-NewEncoder returns a linebuf encoder from an io.Writer. The buffersize will be 4kB
+WithUnbuffered can be provided as option to NewEncoder, thus turning off buffer mode
 */
-func NewEncoder(w io.Writer) *Encoder {
-	e, _ := NewEncoderWithBuffersize(w, "4k")
-	return e
+func WithEncoderUnbuffered() EncoderOption {
+	return func(enc *Encoder) error {
+		enc.unbuffered = true
+		return nil
+	}
 }
 
 /*
-NewEncoderWithBuffersize returns a linebuf encoder from an io.Writer while allowing to specify a custom buffersize.
+WithBuffersize can be used as option to NewEncoder, thus providing another buffersize than the default 4k
+*/
+func WithEncoderBuffersize(bufSize string) EncoderOption {
+	return func(enc *Encoder) error {
+		var (
+			bufBytes uint64
+			err      error
+		)
+		if bufBytes, err = bytefmt.ToBytes(bufSize); err != nil {
+			return err
+		}
+		enc.bufBytes = bufBytes
+		return nil
+	}
+}
+
+/*
+NewEncoder returns a linebuf encoder from an io.Writer while allowing to sspecify custom options like unbuffered mode and/or setting the buffer size.
 
-An error will be returned if parsing of the bufsize fails
+The default buffer size is 4k
+
+An error will be returned e.g. if parsing of the bufsize fails
 */
-func NewEncoderWithBuffersize(w io.Writer, bufSize string) (*Encoder, error) {
+func NewEncoder(w io.Writer, options ...EncoderOption) (*Encoder, error) {
 	var (
-		err      error
-		bufBytes uint64
-		buf      *bufio.Writer
+		err    error
+		option EncoderOption
+		buf    *bufio.Writer
+		enc    = &Encoder{
+			w:   w,
+			s:   make(chan interface{}),
+			err: nil,
+		}
 	)
 
-	if bufBytes, err = bytefmt.ToBytes(bufSize); err != nil {
-		return nil, err
+	for _, option = range options {
+		if err = option(enc); err != nil {
+			return nil, err
+		}
 	}
 
-	buf = bufio.NewWriterSize(w, int(bufBytes))
-	return &Encoder{
-		buf:         buf,
-		w:           w,
-		s:           make(chan interface{}, 10),
-		err:         nil,
-		jsonEncoder: json.NewEncoder(buf),
-	}, nil
+	buf = bufio.NewWriterSize(w, int(math.Max(float64(enc.bufBytes), 4<<10)))
+	enc.buf = buf
+	enc.jsonEncoder = json.NewEncoder(enc.buf)
+	return enc, nil
 }
 
 /*

linebuf.go

@@ -41,17 +41,21 @@ type Encoder struct {
 	once        sync.Once
 	jsonEncoder *json.Encoder
 	closeOnce   sync.Once
+	unbuffered  bool
+	bufBytes    uint64
 }
 
 /*
 Decoder reads line-buffered JSON from the underlaying io.Reader and decodes them in single-entry modde or stream mode
 */
 type Decoder struct {
 	context.Context
-	buf     *bufio.Reader
-	r       io.Reader
-	s       chan interface{}
-	once    sync.Once
-	err     error
-	jsonDec *json.Decoder
+	buf        *bufio.Reader
+	r          io.Reader
+	s          chan interface{}
+	once       sync.Once
+	err        error
+	jsonDec    *json.Decoder
+	unbuffered bool
+	bufBytes   uint64
 }