framing: API cleanup and unit testing

api/v1/lib/encoding/framing/decoder.go

@@ -1,7 +1,6 @@
 package framing
 
 import (
-	"fmt"
 	"io"
 )
 
@@ -10,39 +9,33 @@ type (
 	UnmarshalFunc func([]byte, interface{}) error
 
 	// Decoder reads and decodes Protobuf messages from an io.Reader.
-	Decoder struct {
-		r   Reader
-		buf []byte
-		uf  UnmarshalFunc
+	Decoder interface {
+		// Decode reads the next encoded message from its input and stores it
+		// in the value pointed to by m. If m isn't a proto.Message, Decode will panic.
+		Decode(interface{}) error
 	}
-)
 
-// NewDecoder returns a new Decoder that reads from the given io.Reader.
-// If r does not also implement StringReader, it will be wrapped in a bufio.Reader.
-func NewDecoder(r Reader, uf UnmarshalFunc) *Decoder {
-	return &Decoder{r: r, buf: make([]byte, 4096), uf: uf}
-}
+	// DecoderFunc is the functional adaptation of Decoder
+	DecoderFunc func(interface{}) error
+)
 
-// MaxSize is the maximum decodable message size.
-const MaxSize = 4 << 20 // 4MB
+func (f DecoderFunc) Decode(m interface{}) error { return f(m) }
 
-var (
-	// ErrSize is returned by Decode calls when a message would exceed the maximum
-	// allowed size.
-	ErrSize = fmt.Errorf("proto: message exceeds %dMB", MaxSize>>20)
-)
+var _ = Decoder(DecoderFunc(nil))
 
-// Decode reads the next Protobuf-encoded message from its input and stores it
-// in the value pointed to by m. If m isn't a proto.Message, Decode will panic.
-func (d *Decoder) Decode(m interface{}) error {
-	// Note: the buf returned by ReadFrame will change over time, it can't be sub-sliced
-	// and then those sub-slices retained. Examination of generated proto code seems to indicate
-	// that byte buffers are copied vs. referenced by sub-slice (gogo protoc).
-	frame, err := d.r.ReadFrame()
-	if err == nil || err == io.EOF {
-		if err2 := d.uf(frame, m); err2 != nil {
-			err = err2
+// NewDecoder returns a new Decoder that reads from the given io.Reader.
+// If r does not also implement StringReader, it will be wrapped in a bufio.Reader.
+func NewDecoder(r Reader, uf UnmarshalFunc) DecoderFunc {
+	return func(m interface{}) error {
+		// Note: the buf returned by ReadFrame will change over time, it can't be sub-sliced
+		// and then those sub-slices retained. Examination of generated proto code seems to indicate
+		// that byte buffers are copied vs. referenced by sub-slice (gogo protoc).
+		frame, err := r.ReadFrame()
+		if err == nil || err == io.EOF {
+			if err2 := uf(frame, m); err2 != nil {
+				err = err2
+			}
 		}
+		return err
 	}
-	return err
 }

api/v1/lib/encoding/framing/decoder_test.go

@@ -0,0 +1,61 @@
+package framing
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"reflect"
+	"testing"
+)
+
+func TestNewDecoder(t *testing.T) {
+	var (
+		byteCopy = UnmarshalFunc(func(b []byte, m interface{}) error {
+			if m == nil {
+				return errors.New("unmarshal target may not be nil")
+			}
+			v, ok := m.(*[]byte)
+			if !ok {
+				return fmt.Errorf("expected *[]byte instead of %T", m)
+			}
+			if v == nil {
+				return errors.New("target *[]byte may not be nil")
+			}
+			*v = append((*v)[:0], b...)
+			return nil
+		})
+		singletonReader = func(b []byte) ReaderFunc {
+			eof := false
+			return func() ([]byte, error) {
+				if eof {
+					panic("reader should only be called once")
+				}
+				eof = true
+				return b, io.EOF
+			}
+		}
+		errorReader = func(err error) ReaderFunc {
+			return func() ([]byte, error) { return nil, err }
+		}
+	)
+	for ti, tc := range []struct {
+		r        Reader
+		wants    []byte
+		wantsErr error
+	}{
+		{errorReader(ErrorBadSize), nil, ErrorBadSize},
+		{singletonReader(([]byte)("james")), ([]byte)("james"), io.EOF},
+	} {
+		var (
+			buf []byte
+			d   = NewDecoder(tc.r, byteCopy)
+			err = d.Decode(&buf)
+		)
+		if err != tc.wantsErr {
+			t.Errorf("test case %d failed: expected error %q instead of %q", ti, tc.wantsErr, err)
+		}
+		if !reflect.DeepEqual(buf, tc.wants) {
+			t.Errorf("test case %d failed: expected %#v instead of %#v", tc.wants, buf)
+		}
+	}
+}

api/v1/lib/encoding/framing/framing.go

@@ -10,6 +10,16 @@ const (
 	ErrorOversizedFrame = Error("oversized frame, max size exceeded")
 )
 
-type Reader interface {
-	ReadFrame() (frame []byte, err error)
-}
+type (
+	// Reader generates data frames from some source, returning io.EOF with the final frame.
+	Reader interface {
+		ReadFrame() (frame []byte, err error)
+	}
+
+	// ReaderFunc is the functional adaptation of Reader
+	ReaderFunc func() ([]byte, error)
+)
+
+func (f ReaderFunc) ReadFrame() ([]byte, error) { return f() }
+
+var _ = Reader(ReaderFunc(nil))