object: removed dependency on jsonstream for parsing indirect objects

internal/jsonstream/doc.go

@@ -1,2 +0,0 @@
-// Package jsonstream implements streaming reader and writer for JSON objects.
-package jsonstream

internal/jsonstream/reader.go

@@ -1,112 +0,0 @@
-package jsonstream
-
-import (
-	"encoding/json"
-	"fmt"
-	"io"
-)
-
-// Reader reads a stream of JSON objects.
-type Reader struct {
-	decoder *json.Decoder
-	summary interface{}
-}
-
-// Read reads the next JSON objects from the stream, returns io.EOF on the end of stream.
-func (r *Reader) Read(v interface{}) error {
-	if r.decoder.More() {
-		return r.decoder.Decode(v)
-	}
-
-	if err := ensureDelimiter(r.decoder, json.Delim(']')); err != nil {
-		return invalidStreamFormatError(err)
-	}
-
-	tok, err := r.decoder.Token()
-	if err != nil {
-		return invalidStreamFormatError(err)
-	}
-
-	switch tok {
-	case json.Delim('}'):
-		// end of stream, all good
-		return io.EOF
-
-	case "summary":
-		s := r.summary
-		if s == nil {
-			s = map[string]interface{}{}
-		}
-		if err := r.decoder.Decode(s); err != nil {
-			return invalidStreamFormatError(err)
-		}
-	}
-
-	if err := ensureDelimiter(r.decoder, json.Delim('}')); err != nil {
-		return invalidStreamFormatError(err)
-	}
-
-	return io.EOF
-}
-
-func ensureDelimiter(d *json.Decoder, expected json.Delim) error {
-	t, err := d.Token()
-	if err != nil {
-		return err
-	}
-
-	if t != expected {
-		return fmt.Errorf("expected '%v', got %v", expected.String(), t)
-	}
-
-	return nil
-}
-func ensureStringToken(d *json.Decoder, expected string) error {
-	t, err := d.Token()
-	if err != nil {
-		return err
-	}
-
-	if s, ok := t.(string); ok {
-		if s == expected {
-			return nil
-		}
-	}
-
-	return fmt.Errorf("expected '%v', got '%v'", expected, t)
-}
-
-func invalidStreamFormatError(cause error) error {
-	return fmt.Errorf("invalid stream format: %v", cause)
-}
-
-// NewReader returns new Reader on top of a given buffered reader.
-// The provided header must match the beginning of a stream.
-func NewReader(r io.Reader, header string, summary interface{}) (*Reader, error) {
-	dr := Reader{
-		decoder: json.NewDecoder(r),
-		summary: summary,
-	}
-
-	if err := ensureDelimiter(dr.decoder, json.Delim('{')); err != nil {
-		return nil, invalidStreamFormatError(err)
-	}
-
-	if err := ensureStringToken(dr.decoder, "stream"); err != nil {
-		return nil, invalidStreamFormatError(err)
-	}
-
-	if err := ensureStringToken(dr.decoder, header); err != nil {
-		return nil, invalidStreamFormatError(err)
-	}
-
-	if err := ensureStringToken(dr.decoder, "entries"); err != nil {
-		return nil, invalidStreamFormatError(err)
-	}
-
-	if err := ensureDelimiter(dr.decoder, json.Delim('[')); err != nil {
-		return nil, invalidStreamFormatError(err)
-	}
-
-	return &dr, nil
-}

internal/jsonstream/stream_test.go

@@ -1,163 +0,0 @@
-package jsonstream
-
-import (
-	"bufio"
-	"bytes"
-	"io"
-	"log"
-	"strings"
-	"testing"
-)
-
-type TestObj struct {
-	Name string `json:"name,omitempty"`
-}
-type TestSummary struct {
-	Value int `json:"val"`
-}
-
-var testHeader1 = "01234567"
-var testHeader2 = "0123456x"
-
-func TestStream(t *testing.T) {
-	var buf bytes.Buffer
-
-	data := []TestObj{
-		{Name: "foo"},
-		{Name: "bar"},
-		{Name: "baz"},
-	}
-
-	w := NewWriter(&buf, testHeader1)
-	for _, d := range data {
-		if err := w.Write(&d); err != nil {
-			t.Errorf("write error: %v", err)
-		}
-	}
-	w.Finalize()
-	log.Printf("wrote: %v", buf.String())
-	r, err := NewReader(bufio.NewReader(&buf), testHeader1, nil)
-	if err != nil {
-		t.Errorf("err: %v", err)
-		return
-	}
-	for _, d := range data {
-		v := &TestObj{}
-		if readerr := r.Read(v); readerr != nil {
-			t.Errorf("read error: %v", readerr)
-		}
-		if v.Name != d.Name {
-			t.Errorf("invalid value: '%v', expected '%v'", v.Name, d.Name)
-		}
-	}
-	v := &TestObj{}
-	err = r.Read(v)
-	if err != io.EOF {
-		t.Errorf("expected EOF, got %v", err)
-	}
-}
-
-func TestStreamWithSummary(t *testing.T) {
-	var buf bytes.Buffer
-
-	data := []TestObj{
-		{Name: "foo"},
-		{Name: "bar"},
-		{Name: "baz"},
-	}
-
-	w := NewWriter(&buf, testHeader1)
-	for _, d := range data {
-		if err := w.Write(&d); err != nil {
-			t.Errorf("write error: %v", err)
-		}
-	}
-	w.FinalizeWithSummary(TestSummary{Value: 123})
-	log.Printf("wrote: %v", buf.String())
-
-	var summary TestSummary
-	r, err := NewReader(bufio.NewReader(&buf), testHeader1, &summary)
-	if err != nil {
-		t.Errorf("err: %v", err)
-		return
-	}
-	for _, d := range data {
-		v := &TestObj{}
-		if readerr := r.Read(v); readerr != nil {
-			t.Errorf("read error: %v", readerr)
-		}
-		if v.Name != d.Name {
-			t.Errorf("invalid value: '%v', expected '%v'", v.Name, d.Name)
-		}
-	}
-	v := &TestObj{}
-	err = r.Read(v)
-	if err != io.EOF {
-		t.Errorf("expected EOF, got %v", err)
-	}
-	if got, want := summary.Value, 123; got != want {
-		t.Errorf("unexpected summary value: %v, wanted %v", got, want)
-	}
-}
-
-func TestInvalidHeader(t *testing.T) {
-	var buf bytes.Buffer
-
-	w := NewWriter(&buf, testHeader1)
-	if err := w.Write(&TestObj{Name: "foo"}); err != nil {
-		t.Errorf("write error: %v", err)
-	}
-
-	_, err := NewReader(bufio.NewReader(&buf), testHeader2, nil)
-	if err == nil {
-		t.Errorf("expected error, got none")
-	} else if !strings.Contains(err.Error(), "invalid stream format") {
-		t.Errorf("got incorrect error: %v", err)
-	}
-}
-
-func TestInvalidStream(t *testing.T) {
-	cases := []string{
-		`x`,
-		`{}`,
-		`{"not-stream":"hdr"}`,
-		`{{}}`,
-		`{"stream":"non-hdr"}`,
-		`{"stream":"hdr","nonEntries":[]}`,
-		`{"stream":"hdr","entries":{}}`,
-		`{"stream":"hdr","entries":[]}`,
-		`{"stream":"hdr","entries":[`,
-		`{"stream":"hdr","entries":[}`,
-		`{"stream":"hdr","entries":[]`,
-		`{"stream":"hdr","entries":[],"summary"`,
-		`{"stream":"hdr","entries":[],1.222.33`,
-		`{"stream":"hdr","entries":[],"sxummary":{"x":"1",`,
-	}
-
-	for _, tc := range cases {
-		r, err := NewReader(strings.NewReader(tc), "hdr", nil)
-		if err != nil {
-			if !isInvalidStream(err) {
-				t.Errorf("got invalid error when creating reader: %v", err)
-			}
-			continue
-		}
-
-		for {
-			v := map[string]interface{}{}
-			if err := r.Read(v); err != nil {
-				if err == io.EOF {
-					break
-				}
-				if !isInvalidStream(err) {
-					t.Errorf("got invalid error when creating reader: %v", err)
-				}
-				break
-			}
-		}
-	}
-}
-
-func isInvalidStream(e error) bool {
-	return e != nil && strings.Contains(e.Error(), "invalid stream format")
-}

internal/jsonstream/writer.go

@@ -1,59 +0,0 @@
-package jsonstream
-
-import (
-	"encoding/json"
-	"fmt"
-	"io"
-)
-
-var commaBytes = []byte(",\n")
-
-// Writer writes a stream of JSON objects.
-type Writer struct {
-	output    io.Writer
-	header    string
-	separator []byte
-}
-
-// Write JSON object to the output.
-func (w *Writer) Write(v interface{}) error {
-	if _, err := w.output.Write(w.separator); err != nil {
-		return err
-	}
-	j, err := json.Marshal(v)
-	if err != nil {
-		return err
-	}
-	// log.Printf("*** %v: %v", w.header, string(j))
-	if _, err := w.output.Write(j); err != nil {
-		return err
-	}
-	w.separator = commaBytes
-
-	return nil
-}
-
-// FinalizeWithSummary writes the postamble to the JSON stream with a given summary object.
-func (w *Writer) FinalizeWithSummary(summary interface{}) error {
-	b, err := json.Marshal(summary)
-	if err != nil {
-		return err
-	}
-	_, err = fmt.Fprintf(w.output, "\n],\"summary\":%v}", string(b))
-	return err
-}
-
-// Finalize writes the postamble to the JSON stream.
-func (w *Writer) Finalize() error {
-	_, err := fmt.Fprintf(w.output, "\n]}")
-	return err
-}
-
-// NewWriter creates a new Writer on top of a specified writer with a specified optional header.
-func NewWriter(w io.Writer, header string) *Writer {
-	fmt.Fprintf(w, "{\"stream\":\"%v\",\"entries\":[\n", header) //nolint:errcheck
-	return &Writer{
-		header: header,
-		output: w,
-	}
-}

object/object_manager.go

@@ -2,14 +2,13 @@
 package object
 
 import (
-	"bufio"
 	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
 	"io"
 
 	"github.com/kopia/repo/block"
-	"github.com/kopia/repo/internal/jsonstream"
 )
 
 // Reader allows reading, seeking, getting the length of and closing of a repository object.
@@ -184,29 +183,31 @@ func NewObjectManager(ctx context.Context, bm blockManager, f Format, opts Manag
 	return om, nil
 }
 
+/*
+
+{"stream":"kopia:indirect","entries":[
+{"l":1698099,"o":"D13ea27f9ad891ad4a2edfa983906863d"},
+{"s":1698099,"l":1302081,"o":"De8ca8327cd3af5f4edbd5ed1009c525e"},
+{"s":3000180,"l":4352499,"o":"D6b6eb48ca5361d06d72fe193813e42e1"},
+{"s":7352679,"l":1170821,"o":"Dd14653f76b63802ed48be64a0e67fea9"},
+
+{"s":91094118,"l":1645153,"o":"Daa55df764d881a1daadb5ea9de17abbb"}
+]}
+*/
+
+type indirectObject struct {
+	StreamID string                `json:"stream"`
+	Entries  []indirectObjectEntry `json:"entries"`
+}
+
 func (om *Manager) flattenListChunk(rawReader io.Reader) ([]indirectObjectEntry, error) {
-	pr, err := jsonstream.NewReader(bufio.NewReader(rawReader), indirectStreamType, nil)
-	if err != nil {
-		return nil, err
-	}
-	var seekTable []indirectObjectEntry
+	var ind indirectObject
 
-	for {
-		var oe indirectObjectEntry
-
-		err := pr.Read(&oe)
-		if err == io.EOF {
-			break
-		}
-
-		if err != nil {
-			return nil, fmt.Errorf("failed to read indirect object: %v", err)
-		}
-
-		seekTable = append(seekTable, oe)
+	if err := json.NewDecoder(rawReader).Decode(&ind); err != nil {
+		return nil, fmt.Errorf("invalid indirect object: %v", err)
 	}
 
-	return seekTable, nil
+	return ind.Entries, nil
 }
 
 func (om *Manager) newRawReader(ctx context.Context, objectID ID) (Reader, error) {

object/object_manager_test.go

@@ -6,8 +6,8 @@ import (
 	cryptorand "crypto/rand"
 	"crypto/sha256"
 	"encoding/hex"
+	"encoding/json"
 	"fmt"
-	"io"
 	"io/ioutil"
 	"math/rand"
 	"reflect"
@@ -16,7 +16,6 @@ import (
 	"testing"
 
 	"github.com/kopia/repo/block"
-	"github.com/kopia/repo/internal/jsonstream"
 	"github.com/kopia/repo/storage"
 )
 
@@ -149,20 +148,9 @@ func verifyIndirectBlock(ctx context.Context, t *testing.T, r *Manager, oid ID)
 		}
 		defer rd.Close()
 
-		pr, err := jsonstream.NewReader(rd, indirectStreamType, nil)
-		if err != nil {
-			t.Errorf("cannot open indirect stream: %v", err)
-			return
-		}
-		for {
-			v := indirectObjectEntry{}
-			if err := pr.Read(&v); err != nil {
-				if err == io.EOF {
-					break
-				}
-				t.Errorf("err: %v", err)
-				break
-			}
+		var ind indirectObject
+		if err := json.NewDecoder(rd).Decode(&ind); err != nil {
+			t.Errorf("cannot parse indirect stream: %v", err)
 		}
 	}
 }

object/object_writer.go

@@ -3,11 +3,10 @@ package object
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
 	"io"
 	"sync"
-
-	"github.com/kopia/repo/internal/jsonstream"
 )
 
 // Writer allows writing content to the storage and supports automatic deduplication and encryption
@@ -123,14 +122,13 @@ func (w *objectWriter) Result() (ID, error) {
 		prefix:      w.prefix,
 	}
 
-	jw := jsonstream.NewWriter(iw, indirectStreamType)
-	for _, e := range w.blockIndex {
-		if err := jw.Write(&e); err != nil {
-			return "", fmt.Errorf("unable to write indirect block index: %v", err)
-		}
+	ind := indirectObject{
+		StreamID: "kopia:indirect",
+		Entries:  w.blockIndex,
 	}
-	if err := jw.Finalize(); err != nil {
-		return "", fmt.Errorf("unable to finalize indirect block index: %v", err)
+
+	if err := json.NewEncoder(iw).Encode(ind); err != nil {
+		return "", fmt.Errorf("unable to write indirect block index: %v", err)
 	}
 	oid, err := iw.Result()
 	if err != nil {