snapshot: removed depdendency on jsonstream from dir reader and writer

This commit is contained in:
Jarek Kowalski committed 2019-06-08 22:47:05 -07:00
1 parent 0d89ae3f79
commit c9da993e7b
4 files changed
+40 -78

No files matched your search

+9
View File
@@ -82,6 +82,15 @@ type DirEntry struct {
DirSummary *fs.DirectorySummary `json:"summ,omitempty"`
}
// DirManifest represents serialized contents of a directory.
// The entries are sorted lexicographically and summary only refers to properties of
// entries, so directory with the same contents always serializes to exactly the same JSON.
type DirManifest struct {
StreamType string `json:"stream"` // legacy
Entries []*DirEntry `json:"entries"`
Summary *fs.DirectorySummary `json:"summary"`
}
// RootObjectID returns the ID of a root object.
func (m *Manifest) RootObjectID() object.ID {
if m.RootEntry != nil {
+12 -21
View File
@@ -1,37 +1,28 @@
package snapshotfs
import (
"bufio"
"encoding/json"
"io"
"github.com/pkg/errors"
"github.com/kopia/kopia/fs"
"github.com/kopia/kopia/internal/jsonstream"
"github.com/kopia/kopia/snapshot"
)
var directoryStreamType = "kopia:directory"
// readDirEntries reads all the Entry from the specified reader.
// readDirEntries reads all directory entries from the specified reader.
func readDirEntries(r io.Reader) ([]*snapshot.DirEntry, *fs.DirectorySummary, error) {
var summ fs.DirectorySummary
psr, err := jsonstream.NewReader(bufio.NewReader(r), directoryStreamType, &summ)
if err != nil {
return nil, nil, err
}
var entries []*snapshot.DirEntry
for {
e := &snapshot.DirEntry{}
err := psr.Read(e)
if err == io.EOF {
break
}
var dir snapshot.DirManifest
if err != nil {
return nil, nil, err
}
entries = append(entries, e)
if err := json.NewDecoder(r).Decode(&dir); err != nil {
return nil, nil, errors.Wrap(err, "unable to parse directory object")
}
return entries, &summ, nil
if dir.StreamType != directoryStreamType {
return nil, nil, errors.Errorf("invalid directory stream type")
}
return dir.Entries, dir.Summary, nil
}
-37
View File
@@ -1,37 +0,0 @@
package snapshotfs
import (
"io"
"github.com/kopia/kopia/fs"
"github.com/kopia/kopia/internal/jsonstream"
"github.com/kopia/kopia/snapshot"
)
// dirWriter writes a stream of directory entries.
type dirWriter struct {
w *jsonstream.Writer
}
// WriteEntry writes the specified entry to the output.
func (dw *dirWriter) WriteEntry(e *snapshot.DirEntry) error {
if err := e.ObjectID.Validate(); err != nil {
panic("invalid object ID: " + err.Error())
}
return dw.w.Write(e)
}
// Finalize writes the trailing data to the output stream.
func (dw *dirWriter) Finalize(summ *fs.DirectorySummary) error {
return dw.w.FinalizeWithSummary(summ)
}
// newDirWriter returns new directoryWriter for with the specified output.
func newDirWriter(w io.Writer) *dirWriter {
dw := &dirWriter{
w: jsonstream.NewWriter(w, directoryStreamType),
}
return dw
}
+19 -20
View File
@@ -3,6 +3,7 @@ package snapshotfs
import (
"bytes"
"context"
"encoding/json"
"hash/fnv"
"io"
"os"
@@ -285,7 +286,7 @@ type entryResult struct {
de *snapshot.DirEntry
}
func (u *Uploader) processSubdirectories(ctx context.Context, relativePath string, entries fs.Entries, previousEntries []fs.Entries, dw *dirWriter, summ *fs.DirectorySummary) error {
func (u *Uploader) processSubdirectories(ctx context.Context, relativePath string, entries fs.Entries, previousEntries []fs.Entries, dirManifest *snapshot.DirManifest, summ *fs.DirectorySummary) error {
return u.foreachEntryUnlessCancelled(relativePath, entries, func(entry fs.Entry, entryRelativePath string) error {
dir, ok := entry.(fs.Directory)
if !ok {
@@ -324,10 +325,7 @@ func (u *Uploader) processSubdirectories(ctx context.Context, relativePath strin
}
de.DirSummary = &subdirsumm
if err := dw.WriteEntry(de); err != nil {
return errors.Wrap(err, "unable to write dir entry")
}
dirManifest.Entries = append(dirManifest.Entries, de)
return nil
})
}
@@ -510,7 +508,7 @@ func (u *Uploader) launchWorkItems(workItems []*uploadWorkItem, wg *sync.WaitGro
}
}
func (u *Uploader) processUploadWorkItems(workItems []*uploadWorkItem, dw *dirWriter) error {
func (u *Uploader) processUploadWorkItems(workItems []*uploadWorkItem, dirManifest *snapshot.DirManifest) error {
var wg sync.WaitGroup
u.launchWorkItems(workItems, &wg)
@@ -531,9 +529,7 @@ func (u *Uploader) processUploadWorkItems(workItems []*uploadWorkItem, dw *dirWr
return errors.Errorf("unable to process %q: %s", it.entryRelativePath, result.err)
}
if err := dw.WriteEntry(result.de); err != nil {
return errors.Wrap(err, "unable to write directory entry")
}
dirManifest.Entries = append(dirManifest.Entries, result.de)
}
// wait for workers, this is technically not needed, but let's make sure we don't leak goroutines
@@ -609,15 +605,11 @@ func uploadDirInternal(
summ.MaxModTime = directory.ModTime()
}
writer := u.repo.Objects.NewWriter(ctx, object.WriterOptions{
Description: "DIR:" + dirRelativePath,
Prefix: "k",
})
dirManifest := &snapshot.DirManifest{
StreamType: directoryStreamType,
}
dw := newDirWriter(writer)
defer writer.Close() //nolint:errcheck
if err := u.processSubdirectories(ctx, dirRelativePath, entries, prevEntries, dw, &summ); err != nil && err != errCancelled {
if err := u.processSubdirectories(ctx, dirRelativePath, entries, prevEntries, dirManifest, &summ); err != nil && err != errCancelled {
return "", fs.DirectorySummary{}, err
}
u.prepareProgress(dirRelativePath, entries)
@@ -626,11 +618,18 @@ func uploadDirInternal(
if workItemErr != nil && workItemErr != errCancelled {
return "", fs.DirectorySummary{}, workItemErr
}
if err := u.processUploadWorkItems(workItems, dw); err != nil && err != errCancelled {
if err := u.processUploadWorkItems(workItems, dirManifest); err != nil && err != errCancelled {
return "", fs.DirectorySummary{}, err
}
if err := dw.Finalize(&summ); err != nil {
return "", fs.DirectorySummary{}, errors.Wrap(err, "unable to finalize directory")
dirManifest.Summary = &summ
writer := u.repo.Objects.NewWriter(ctx, object.WriterOptions{
Description: "DIR:" + dirRelativePath,
Prefix: "k",
})
if err := json.NewEncoder(writer).Encode(&dirManifest); err != nil {
return "", fs.DirectorySummary{}, errors.Wrap(err, "unable to encode directory JSON")
}
oid, err := writer.Result()