From e997e07b0c52008e3260d83d32446ada3d63282d Mon Sep 17 00:00:00 2001 From: Jarek Kowalski Date: Sat, 24 Mar 2018 09:07:59 -0700 Subject: [PATCH] added support for summaries to JSON stream --- internal/dir/dir_reader.go | 2 +- internal/hashcache/hashcache_reader.go | 2 +- internal/jsonstream/reader.go | 24 ++++++++++++- internal/jsonstream/stream_test.go | 50 ++++++++++++++++++++++++-- internal/jsonstream/writer.go | 10 ++++++ object/object_manager.go | 2 +- object/object_manager_test.go | 2 +- 7 files changed, 85 insertions(+), 7 deletions(-) diff --git a/internal/dir/dir_reader.go b/internal/dir/dir_reader.go index fa939629b..f1b6eeb19 100644 --- a/internal/dir/dir_reader.go +++ b/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 } diff --git a/internal/hashcache/hashcache_reader.go b/internal/hashcache/hashcache_reader.go index 3ae186844..55160b41a 100644 --- a/internal/hashcache/hashcache_reader.go +++ b/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{} } diff --git a/internal/jsonstream/reader.go b/internal/jsonstream/reader.go index 03512b6f0..c1ddb2fbc 100644 --- a/internal/jsonstream/reader.go +++ b/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 { diff --git a/internal/jsonstream/stream_test.go b/internal/jsonstream/stream_test.go index 6b3481fe4..3e1c5f00a 100644 --- a/internal/jsonstream/stream_test.go +++ b/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") { diff --git a/internal/jsonstream/writer.go b/internal/jsonstream/writer.go index 1fef488fb..bce961ded 100644 --- a/internal/jsonstream/writer.go +++ b/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]}") diff --git a/object/object_manager.go b/object/object_manager.go index d552bbeb0..1f8ce957e 100644 --- a/object/object_manager.go +++ b/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 } diff --git a/object/object_manager_test.go b/object/object_manager_test.go index f6e7ebfe4..1af1de421 100644 --- a/object/object_manager_test.go +++ b/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