jsontext: proof-of-concept ReadStringAsStream

This is an idea for making it possible to have explicit control over
what happens when reading strings. For example, an implementation
could choose to decode on the fly or discard large strings but still
keep going without using a large amount of memory.

We include a somewhat involved example demonstrating some possibilities.

Signed-off-by: Roger Peppe <[email protected]>

Author: rogpeppe

bench_test.go

@@ -29,12 +29,13 @@ import (
 var benchVersion = cmp.Or(os.Getenv("BENCHMARK_VERSION"), "v2")
 
 var jsonFuncs = func() (funcs struct {
-	marshal      func(any) ([]byte, error)
-	unmarshal    func([]byte, any) error
-	encodeValue  func(w io.Writer, b []byte) error
-	encodeTokens func(w io.Writer, toks []jsontext.Token) error
-	decodeValue  func(r io.Reader) error
-	decodeTokens func(r io.Reader) error
+	marshal            func(any) ([]byte, error)
+	unmarshal          func([]byte, any) error
+	encodeValue        func(w io.Writer, b []byte) error
+	encodeTokens       func(w io.Writer, toks []jsontext.Token) error
+	decodeValue        func(r io.Reader) error
+	decodeTokens       func(r io.Reader) error
+	decodeTokensStream func(r io.Reader) error
 }) {
 	ignoreEOF := func(err error) error {
 		if err == io.EOF {
@@ -62,6 +63,8 @@ var jsonFuncs = func() (funcs struct {
 				}
 			}
 		}
+		funcs.decodeTokensStream = funcs.decodeTokens
+
 	case "v1in2":
 		funcs.marshal = jsonv1in2.Marshal
 		funcs.unmarshal = jsonv1in2.Unmarshal
@@ -80,6 +83,7 @@ var jsonFuncs = func() (funcs struct {
 				}
 			}
 		}
+		funcs.decodeTokensStream = funcs.decodeTokens
 	case "v2":
 		funcs.marshal = func(v any) ([]byte, error) { return jsonv2.Marshal(v) }
 		funcs.unmarshal = func(b []byte, v any) error { return jsonv2.Unmarshal(b, v) }
@@ -107,6 +111,22 @@ var jsonFuncs = func() (funcs struct {
 				}
 			}
 		}
+		funcs.decodeTokensStream = func(r io.Reader) error {
+			d := jsontext.NewDecoder(r, jsontext.AllowDuplicateNames(true))
+			for {
+				if d.PeekKind() == '"' {
+					for _, err := range d.ReadStringAsSeq() {
+						if err != nil {
+							return err
+						}
+					}
+					return nil
+				}
+				if _, err := d.ReadToken(); err != nil {
+					return ignoreEOF(err)
+				}
+			}
+		}
 	default:
 		panic("unknown version: " + benchVersion)
 	}
@@ -413,11 +433,14 @@ func runAllTestdata(tb testing.TB) {
 				})
 			}
 		}
-
 		tokens := mustDecodeTokens(tb, td.Data())
 		buffer := make([]byte, 0, 2*len(td.Data()))
 		for _, codeName := range []string{"Encode", "Decode"} {
-			for _, typeName := range []string{"Token", "Value"} {
+			for _, typeName := range []string{"Token", "TokenStream", "Value"} {
+				if codeName == "Encode" && typeName == "TokenStream" {
+					// no token stream for encoding.
+					continue
+				}
 				for _, modeName := range []string{"Streaming", "Buffered"} {
 					name := path.Join(td.Name, codeName, typeName, modeName)
 					runTestOrBench(tb, name, int64(td.Size), func(tb testing.TB) {
@@ -531,6 +554,10 @@ func runDecode(t testing.TB, typeName, modeName string, buffer, data []byte, tok
 		if err := jsonFuncs.decodeTokens(r); err != nil {
 			t.Fatalf("Decoder.ReadToken error: %v", err)
 		}
+	case "TokenStream":
+		if err := jsonFuncs.decodeTokensStream(r); err != nil {
+			t.Fatalf("Decoder.ReadTokenStreamv error: %v", err)
+		}
 	case "Value":
 		if err := jsonFuncs.decodeValue(r); err != nil {
 			t.Fatalf("Decoder.ReadValue error: %v", err)
@@ -555,7 +582,7 @@ func BenchmarkSlowStreamingDecode(b *testing.B)     { runAllSlowStreamingDecode(
 
 func runAllSlowStreamingDecode(tb testing.TB) {
 	for _, td := range slowStreamingDecodeTestdata {
-		for _, typeName := range []string{"Token", "Value"} {
+		for _, typeName := range []string{"Token", "TokenStream", "Value"} {
 			name := path.Join(td.name, typeName)
 			runTestOrBench(tb, name, len64(td.data), func(tb testing.TB) {
 				runSlowStreamingDecode(tb, typeName, td.data)
@@ -570,6 +597,10 @@ func runAllSlowStreamingDecode(tb testing.TB) {
 func runSlowStreamingDecode(t testing.TB, typeName string, data []byte) {
 	r := iotest.OneByteReader(bytes.NewReader(data))
 	switch typeName {
+	case "TokenStream":
+		if err := jsonFuncs.decodeTokensStream(r); err != nil {
+			t.Fatalf("Decoder.ReadToken error: %v", err)
+		}
 	case "Token":
 		if err := jsonFuncs.decodeTokens(r); err != nil {
 			t.Fatalf("Decoder.ReadToken error: %v", err)

go.mod

@@ -4,3 +4,5 @@
 module github.com/go-json-experiment/json
 
 go 1.24
+
+require github.com/rogpeppe/ioseq v0.0.0-20250405133649-1cc137273d25

go.sum

@@ -0,0 +1,2 @@
+github.com/rogpeppe/ioseq v0.0.0-20250405133649-1cc137273d25 h1:HikF06tRfzHScFVzW7N1v+BDqcv4e34b5SbEMZnKP10=
+github.com/rogpeppe/ioseq v0.0.0-20250405133649-1cc137273d25/go.mod h1:Rs8CeC6rysLMoBy6MSLofmizxZm7lg52K8Wu2AYHVpA=

internal/jsonwire/decode.go

@@ -93,16 +93,28 @@ func ConsumeLiteral(b []byte, lit string) (n int, err error) {
 // non-zero then we know that the string would be encoded the same way
 // under both v1 or v2 escape semantics.
 func ConsumeSimpleString(b []byte) (n int) {
+	return ConsumeSimpleStringPartial(b, true)
+}
+
+// ConsumeSimpleStringPartial is like ConsumeSimpleString but can
+// also work when not at the start of a string, indicated by atStart.
+func ConsumeSimpleStringPartial(b []byte, atStart bool) (n int) {
 	// NOTE: The arguments and logic are kept simple to keep this inlinable.
-	if len(b) > 0 && b[0] == '"' {
-		n++
-		for len(b) > n && b[n] < utf8.RuneSelf && escapeASCII[b[n]] == 0 {
-			n++
-		}
-		if uint(len(b)) > uint(n) && b[n] == '"' {
-			n++
-			return n
+	if len(b) == 0 {
+		return 0
+	}
+	if atStart {
+		if b[0] != '"' {
+			return 0
 		}
+		n++
+	}
+	for len(b) > n && b[n] < utf8.RuneSelf && escapeASCII[b[n]] == 0 {
+		n++
+	}
+	if uint(len(b)) > uint(n) && b[n] == '"' {
+		n++
+		return n
 	}
 	return 0
 }
@@ -116,12 +128,18 @@ func ConsumeString(flags *ValueFlags, b []byte, validateUTF8 bool) (n int, err e
 	return ConsumeStringResumable(flags, b, 0, validateUTF8)
 }
 
-// ConsumeStringResumable is identical to consumeString but supports resuming
+// ConsumeStringResumable is identical to ConsumeString but supports resuming
 // from a previous call that returned io.ErrUnexpectedEOF.
 func ConsumeStringResumable(flags *ValueFlags, b []byte, resumeOffset int, validateUTF8 bool) (n int, err error) {
+	return ConsumeStringResumablePartial(flags, b, resumeOffset, true, validateUTF8)
+}
+
+// ConsumeStringResumable is identical to ConsumeStringResumable but also supports
+// starting midway through the string, indicated by atStart.
+func ConsumeStringResumablePartial(flags *ValueFlags, b []byte, resumeOffset int, atStart bool, validateUTF8 bool) (n int, err error) {
 	// Consume the leading double quote.
 	switch {
-	case resumeOffset > 0:
+	case !atStart:
 		n = resumeOffset // already handled the leading quote
 	case uint(len(b)) == 0:
 		return n, io.ErrUnexpectedEOF
@@ -248,18 +266,28 @@ func ConsumeStringResumable(flags *ValueFlags, b []byte, resumeOffset int, valid
 	return n, io.ErrUnexpectedEOF
 }
 
-// AppendUnquote appends the unescaped form of a JSON string in src to dst.
+// AppendUnquotePartial appends the unescaped form of a JSON string in src to dst.
 // Any invalid UTF-8 within the string will be replaced with utf8.RuneError,
 // but the error will be specified as having encountered such an error.
 // The input must be an entire JSON string with no surrounding whitespace.
+//
+// It returns [io.ErrUnexpectedEOF] if the source terminates before
+// the final quote.
 func AppendUnquote[Bytes ~[]byte | ~string](dst []byte, src Bytes) (v []byte, err error) {
+	return AppendUnquotePartial(dst, src, true)
+}
+
+// AppendUnquotePartial is like AppendUnquote except that the string
+// need not start at the beginning (atStart indicates whether it does).
+func AppendUnquotePartial[Bytes ~[]byte | ~string](dst []byte, src Bytes, atStart bool) (v []byte, err error) {
 	dst = slices.Grow(dst, len(src))
 
 	// Consume the leading double quote.
 	var i, n int
 	switch {
 	case uint(len(src)) == 0:
 		return dst, io.ErrUnexpectedEOF
+	case !atStart:
 	case src[0] == '"':
 		i, n = 1, 1
 	default:
@@ -409,11 +437,26 @@ func hasEscapedUTF16Prefix[Bytes ~[]byte | ~string](b Bytes, lowerSurrogateHalf
 // Otherwise, a new buffer is allocated for the output.
 // It assumes the input is valid.
 func UnquoteMayCopy(b []byte, isVerbatim bool) []byte {
+	return UnquoteMayCopyPartial(b, true, isVerbatim)
+}
+
+// UnquoteMayCopyPartial is like UnquoteMayCopy except that
+// it can also be used to unquote a string that's not at the start.
+// The atStart argument indicates whether that'sthe case.
+func UnquoteMayCopyPartial(b []byte, atStart bool, isVerbatim bool) []byte {
 	// NOTE: The arguments and logic are kept simple to keep this inlinable.
 	if isVerbatim {
-		return b[len(`"`) : len(b)-len(`"`)]
+		if atStart {
+			b = b[len(`"`):]
+		}
+		// When unquoting a streaming string, the final quote may be
+		// omitted.
+		if len(b) > 0 && b[len(b)-1] == '"' {
+			b = b[:len(b)-len(`"`)]
+		}
+		return b
 	}
-	b, _ = AppendUnquote(nil, b)
+	b, _ = AppendUnquotePartial(nil, b, atStart)
 	return b
 }
 

internal/jsonwire/encode.go

@@ -61,10 +61,19 @@ func NeedEscape[Bytes ~[]byte | ~string](src Bytes) bool {
 // If no escape flags are set, then the shortest representable form is used,
 // which is also the canonical form for strings (RFC 8785, section 3.2.2.2).
 func AppendQuote[Bytes ~[]byte | ~string](dst []byte, src Bytes, flags *jsonflags.Flags) ([]byte, error) {
-	var i, n int
-	var hasInvalidUTF8 bool
 	dst = slices.Grow(dst, len(`"`)+len(src)+len(`"`))
 	dst = append(dst, '"')
+	dst, err := AppendQuotePartial(dst, src, flags)
+	dst = append(dst, '"')
+	return dst, err
+}
+
+// AppendQuotePartial is like AppendQuote except that it can also be used to encode content
+// within a string. If entire is false, then neither the starting or ending quotes will be appended.
+func AppendQuotePartial[Bytes ~[]byte | ~string](dst []byte, src Bytes, flags *jsonflags.Flags) ([]byte, error) {
+	dst = slices.Grow(dst, len(src))
+	var i, n int
+	var hasInvalidUTF8 bool
 	for uint(len(src)) > uint(n) {
 		if c := src[n]; c < utf8.RuneSelf {
 			// Handle single-byte ASCII.
@@ -104,7 +113,6 @@ func AppendQuote[Bytes ~[]byte | ~string](dst []byte, src Bytes, flags *jsonflag
 		}
 	}
 	dst = append(dst, src[i:n]...)
-	dst = append(dst, '"')
 	if hasInvalidUTF8 && !flags.Get(jsonflags.AllowInvalidUTF8) {
 		return dst, ErrInvalidUTF8
 	}

jsontext/decode.go

@@ -7,7 +7,9 @@ package jsontext
 import (
 	"bytes"
 	"errors"
+	"fmt"
 	"io"
+	"iter"
 
 	"github.com/go-json-experiment/json/internal/jsonflags"
 	"github.com/go-json-experiment/json/internal/jsonopts"
@@ -459,6 +461,101 @@ func (d *decoderState) SkipUntil(depth int, length int64) error {
 func (d *Decoder) ReadToken() (Token, error) {
 	return d.s.ReadToken()
 }
+
+// ReadStringAsSeq reads a string token as a sequence of
+// byte slices that are available as the contents of the string
+// are accumulated. If the next token is not a string, the sequence
+// immediately terminates with an error.
+//
+// The byte slices must not escape the iterator.
+//
+// It also returns an error if the AllowDuplicateNames option
+// is not set and the string is an object key.
+//
+// If the iterator terminates before completion, no
+// more tokens may be read from the decoder.
+// Any attempt to read tokens with the decoder while the iteration
+// is in progress will result in an error.
+func (dec *Decoder) ReadStringAsSeq() iter.Seq2[[]byte, error] {
+	return dec.s.ReadStringAsSeq()
+}
+
+var errNotString = fmt.Errorf("cannot stream non-string")
+var errCannotStreamObjectKeyWithoutAllowDuplicateNames = fmt.Errorf("cannot stream object key without enabling duplicate names")
+var errStringStreamAborted = fmt.Errorf("stream aborted")
+
+func (d *decoderState) ReadStringAsSeq() iter.Seq2[[]byte, error] {
+	return func(yield func([]byte, error) bool) {
+		if err := d.readStringAsSeq(yield); err != nil {
+			yield(nil, err)
+		}
+	}
+}
+
+func (d *decoderState) readStringAsSeq(yield func([]byte, error) bool) error {
+	if d.PeekKind() != '"' {
+		err := d.peekErr
+		if err == nil {
+			err = errNotString
+		}
+		return err
+	}
+	if d.Tokens.Last.NeedObjectName() && !d.Flags.Get(jsonflags.AllowDuplicateNames) {
+		return errCannotStreamObjectKeyWithoutAllowDuplicateNames
+	}
+	pos := d.peekPos
+	d.peekPos = 0
+	if err := d.Tokens.appendString(); err != nil {
+		return wrapSyntacticError(d, err, pos, +1) // report position at start of string
+	}
+	var flags jsonwire.ValueFlags
+	if n := jsonwire.ConsumeSimpleString(d.buf[pos:]); n > 0 {
+		if data := jsonwire.UnquoteMayCopy(d.buf[pos:pos+n], true); len(data) > 0 {
+			yield(data, nil)
+		}
+		return nil
+	}
+	atStart := true
+	for {
+		n, err := jsonwire.ConsumeStringResumablePartial(&flags, d.buf[pos:], 0, atStart, true)
+		if err == io.ErrUnexpectedEOF {
+			pos += n
+			if n > 0 {
+				if data := jsonwire.UnquoteMayCopyPartial(d.buf[pos-n:pos], atStart, flags.IsVerbatim()); len(data) > 0 {
+					if !yield(data, nil) {
+						d.peekPos = -1
+						d.peekErr = errStringStreamAborted
+						return nil
+					}
+				} else {
+				}
+			} else {
+			}
+			// Discard previous string content.
+			d.prevStart, d.prevEnd = pos, pos
+			absPos := d.baseOffset + int64(pos)
+			err := d.fetch()
+			pos = int(absPos - d.baseOffset)
+			if err != nil {
+				return err
+			}
+			atStart = false
+			continue
+		}
+		pos += n
+		// No need to set errStringStreamAborted because we know that
+		// we're in a good state to proceed and we're not going to call yield
+		// again.
+		if data := jsonwire.UnquoteMayCopyPartial(d.buf[pos-n:pos], atStart, flags.IsVerbatim()); len(data) > 0 {
+			yield(data, err)
+		}
+
+		// Discard previous string content.
+		d.prevStart, d.prevEnd = pos, pos
+		return err
+	}
+}
+
 func (d *decoderState) ReadToken() (Token, error) {
 	// Determine the next kind.
 	var err error