added support for summaries to JSON stream

internal/dir/dir_reader.go

@@ -11,7 +11,7 @@ var directoryStreamType = "kopia:directory"
 
 // ReadEntries reads all the Entry from the specified reader.
 func ReadEntries(r io.Reader) ([]*Entry, error) {
-	psr, err := jsonstream.NewReader(bufio.NewReader(r), directoryStreamType)
+	psr, err := jsonstream.NewReader(bufio.NewReader(r), directoryStreamType, nil)
 	if err != nil {
 		return nil, err
 	}

internal/hashcache/hashcache_reader.go

@@ -100,7 +100,7 @@ func Open(r io.Reader) Reader {
 		return &nullReader{}
 	}
 
-	jsr, err := jsonstream.NewReader(bufio.NewReader(r), hashCacheStreamType)
+	jsr, err := jsonstream.NewReader(bufio.NewReader(r), hashCacheStreamType, nil)
 	if err != nil {
 		return &nullReader{}
 	}

internal/jsonstream/reader.go

@@ -9,6 +9,7 @@ import (
 // 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.
@@ -21,6 +22,26 @@ func (r *Reader) Read(v interface{}) error {
 		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)
 	}
@@ -61,9 +82,10 @@ func invalidStreamFormatError(cause error) error {
 
 // 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) (*Reader, error) {
+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 {

internal/jsonstream/stream_test.go

@@ -13,6 +13,9 @@ import (
 type TestObj struct {
 	Name string `json:"name,omitempty"`
 }
+type TestSummary struct {
+	Value int `json:"val"`
+}
 
 var testHeader1 = "01234567"
 var testHeader2 = "0123456x"
@@ -34,7 +37,7 @@ func TestStream(t *testing.T) {
 	}
 	w.Finalize()
 	log.Printf("wrote: %v", buf.String())
-	r, err := NewReader(bufio.NewReader(&buf), testHeader1)
+	r, err := NewReader(bufio.NewReader(&buf), testHeader1, nil)
 	if err != nil {
 		t.Errorf("err: %v", err)
 		return
@@ -55,6 +58,49 @@ func TestStream(t *testing.T) {
 	}
 }
 
+func TestStreamWithSummary(t *testing.T) {
+	var buf bytes.Buffer
+
+	data := []TestObj{
+		TestObj{Name: "foo"},
+		TestObj{Name: "bar"},
+		TestObj{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
 
@@ -63,7 +109,7 @@ func TestInvalidHeader(t *testing.T) {
 		t.Errorf("write error: %v", err)
 	}
 
-	_, err := NewReader(bufio.NewReader(&buf), testHeader2)
+	_, 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") {

internal/jsonstream/writer.go

@@ -33,6 +33,16 @@ func (w *Writer) Write(v interface{}) error {
 	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
+	}
+	fmt.Fprintf(w.output, "\n],\"summary\":%v}", string(b))
+	return nil
+}
+
 // Finalize writes the postamble to the JSON stream.
 func (w *Writer) Finalize() error {
 	fmt.Fprintf(w.output, "\n]}")

object/object_manager.go

@@ -235,7 +235,7 @@ func NewObjectManager(bm blockManager, f config.RepositoryObjectFormat, opts Man
 }
 
 func (om *Manager) flattenListChunk(rawReader io.Reader) ([]indirectObjectEntry, error) {
-	pr, err := jsonstream.NewReader(bufio.NewReader(rawReader), indirectStreamType)
+	pr, err := jsonstream.NewReader(bufio.NewReader(rawReader), indirectStreamType, nil)
 	if err != nil {
 		return nil, err
 	}

object/object_manager_test.go

@@ -156,7 +156,7 @@ func verifyIndirectBlock(t *testing.T, r *Manager, oid ID) {
 		}
 		defer rd.Close()
 
-		pr, err := jsonstream.NewReader(rd, indirectStreamType)
+		pr, err := jsonstream.NewReader(rd, indirectStreamType, nil)
 		if err != nil {
 			t.Errorf("cannot open indirect stream: %v", err)
 			return