moved most methods from Repository to casManager

cli/command_backups.go

@@ -95,7 +95,7 @@ func runBackupsCommand(context *kingpin.ParseContext) error {
 		if count < *maxResultsPerPath {
 			fmt.Printf(
 				"  %v%v %v %10v %v\n",
-				m.Handle,
+				m.RootObjectID,
 				relPath,
 				m.StartTime.Format("2006-01-02 15:04:05 MST"),
 				units.BytesString(m.Stats.TotalFileSize),

client/connection.go

@@ -106,7 +106,7 @@ func Open(ctx context.Context, configFile string, options *Options) (*Connection
 		}
 	}
 
-	conn.Repository, err = repo.New(repositoryStorage, conn.Vault.RepoConfig.Format)
+	conn.Repository, err = repo.NewRepository(repositoryStorage, conn.Vault.RepoConfig.Format)
 	if err != nil {
 		vaultStorage.Close()
 		repositoryStorage.Close()

repo/blocks.go

@@ -1,7 +1,7 @@
 package repo
 
 // GetStorageBlocks returns the list of storage blocks used by the specified object ID.
-func (r *Repository) GetStorageBlocks(oid ObjectID) ([]string, error) {
+func (r *casManager) GetStorageBlocks(oid ObjectID) ([]string, error) {
 	result := map[string]bool{}
 	if err := r.addStorageBlocks(result, oid); err != nil {
 		return nil, err
@@ -14,7 +14,7 @@ func (r *Repository) GetStorageBlocks(oid ObjectID) ([]string, error) {
 	return b, nil
 }
 
-func (r *Repository) addStorageBlocks(result map[string]bool, oid ObjectID) error {
+func (r *casManager) addStorageBlocks(result map[string]bool, oid ObjectID) error {
 	if oid.Section != nil {
 		return r.addStorageBlocks(result, oid.Section.Base)
 	}

repo/cas_manager.go

@@ -22,8 +22,8 @@ type ObjectReader interface {
 	Length() int64
 }
 
-// Repository implements a content-addressable storage on top of blob storage.
-type Repository struct {
+// casManager implements a content-addressable storage on top of blob storage.
+type casManager struct {
 	Stats   Stats        // vital statistics
 	Storage blob.Storage // underlying blob storage
 
@@ -38,24 +38,15 @@ type Repository struct {
 }
 
 // Close closes the connection to the underlying blob storage and releases any resources.
-func (r *Repository) Close() error {
-	r.Flush()
-	if err := r.Storage.Close(); err != nil {
-		return err
-	}
+func (r *casManager) Close() error {
+	r.writeBack.flush()
 	r.bufferManager.close()
 
 	return nil
 }
 
-// Flush waits for all in-flight writes to complete.
-func (r *Repository) Flush() error {
-	r.writeBack.flush()
-	return nil
-}
-
 // NewWriter creates an ObjectWriter for writing to the repository.
-func (r *Repository) NewWriter(options ...WriterOption) ObjectWriter {
+func (r *casManager) NewWriter(options ...WriterOption) ObjectWriter {
 	result := &objectWriter{
 		repo:         r,
 		blockTracker: &blockTracker{},
@@ -70,12 +61,12 @@ func (r *Repository) NewWriter(options ...WriterOption) ObjectWriter {
 }
 
 // Open creates new ObjectReader for reading given object from a repository.
-func (r *Repository) Open(objectID ObjectID) (ObjectReader, error) {
+func (r *casManager) Open(objectID ObjectID) (ObjectReader, error) {
 	// log.Printf("Repository::Open %v", objectID.String())
 	// defer log.Printf("finished Repository::Open() %v", objectID.String())
 
 	// Flush any pending writes.
-	r.Flush()
+	r.writeBack.flush()
 
 	if objectID.Section != nil {
 		baseReader, err := r.Open(objectID.Section.Base)
@@ -111,32 +102,32 @@ func (r *Repository) Open(objectID ObjectID) (ObjectReader, error) {
 }
 
 // RepositoryOption controls the behavior of Repository.
-type RepositoryOption func(o *Repository)
+type RepositoryOption func(o *casManager)
 
 // WriteBack is an RepositoryOption that enables asynchronous writes to the storage using the pool
 // of goroutines.
 func WriteBack(writeBackWorkers int) RepositoryOption {
-	return func(o *Repository) {
+	return func(o *casManager) {
 		o.writeBack.workers = writeBackWorkers
 	}
 }
 
 // EnableLogging is an RepositoryOption that causes all storage access to be logged.
 func EnableLogging(options ...logging.Option) RepositoryOption {
-	return func(o *Repository) {
+	return func(o *casManager) {
 		o.Storage = logging.NewWrapper(o.Storage, options...)
 	}
 }
 
-// New creates a Repository with the specified storage, format and options.
-func New(s blob.Storage, f *Format, options ...RepositoryOption) (*Repository, error) {
+// newCASManager creates a casManager with the specified storage, format and options.
+func newCASManager(s blob.Storage, f *Format, options ...RepositoryOption) (*casManager, error) {
 	if err := f.Validate(); err != nil {
 		return nil, err
 	}
 
 	sf := SupportedFormats[f.ObjectFormat]
 
-	r := &Repository{
+	r := &casManager{
 		Storage: s,
 		format:  *f,
 	}
@@ -177,7 +168,7 @@ func New(s blob.Storage, f *Format, options ...RepositoryOption) (*Repository, e
 // hashEncryptAndWriteMaybeAsync computes hash of a given buffer, optionally encrypts and writes it to storage.
 // The write is not guaranteed to complete synchronously in case write-back is used, but by the time
 // Repository.Close() returns all writes are guaranteed be over.
-func (r *Repository) hashEncryptAndWriteMaybeAsync(buffer *bytes.Buffer, prefix string) (ObjectID, error) {
+func (r *casManager) hashEncryptAndWriteMaybeAsync(buffer *bytes.Buffer, prefix string) (ObjectID, error) {
 	var data []byte
 	if buffer != nil {
 		data = buffer.Bytes()
@@ -211,7 +202,7 @@ func (r *Repository) hashEncryptAndWriteMaybeAsync(buffer *bytes.Buffer, prefix
 	return r.encryptAndMaybeWrite(objectID, buffer, prefix)
 }
 
-func (r *Repository) encryptAndMaybeWrite(objectID ObjectID, buffer *bytes.Buffer, prefix string) (ObjectID, error) {
+func (r *casManager) encryptAndMaybeWrite(objectID ObjectID, buffer *bytes.Buffer, prefix string) (ObjectID, error) {
 	defer r.bufferManager.returnBuffer(buffer)
 
 	var data []byte
@@ -250,7 +241,7 @@ func (r *Repository) encryptAndMaybeWrite(objectID ObjectID, buffer *bytes.Buffe
 	return objectID, nil
 }
 
-func (r *Repository) flattenListChunk(rawReader io.Reader) ([]indirectObjectEntry, error) {
+func (r *casManager) flattenListChunk(rawReader io.Reader) ([]indirectObjectEntry, error) {
 	pr, err := jsonstream.NewReader(bufio.NewReader(rawReader), indirectStreamType)
 	if err != nil {
 		return nil, err
@@ -284,7 +275,7 @@ func removeIndirection(o ObjectID) ObjectID {
 	return o
 }
 
-func (r *Repository) newRawReader(objectID ObjectID) (ObjectReader, error) {
+func (r *casManager) newRawReader(objectID ObjectID) (ObjectReader, error) {
 	if objectID.BinaryContent != nil {
 		return newObjectReaderWithData(objectID.BinaryContent), nil
 	}
@@ -317,7 +308,7 @@ func (r *Repository) newRawReader(objectID ObjectID) (ObjectReader, error) {
 	return newObjectReaderWithData(payload), nil
 }
 
-func (r *Repository) verifyChecksum(data []byte, blockID string) error {
+func (r *casManager) verifyChecksum(data []byte, blockID string) error {
 	expected := r.formatter.ComputeObjectID(data)
 	if !strings.HasSuffix(blockID, expected.StorageBlock) {
 		atomic.AddInt32(&r.Stats.InvalidBlocks, 1)

repo/cas_manager_test.go

@@ -44,11 +44,11 @@ func getMd5LObjectID(data []byte) string {
 	return fmt.Sprintf("L%s", getMd5Digest(data))
 }
 
-func setupTest(t *testing.T) (data map[string][]byte, repo *Repository) {
+func setupTest(t *testing.T) (data map[string][]byte, repo *casManager) {
 	data = map[string][]byte{}
 	st := storagetesting.NewMapStorage(data)
 
-	repo, err := New(st, testFormat(), WriteBack(5))
+	repo, err := newCASManager(st, testFormat(), WriteBack(5))
 	if err != nil {
 		t.Errorf("cannot create manager: %v", err)
 	}
@@ -88,7 +88,7 @@ func TestWriters(t *testing.T) {
 			continue
 		}
 
-		repo.Flush()
+		repo.writeBack.flush()
 
 		if !objectIDsEqual(result, c.objectID) {
 			t.Errorf("incorrect result for %v, expected: %v got: %v %#v", c.data, c.objectID.String(), result.String(), result.BinaryContent)
@@ -125,7 +125,7 @@ func TestWriterCompleteChunkInTwoWrites(t *testing.T) {
 	}
 }
 
-func verifyIndirectBlock(t *testing.T, r *Repository, oid ObjectID) {
+func verifyIndirectBlock(t *testing.T, r *casManager, oid ObjectID) {
 	for level := int32(0); level < oid.Indirect; level++ {
 		direct := oid
 		direct.Indirect = level
@@ -178,7 +178,7 @@ func TestIndirection(t *testing.T) {
 		writer := repo.NewWriter()
 		writer.Write(contentBytes)
 		result, err := writer.Result(false)
-		repo.Flush()
+		repo.writeBack.flush()
 		if err != nil {
 			t.Errorf("error getting writer results: %v", err)
 		}
@@ -211,7 +211,7 @@ func TestHMAC(t *testing.T) {
 	s := testFormat()
 	s.ObjectFormat = "TESTONLY_MD5"
 
-	repo, err := New(storagetesting.NewMapStorage(data), s)
+	repo, err := newCASManager(storagetesting.NewMapStorage(data), s)
 	if err != nil {
 		t.Errorf("cannot create manager: %v", err)
 	}
@@ -342,7 +342,7 @@ func TestEndToEndReadAndSeek(t *testing.T) {
 	}
 }
 
-func verify(t *testing.T, repo *Repository, objectID ObjectID, expectedData []byte, testCaseID string) {
+func verify(t *testing.T, repo *casManager, objectID ObjectID, expectedData []byte, testCaseID string) {
 	reader, err := repo.Open(objectID)
 	if err != nil {
 		t.Errorf("cannot get reader for %v: %v", testCaseID, err)
@@ -422,7 +422,7 @@ func TestFormats(t *testing.T) {
 		data := map[string][]byte{}
 		st := storagetesting.NewMapStorage(data)
 
-		repo, err := New(st, c.format)
+		repo, err := newCASManager(st, c.format)
 		if err != nil {
 			t.Errorf("cannot create manager: %v", err)
 			continue

repo/object_reader.go

@@ -10,7 +10,7 @@ func (i *indirectObjectEntry) endOffset() int64 {
 }
 
 type objectReader struct {
-	repo *Repository
+	repo *casManager
 
 	seekTable []indirectObjectEntry
 

repo/object_writer.go

@@ -37,7 +37,7 @@ func (t *blockTracker) blockIDs() []string {
 }
 
 type objectWriter struct {
-	repo *Repository
+	repo *casManager
 
 	buffer      *bytes.Buffer
 	totalLength int64

repo/repository.go

@@ -0,0 +1,40 @@
+package repo
+
+import (
+	"github.com/kopia/kopia/blob"
+)
+
+// Repository represents storage where both content-addressable and user-addressable data is kept.
+type Repository struct {
+	*casManager
+	Storage blob.Storage
+}
+
+// Close closes the repository and releases all resources.
+func (r *Repository) Close() error {
+	if err := r.casManager.Close(); err != nil {
+		return err
+	}
+	if err := r.Storage.Close(); err != nil {
+		return err
+	}
+	return nil
+}
+
+// Flush waits for all in-flight writes to complete.
+func (r *Repository) Flush() error {
+	r.casManager.writeBack.flush()
+	return nil
+}
+
+// NewRepository initializes a new repository with a given format.
+func NewRepository(s blob.Storage, f *Format, options ...RepositoryOption) (*Repository, error) {
+	cm, err := newCASManager(s, f, options...)
+	if err != nil {
+		return nil, err
+	}
+	return &Repository{
+		Storage:    s,
+		casManager: cm,
+	}, nil
+}

snapshot/manifest.go

@@ -14,7 +14,6 @@ type Manifest struct {
 	StartTime   time.Time `json:"startTime"`
 	EndTime     time.Time `json:"endTime"`
 
-	Handle       string        `json:"handle"`
 	RootObjectID repo.ObjectID `json:"root"`
 	HashCacheID  repo.ObjectID `json:"hashCache"`
 

snapshot/upload_test.go

@@ -51,7 +51,7 @@ func newUploadTestHarness() *uploadTestHarness {
 		MaxInlineContentLength: 0,
 	}
 
-	repo, err := repo.New(storage, &format)
+	repo, err := repo.NewRepository(storage, &format)
 	if err != nil {
 		panic("unable to create repository: " + err.Error())
 	}