You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

137 lines
3.8 KiB

5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
  1. package weed_server
  2. import (
  3. "context"
  4. "fmt"
  5. "path/filepath"
  6. "github.com/chrislusf/seaweedfs/weed/filer"
  7. "github.com/chrislusf/seaweedfs/weed/glog"
  8. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  9. "github.com/chrislusf/seaweedfs/weed/util"
  10. )
  11. func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.AtomicRenameEntryRequest) (*filer_pb.AtomicRenameEntryResponse, error) {
  12. glog.V(1).Infof("AtomicRenameEntry %v", req)
  13. oldParent := util.FullPath(filepath.ToSlash(req.OldDirectory))
  14. newParent := util.FullPath(filepath.ToSlash(req.NewDirectory))
  15. if err := fs.filer.CanRename(oldParent, newParent); err != nil {
  16. return nil, err
  17. }
  18. ctx, err := fs.filer.BeginTransaction(ctx)
  19. if err != nil {
  20. return nil, err
  21. }
  22. oldEntry, err := fs.filer.FindEntry(ctx, oldParent.Child(req.OldName))
  23. if err != nil {
  24. fs.filer.RollbackTransaction(ctx)
  25. return nil, fmt.Errorf("%s/%s not found: %v", req.OldDirectory, req.OldName, err)
  26. }
  27. moveErr := fs.moveEntry(ctx, oldParent, oldEntry, newParent, req.NewName)
  28. if moveErr != nil {
  29. fs.filer.RollbackTransaction(ctx)
  30. return nil, fmt.Errorf("%s/%s move error: %v", req.OldDirectory, req.OldName, moveErr)
  31. } else {
  32. if commitError := fs.filer.CommitTransaction(ctx); commitError != nil {
  33. fs.filer.RollbackTransaction(ctx)
  34. return nil, fmt.Errorf("%s/%s move commit error: %v", req.OldDirectory, req.OldName, commitError)
  35. }
  36. }
  37. return &filer_pb.AtomicRenameEntryResponse{}, nil
  38. }
  39. func (fs *FilerServer) moveEntry(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string) error {
  40. if err := fs.moveSelfEntry(ctx, oldParent, entry, newParent, newName, func() error {
  41. if entry.IsDirectory() {
  42. if err := fs.moveFolderSubEntries(ctx, oldParent, entry, newParent, newName); err != nil {
  43. return err
  44. }
  45. }
  46. return nil
  47. }); err != nil {
  48. return fmt.Errorf("fail to move %s => %s: %v", oldParent.Child(entry.Name()), newParent.Child(newName), err)
  49. }
  50. return nil
  51. }
  52. func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string) error {
  53. currentDirPath := oldParent.Child(entry.Name())
  54. newDirPath := newParent.Child(newName)
  55. glog.V(1).Infof("moving folder %s => %s", currentDirPath, newDirPath)
  56. lastFileName := ""
  57. includeLastFile := false
  58. for {
  59. entries, hasMore, err := fs.filer.ListDirectoryEntries(ctx, currentDirPath, lastFileName, includeLastFile, 1024, "", "", "")
  60. if err != nil {
  61. return err
  62. }
  63. // println("found", len(entries), "entries under", currentDirPath)
  64. for _, item := range entries {
  65. lastFileName = item.Name()
  66. // println("processing", lastFileName)
  67. err := fs.moveEntry(ctx, currentDirPath, item, newDirPath, item.Name())
  68. if err != nil {
  69. return err
  70. }
  71. }
  72. if !hasMore {
  73. break
  74. }
  75. }
  76. return nil
  77. }
  78. func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, moveFolderSubEntries func() error) error {
  79. oldPath, newPath := oldParent.Child(entry.Name()), newParent.Child(newName)
  80. glog.V(1).Infof("moving entry %s => %s", oldPath, newPath)
  81. if oldPath == newPath {
  82. glog.V(1).Infof("skip moving entry %s => %s", oldPath, newPath)
  83. return nil
  84. }
  85. // add to new directory
  86. newEntry := &filer.Entry{
  87. FullPath: newPath,
  88. Attr: entry.Attr,
  89. Chunks: entry.Chunks,
  90. Extended: entry.Extended,
  91. Content: entry.Content,
  92. }
  93. createErr := fs.filer.CreateEntry(ctx, newEntry, false, false, nil)
  94. if createErr != nil {
  95. return createErr
  96. }
  97. if moveFolderSubEntries != nil {
  98. if moveChildrenErr := moveFolderSubEntries(); moveChildrenErr != nil {
  99. return moveChildrenErr
  100. }
  101. }
  102. // delete old entry
  103. deleteErr := fs.filer.DeleteEntryMetaAndData(ctx, oldPath, false, false, false, false, nil)
  104. if deleteErr != nil {
  105. return deleteErr
  106. }
  107. return nil
  108. }