Extend valueReader tests for both streaming and buffered

Author: prestonvasquez

bson/buffered_value_reader.go

@@ -23,7 +23,7 @@ var _ valueReaderByteSrc = (*bufferedValueReader)(nil)
 
 // Read reads up to len(p) bytes from the in-memory buffer, advancing the offset
 // by the number of bytes read.
-func (b *bufferedValueReader) Read(p []byte) (int, error) {
+func (b *bufferedValueReader) readExact(p []byte) (int, error) {
 	if b.offset >= int64(len(b.buf)) {
 		return 0, io.EOF
 	}

bson/streaming_value_reader.go

@@ -8,6 +8,7 @@ package bson
 
 import (
 	"bufio"
+	"io"
 )
 
 // streamingValueReader reads from an ioReader wrapped in a bufio.Reader. It
@@ -25,9 +26,12 @@ var _ valueReaderByteSrc = (*streamingValueReader)(nil)
 
 // Read reads up to len(p) bytes from the underlying bufio.Reader, advancing
 // the offset by the number of bytes read.
-func (s *streamingValueReader) Read(p []byte) (int, error) {
-	n, err := s.br.Read(p)
-	s.offset += int64(n)
+func (s *streamingValueReader) readExact(p []byte) (int, error) {
+	n, err := io.ReadFull(s.br, p)
+	if err == nil {
+		s.offset += int64(n)
+	}
+
 	return n, err
 }
 

bson/value_reader.go

@@ -17,9 +17,10 @@ import (
 )
 
 type valueReaderByteSrc interface {
-	io.Reader
 	io.ByteReader
 
+	readExact(p []byte) (int, error)
+
 	// Peek returns the next n bytes without advancing the cursor. It must return
 	// exactly n bytes or n error if fewer are available.
 	peek(n int) ([]byte, error)
@@ -61,12 +62,6 @@ type vrState struct {
 	end   int64
 }
 
-var bufioReaderPool = sync.Pool{
-	New: func() interface{} {
-		return bufio.NewReader(nil)
-	},
-}
-
 var vrPool = sync.Pool{
 	New: func() interface{} {
 		return &valueReader{
@@ -286,10 +281,7 @@ func peekNextValueSize(vr *valueReader) (int32, error) {
 func readBytes(src valueReaderByteSrc, n int) ([]byte, error) {
 	if src.streamable() {
 		data := make([]byte, n)
-		if _, err := io.ReadFull(src, data); err != nil {
-			if errors.Is(err, io.ErrUnexpectedEOF) {
-				err = io.EOF // Convert io.ErrUnexpectedEOF to io.EOF for consistency.
-			}
+		if _, err := src.readExact(data); err != nil {
 			return nil, err
 		}
 
@@ -315,6 +307,7 @@ func (vr *valueReader) readBytes(n int32) ([]byte, error) {
 	return readBytes(vr.src, int(n))
 }
 
+//nolint:unparam
 func (vr *valueReader) readValueBytes(dst []byte) (Type, []byte, error) {
 	switch vr.stack[vr.frame].mode {
 	case mTopLevel: