diff --git a/internal/jsonstream/doc.go b/internal/jsonstream/doc.go deleted file mode 100644 index 157858468..000000000 --- a/internal/jsonstream/doc.go +++ /dev/null @@ -1,2 +0,0 @@ -// Package jsonstream implements streaming reader and writer for JSON objects. -package jsonstream diff --git a/internal/jsonstream/reader.go b/internal/jsonstream/reader.go deleted file mode 100644 index c1ddb2fbc..000000000 --- a/internal/jsonstream/reader.go +++ /dev/null @@ -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 -} diff --git a/internal/jsonstream/stream_test.go b/internal/jsonstream/stream_test.go deleted file mode 100644 index 6beff5f0c..000000000 --- a/internal/jsonstream/stream_test.go +++ /dev/null @@ -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") -} diff --git a/internal/jsonstream/writer.go b/internal/jsonstream/writer.go deleted file mode 100644 index a79049bca..000000000 --- a/internal/jsonstream/writer.go +++ /dev/null @@ -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, - } -} diff --git a/object/object_manager.go b/object/object_manager.go index 0ebe9538e..c205bd076 100644 --- a/object/object_manager.go +++ b/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) { diff --git a/object/object_manager_test.go b/object/object_manager_test.go index f2b712fb4..1bc65fa86 100644 --- a/object/object_manager_test.go +++ b/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) } } } diff --git a/object/object_writer.go b/object/object_writer.go index 4b40ec4f7..a0e7e8f5b 100644 --- a/object/object_writer.go +++ b/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 {