weed replicate: replicate atomic rename to other systems

weed/server/filer_grpc_server_rename.go

@@ -24,7 +24,8 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom
 		return nil, fmt.Errorf("%s/%s not found: %v", req.OldDirectory, req.OldName, err)
 	}
 
-	moveErr := fs.moveEntry(ctx, oldParent, oldEntry, filer2.FullPath(filepath.ToSlash(req.NewDirectory)), req.NewName)
+	var events MoveEvents
+	moveErr := fs.moveEntry(ctx, oldParent, oldEntry, filer2.FullPath(filepath.ToSlash(req.NewDirectory)), req.NewName, &events)
 	if moveErr != nil {
 		fs.filer.RollbackTransaction(ctx)
 		return nil, fmt.Errorf("%s/%s move error: %v", req.OldDirectory, req.OldName, err)
@@ -35,19 +36,26 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom
 		}
 	}
 
+	for _, entry := range events.newEntries {
+		fs.filer.NotifyUpdateEvent(nil, entry, false)
+	}
+	for _, entry := range events.oldEntries {
+		fs.filer.NotifyUpdateEvent(entry, nil, false)
+	}
+
 	return &filer_pb.AtomicRenameEntryResponse{}, nil
 }
 
-func (fs *FilerServer) moveEntry(ctx context.Context, oldParent filer2.FullPath, entry *filer2.Entry, newParent filer2.FullPath, newName string) error {
+func (fs *FilerServer) moveEntry(ctx context.Context, oldParent filer2.FullPath, entry *filer2.Entry, newParent filer2.FullPath, newName string, events *MoveEvents) error {
 	if entry.IsDirectory() {
-		if err := fs.moveFolderSubEntries(ctx, oldParent, entry, newParent, newName); err != nil {
+		if err := fs.moveFolderSubEntries(ctx, oldParent, entry, newParent, newName, events); err != nil {
 			return err
 		}
 	}
-	return fs.moveSelfEntry(ctx, oldParent, entry, newParent, newName)
+	return fs.moveSelfEntry(ctx, oldParent, entry, newParent, newName, events)
 }
 
-func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, oldParent filer2.FullPath, entry *filer2.Entry, newParent filer2.FullPath, newName string) error {
+func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, oldParent filer2.FullPath, entry *filer2.Entry, newParent filer2.FullPath, newName string, events *MoveEvents) error {
 
 	currentDirPath := oldParent.Child(entry.Name())
 	newDirPath := newParent.Child(newName)
@@ -68,7 +76,7 @@ func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, oldParent filer
 		for _, item := range entries {
 			lastFileName = item.Name()
 			println("processing", lastFileName)
-			err := fs.moveEntry(ctx, currentDirPath, item, newDirPath, item.Name())
+			err := fs.moveEntry(ctx, currentDirPath, item, newDirPath, item.Name(), events)
 			if err != nil {
 				return err
 			}
@@ -80,18 +88,19 @@ func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, oldParent filer
 	return nil
 }
 
-func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent filer2.FullPath, entry *filer2.Entry, newParent filer2.FullPath, newName string) error {
+func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent filer2.FullPath, entry *filer2.Entry, newParent filer2.FullPath, newName string, events *MoveEvents) (error) {
 
 	oldPath, newPath := oldParent.Child(entry.Name()), newParent.Child(newName)
 
 	glog.V(1).Infof("moving entry %s => %s", oldPath, newPath)
 
 	// add to new directory
-	createErr := fs.filer.CreateEntry(ctx, &filer2.Entry{
+	newEntry := &filer2.Entry{
 		FullPath: newPath,
 		Attr:     entry.Attr,
 		Chunks:   entry.Chunks,
-	})
+	}
+	createErr := fs.filer.CreateEntry(ctx, newEntry)
 	if createErr != nil {
 		return createErr
 	}
@@ -101,6 +110,14 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent filer2.FullP
 	if deleteErr != nil {
 		return deleteErr
 	}
+
+	events.oldEntries = append(events.oldEntries, entry)
+	events.newEntries = append(events.newEntries, newEntry)
 	return nil
 
 }
+
+type MoveEvents struct {
+	oldEntries []*filer2.Entry
+	newEntries []*filer2.Entry
+}