Create bufferedValueReader and streamingValueReader

Author: prestonvasquez

bson/buffered_value_reader.go

@@ -0,0 +1,111 @@
+// Copyright (C) MongoDB, Inc. 2025-present.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License. You may obtain
+// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+
+package bson
+
+import (
+	"bytes"
+	"io"
+)
+
+// bufferedValueReader implements the low-level byteSrc interface by reading
+// directly from an in-memory byte slice. It provides efficient, zero-copy
+// access for parsing BSON when the entire document is buffered in memory.
+type bufferedValueReader struct {
+	buf    []byte // entire BSON document
+	offset int64  // Current read index into buf
+}
+
+var _ valueReaderByteSrc = (*bufferedValueReader)(nil)
+
+// Read reads up to len(p) bytes from buf[offset:] into p, advancing offset.
+//
+// TODO: Depending on streaming requirements, we might be able to remove this.
+func (b *bufferedValueReader) Read([]byte) (n int, err error) {
+	panic("Read not implemented for bufferedValueReader")
+}
+
+// ReadByte returns the single byte at buf[offset] and advances offset by 1.
+func (b *bufferedValueReader) ReadByte() (byte, error) {
+	if b.offset >= int64(len(b.buf)) {
+		return 0, io.EOF
+	}
+	b.offset++
+	return b.buf[b.offset-1], nil
+}
+
+// peek returns buf[offset:offset+n] without advancing offset.
+func (b *bufferedValueReader) peek(n int) ([]byte, error) {
+	// Ensure we don't read past the end of the buffer.
+	if int64(n)+b.offset > int64(len(b.buf)) {
+		return nil, io.EOF
+	}
+
+	// Return the next n bytes without advancing the offset
+	return b.buf[b.offset : b.offset+int64(n)], nil
+}
+
+// discard advances offset by n bytes, returning the number of bytes discarded.
+func (b *bufferedValueReader) discard(n int) (int, error) {
+	// Ensure we don't read past the end of the buffer.
+	if int64(n)+b.offset > int64(len(b.buf)) {
+		return 0, io.EOF
+	}
+
+	// Advance the read position
+	b.offset += int64(n)
+	return n, nil
+}
+
+// readSlice scans buf[offset:] for the first occurrence of delim, returns
+// buf[offset:idx+1], and advances offset past it; errors if delim not found.
+func (b *bufferedValueReader) readSlice(delim byte) ([]byte, error) {
+	// Ensure we don't read past the end of the buffer.
+	if b.offset >= int64(len(b.buf)) {
+		return nil, io.EOF
+	}
+
+	// Look for the delimiter in the remaining bytes
+	rem := b.buf[b.offset:]
+	idx := bytes.IndexByte(rem, delim)
+	if idx < 0 {
+		return nil, io.EOF
+	}
+
+	// Build the result slice up through the delimiter.
+	result := rem[:idx+1]
+
+	// Advance the offset past the delimiter.
+	b.offset += int64(idx + 1)
+
+	return result, nil
+}
+
+// pos returns the current read position in the buffer.
+func (b *bufferedValueReader) pos() int64 {
+	return b.offset
+}
+
+// regexLength will return the total byte length of a BSON regex value.
+func (b *bufferedValueReader) regexLength() (int32, error) {
+	rem := b.buf[b.offset:]
+
+	// Find end of the first C-string (pattern).
+	i := bytes.IndexByte(rem, 0x00)
+	if i < 0 {
+		return 0, io.EOF
+	}
+
+	// Find end of second C-string (options).
+	j := bytes.IndexByte(rem[i+1:], 0x00)
+	if j < 0 {
+		return 0, io.EOF
+	}
+
+	// Total length = first C-string length (pattern) + second C-string length
+	// (options) + 2 null terminators
+	return int32(i + j + 2), nil
+}

bson/streaming_value_reader.go

@@ -0,0 +1,57 @@
+// Copyright (C) MongoDB, Inc. 2025-present.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License. You may obtain
+// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+
+package bson
+
+import "bufio"
+
+// streamingValueReader reads from an ioReader wrapped in a bufio.Reader. It
+// first reads the BSON length header, then ensures it only ever reads exactly
+// that many bytes.
+//
+// Note: this approach trades memory usage for extra buffering and reader calls,
+// so it is less performanted than the in-memory bufferedValueReader.
+type streamingValueReader struct {
+	br *bufio.Reader
+}
+
+var _ valueReaderByteSrc = (*streamingValueReader)(nil)
+
+// Read reads up to len(p) bytes from buf[offset:] into p, advancing offset.
+func (b *streamingValueReader) Read([]byte) (n int, err error) {
+	panic("Read not implemented for streamingValueReader")
+}
+
+// ReadByte returns the single byte at buf[offset] and advances offset by 1.
+func (b *streamingValueReader) ReadByte() (byte, error) {
+	panic("ReadByte not implemented for streamingValueReader")
+}
+
+// peek returns buf[offset:offset+n] without advancing offset.
+func (b *streamingValueReader) peek(int) ([]byte, error) {
+	panic("peek not implemented for streamingValueReader")
+}
+
+// discard advances offset by n bytes, returning the number of bytes discarded.
+func (b *streamingValueReader) discard(int) (int, error) {
+	panic("discard not implemented for streamingValueReader")
+}
+
+// readSlice scans buf[offset:] for the first occurrence of delim, returns
+// buf[offset:idx+1], and advances offset past it; errors if delim not found.
+func (b *streamingValueReader) readSlice(byte) ([]byte, error) {
+	panic("readSlice not implemented for streamingValueReader")
+}
+
+// pos returns the current read position in the buffer.
+func (b *streamingValueReader) pos() int64 {
+	panic("pos not implemented for streamingValueReader")
+}
+
+// regexLength will return the total byte length of a BSON regex value.
+func (b *streamingValueReader) regexLength() (int32, error) {
+	panic("regexLength not implemented for streamingValueReader")
+}