1 files changed, 31 insertions, 35 deletions

diff --git a/weed/replication/sink/gcssink/gcs_sink.go b/weed/replication/sink/gcssink/gcs_sink.go
index abd7c49b9..5cf5b7317 100644
--- a/weed/replication/sink/gcssink/gcs_sink.go
+++ b/weed/replication/sink/gcssink/gcs_sink.go
@@ -3,23 +3,26 @@ package gcssink
 import (
 	"context"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/replication/repl_util"
 	"os"
 
 	"cloud.google.com/go/storage"
-	"github.com/chrislusf/seaweedfs/weed/filer2"
+	"google.golang.org/api/option"
+
+	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/replication/sink"
 	"github.com/chrislusf/seaweedfs/weed/replication/source"
 	"github.com/chrislusf/seaweedfs/weed/util"
-	"google.golang.org/api/option"
 )
 
 type GcsSink struct {
-	client      *storage.Client
-	bucket      string
-	dir         string
-	filerSource *source.FilerSource
+	client        *storage.Client
+	bucket        string
+	dir           string
+	filerSource   *source.FilerSource
+	isIncremental bool
 }
 
 func init() {
@@ -34,11 +37,16 @@ func (g *GcsSink) GetSinkToDirectory() string {
 	return g.dir
 }
 
-func (g *GcsSink) Initialize(configuration util.Configuration) error {
+func (g *GcsSink) IsIncremental() bool {
+	return g.isIncremental
+}
+
+func (g *GcsSink) Initialize(configuration util.Configuration, prefix string) error {
+	g.isIncremental = configuration.GetBool(prefix + "is_incremental")
 	return g.initialize(
-		configuration.GetString("google_application_credentials"),
-		configuration.GetString("bucket"),
-		configuration.GetString("directory"),
+		configuration.GetString(prefix+"google_application_credentials"),
+		configuration.GetString(prefix+"bucket"),
+		configuration.GetString(prefix+"directory"),
 	)
 }
 
@@ -50,7 +58,6 @@ func (g *GcsSink) initialize(google_application_credentials, bucketName, dir str
 	g.bucket = bucketName
 	g.dir = dir
 
-	ctx := context.Background()
 	// Creates a client.
 	if google_application_credentials == "" {
 		var found bool
@@ -59,7 +66,7 @@ func (g *GcsSink) initialize(google_application_credentials, bucketName, dir str
 			glog.Fatalf("need to specific GOOGLE_APPLICATION_CREDENTIALS env variable or google_application_credentials in replication.toml")
 		}
 	}
-	client, err := storage.NewClient(ctx, option.WithCredentialsFile(google_application_credentials))
+	client, err := storage.NewClient(context.Background(), option.WithCredentialsFile(google_application_credentials))
 	if err != nil {
 		glog.Fatalf("Failed to create client: %v", err)
 	}
@@ -69,13 +76,13 @@ func (g *GcsSink) initialize(google_application_credentials, bucketName, dir str
 	return nil
 }
 
-func (g *GcsSink) DeleteEntry(ctx context.Context, key string, isDirectory, deleteIncludeChunks bool) error {
+func (g *GcsSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error {
 
 	if isDirectory {
 		key = key + "/"
 	}
 
-	if err := g.client.Bucket(g.bucket).Object(key).Delete(ctx); err != nil {
+	if err := g.client.Bucket(g.bucket).Object(key).Delete(context.Background()); err != nil {
 		return fmt.Errorf("gcs delete %s%s: %v", g.bucket, key, err)
 	}
 
@@ -83,35 +90,24 @@ func (g *GcsSink) DeleteEntry(ctx context.Context, key string, isDirectory, dele
 
 }
 
-func (g *GcsSink) CreateEntry(ctx context.Context, key string, entry *filer_pb.Entry) error {
+func (g *GcsSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error {
 
 	if entry.IsDirectory {
 		return nil
 	}
 
-	totalSize := filer2.TotalSize(entry.Chunks)
-	chunkViews := filer2.ViewFromChunks(entry.Chunks, 0, int(totalSize))
-
-	wc := g.client.Bucket(g.bucket).Object(key).NewWriter(ctx)
+	totalSize := filer.FileSize(entry)
+	chunkViews := filer.ViewFromChunks(g.filerSource.LookupFileId, entry.Chunks, 0, int64(totalSize))
 
-	for _, chunk := range chunkViews {
-
-		fileUrl, err := g.filerSource.LookupFileId(ctx, chunk.FileId)
-		if err != nil {
-			return err
-		}
-
-		_, err = util.ReadUrlAsStream(fileUrl, chunk.Offset, int(chunk.Size), func(data []byte) {
-			wc.Write(data)
-		})
-
-		if err != nil {
-			return err
-		}
+	wc := g.client.Bucket(g.bucket).Object(key).NewWriter(context.Background())
+	defer wc.Close()
 
+	writeFunc := func(data []byte) error {
+		_, writeErr := wc.Write(data)
+		return writeErr
 	}
 
-	if err := wc.Close(); err != nil {
+	if err := repl_util.CopyFromChunkViews(chunkViews, g.filerSource, writeFunc); err != nil {
 		return err
 	}
 
@@ -119,7 +115,7 @@ func (g *GcsSink) CreateEntry(ctx context.Context, key string, entry *filer_pb.E
 
 }
 
-func (g *GcsSink) UpdateEntry(ctx context.Context, key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool) (foundExistingEntry bool, err error) {
+func (g *GcsSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error) {
 	// TODO improve efficiency
 	return false, nil
 }