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.

102 lines
3.0 KiB

  1. package filer2
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/chrislusf/seaweedfs/weed/glog"
  6. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  7. )
  8. func (f *Filer) DeleteEntryMetaAndData(ctx context.Context, p FullPath, isRecursive bool, ignoreRecursiveError, shouldDeleteChunks bool) (err error) {
  9. if p == "/" {
  10. return nil
  11. }
  12. entry, findErr := f.FindEntry(ctx, p)
  13. if findErr != nil {
  14. return findErr
  15. }
  16. var chunks []*filer_pb.FileChunk
  17. chunks = append(chunks, entry.Chunks...)
  18. if entry.IsDirectory() {
  19. // delete the folder children, not including the folder itself
  20. var dirChunks []*filer_pb.FileChunk
  21. dirChunks, err = f.doBatchDeleteFolderMetaAndData(ctx, entry, isRecursive, ignoreRecursiveError, shouldDeleteChunks)
  22. if err != nil {
  23. return fmt.Errorf("delete directory %s: %v", p, err)
  24. }
  25. chunks = append(chunks, dirChunks...)
  26. f.cacheDelDirectory(string(p))
  27. }
  28. // delete the file or folder
  29. err = f.doDeleteEntryMetaAndData(ctx, entry, shouldDeleteChunks)
  30. if err != nil {
  31. return fmt.Errorf("delete file %s: %v", p, err)
  32. }
  33. if shouldDeleteChunks {
  34. go f.DeleteChunks(chunks)
  35. }
  36. return nil
  37. }
  38. func (f *Filer) doBatchDeleteFolderMetaAndData(ctx context.Context, entry *Entry, isRecursive bool, ignoreRecursiveError, shouldDeleteChunks bool) (chunks []*filer_pb.FileChunk, err error) {
  39. lastFileName := ""
  40. includeLastFile := false
  41. for {
  42. entries, err := f.ListDirectoryEntries(ctx, entry.FullPath, lastFileName, includeLastFile, PaginationSize)
  43. if err != nil {
  44. glog.Errorf("list folder %s: %v", entry.FullPath, err)
  45. return nil, fmt.Errorf("list folder %s: %v", entry.FullPath, err)
  46. }
  47. if lastFileName == "" && !isRecursive && len(entries) > 0 {
  48. // only for first iteration in the loop
  49. return nil, fmt.Errorf("fail to delete non-empty folder: %s", entry.FullPath)
  50. }
  51. for _, sub := range entries {
  52. lastFileName = sub.Name()
  53. var dirChunks []*filer_pb.FileChunk
  54. if sub.IsDirectory() {
  55. dirChunks, err = f.doBatchDeleteFolderMetaAndData(ctx, sub, isRecursive, ignoreRecursiveError, shouldDeleteChunks)
  56. }
  57. if err != nil && !ignoreRecursiveError {
  58. return nil, err
  59. }
  60. if shouldDeleteChunks {
  61. chunks = append(chunks, dirChunks...)
  62. }
  63. }
  64. if len(entries) < PaginationSize {
  65. break
  66. }
  67. }
  68. f.cacheDelDirectory(string(entry.FullPath))
  69. glog.V(3).Infof("deleting directory %v", entry.FullPath)
  70. if storeDeletionErr := f.store.DeleteFolderChildren(ctx, entry.FullPath); storeDeletionErr != nil {
  71. return nil, fmt.Errorf("filer store delete: %v", storeDeletionErr)
  72. }
  73. f.NotifyUpdateEvent(entry, nil, shouldDeleteChunks)
  74. return chunks, nil
  75. }
  76. func (f *Filer) doDeleteEntryMetaAndData(ctx context.Context, entry *Entry, shouldDeleteChunks bool) (err error) {
  77. glog.V(3).Infof("deleting entry %v", entry.FullPath)
  78. if storeDeletionErr := f.store.DeleteEntry(ctx, entry.FullPath); storeDeletionErr != nil {
  79. return fmt.Errorf("filer store delete: %v", storeDeletionErr)
  80. }
  81. f.NotifyUpdateEvent(entry, nil, shouldDeleteChunks)
  82. return nil
  83. }