merge master, resolve conflicts

1 files changed, 107 insertions, 31 deletions

diff --git a/weed/filesys/dir_rename.go b/weed/filesys/dir_rename.go
index d50c6dab0..dd76577b0 100644
--- a/weed/filesys/dir_rename.go
+++ b/weed/filesys/dir_rename.go
@@ -2,6 +2,9 @@ package filesys
 
 import (
 	"context"
+	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"math"
 
 	"github.com/seaweedfs/fuse"
 	"github.com/seaweedfs/fuse/fs"
@@ -37,6 +40,7 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
 			OldName:      req.OldName,
 			NewDirectory: newDir.FullPath(),
 			NewName:      req.NewName,
+			Signatures:   []int32{dir.wfs.signature},
 		}
 
 		_, err := client.AtomicRenameEntry(ctx, request)
@@ -53,46 +57,118 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
 		return fuse.EIO
 	}
 
-	// TODO: replicate renaming logic on filer
-	if err := dir.wfs.metaCache.DeleteEntry(context.Background(), oldPath); err != nil {
-		glog.V(0).Infof("dir Rename delete local %s => %s : %v", oldPath, newPath, err)
+	err = dir.moveEntry(context.Background(), util.FullPath(dir.FullPath()), oldEntry, util.FullPath(newDir.FullPath()), req.NewName)
+	if err != nil {
+		glog.V(0).Infof("dir local Rename %s => %s : %v", oldPath, newPath, err)
 		return fuse.EIO
 	}
-	oldEntry.FullPath = newPath
-	if err := dir.wfs.metaCache.InsertEntry(context.Background(), oldEntry); err != nil {
+
+	return nil
+}
+
+func (dir *Dir) moveEntry(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string) error {
+
+	oldName := entry.Name()
+
+	oldPath := oldParent.Child(oldName)
+	newPath := newParent.Child(newName)
+	if err := dir.moveSelfEntry(ctx, oldParent, entry, newParent, newName, func() error {
+
+		oldFsNode := NodeWithId(oldPath.AsInode())
+		newFsNode := NodeWithId(newPath.AsInode())
+		newDirNode, found := dir.wfs.Server.FindInternalNode(NodeWithId(newParent.AsInode()))
+		var newDir *Dir
+		if found {
+			newDir = newDirNode.(*Dir)
+		}
+		dir.wfs.Server.InvalidateInternalNode(oldFsNode, newFsNode, func(internalNode fs.Node) {
+			if file, ok := internalNode.(*File); ok {
+				glog.V(4).Infof("internal file node %s", oldParent.Child(oldName))
+				file.Name = newName
+				file.id = uint64(newFsNode)
+				if found {
+					file.dir = newDir
+				}
+			}
+			if dir, ok := internalNode.(*Dir); ok {
+				glog.V(4).Infof("internal dir node %s", oldParent.Child(oldName))
+				dir.name = newName
+				dir.id = uint64(newFsNode)
+				if found {
+					dir.parent = newDir
+				}
+			}
+		})
+
+		// change file handle
+		inodeId := oldPath.AsInode()
+		dir.wfs.handlesLock.Lock()
+		if existingHandle, found := dir.wfs.handles[inodeId]; found && existingHandle == nil {
+			glog.V(4).Infof("opened file handle %s => %s", oldPath, newPath)
+			delete(dir.wfs.handles, inodeId)
+			dir.wfs.handles[newPath.AsInode()] = existingHandle
+		}
+		dir.wfs.handlesLock.Unlock()
+
+		if entry.IsDirectory() {
+			if err := dir.moveFolderSubEntries(ctx, oldParent, oldName, newParent, newName); err != nil {
+				return err
+			}
+		}
+		return nil
+	}); err != nil {
+		return fmt.Errorf("fail to move %s => %s: %v", oldPath, newPath, err)
+	}
+
+	return nil
+}
+
+func (dir *Dir) moveFolderSubEntries(ctx context.Context, oldParent util.FullPath, oldName string, newParent util.FullPath, newName string) error {
+
+	currentDirPath := oldParent.Child(oldName)
+	newDirPath := newParent.Child(newName)
+
+	glog.V(1).Infof("moving folder %s => %s", currentDirPath, newDirPath)
+
+	var moveErr error
+	listErr := dir.wfs.metaCache.ListDirectoryEntries(ctx, currentDirPath, "", false, int64(math.MaxInt32), func(item *filer.Entry) bool {
+		moveErr = dir.moveEntry(ctx, currentDirPath, item, newDirPath, item.Name())
+		if moveErr != nil {
+			return false
+		}
+		return true
+	})
+	if listErr != nil {
+		return listErr
+	}
+	if moveErr != nil {
+		return moveErr
+	}
+
+	return nil
+}
+
+func (dir *Dir) moveSelfEntry(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, moveFolderSubEntries func() error) error {
+
+	newPath := newParent.Child(newName)
+	oldPath := oldParent.Child(entry.Name())
+
+	entry.FullPath = newPath
+	if err := dir.wfs.metaCache.InsertEntry(ctx, entry); err != nil {
 		glog.V(0).Infof("dir Rename insert local %s => %s : %v", oldPath, newPath, err)
 		return fuse.EIO
 	}
 
-	oldFsNode := NodeWithId(oldPath.AsInode())
-	newFsNode := NodeWithId(newPath.AsInode())
-	dir.wfs.Server.InvalidateInternalNode(oldFsNode, newFsNode, func(internalNode fs.Node) {
-		if file, ok := internalNode.(*File); ok {
-			glog.V(4).Infof("internal file node %s", file.Name)
-			file.Name = req.NewName
-			file.id = uint64(newFsNode)
-			file.dir = newDir
+	if moveFolderSubEntries != nil {
+		if moveChildrenErr := moveFolderSubEntries(); moveChildrenErr != nil {
+			return moveChildrenErr
 		}
-		if dir, ok := internalNode.(*Dir); ok {
-			glog.V(4).Infof("internal dir node %s", dir.name)
-			dir.name = req.NewName
-			dir.id = uint64(newFsNode)
-			dir.parent = newDir
-		}
-	})
+	}
 
-	// change file handle
-	dir.wfs.handlesLock.Lock()
-	defer dir.wfs.handlesLock.Unlock()
-	inodeId := oldPath.AsInode()
-	existingHandle, found := dir.wfs.handles[inodeId]
-	glog.V(4).Infof("has open filehandle %s: %v", oldPath, found)
-	if !found || existingHandle == nil {
-		return nil
+	if err := dir.wfs.metaCache.DeleteEntry(ctx, oldPath); err != nil {
+		glog.V(0).Infof("dir Rename delete local %s => %s : %v", oldPath, newPath, err)
+		return fuse.EIO
 	}
-	glog.V(4).Infof("opened filehandle %s => %s", oldPath, newPath)
-	delete(dir.wfs.handles, inodeId)
-	dir.wfs.handles[newPath.AsInode()] = existingHandle
 
 	return nil
 }