Add option to recursively delete a folder.

2 files changed, 42 insertions, 6 deletions

diff --git a/go/filer/filer.go b/go/filer/filer.go
index e91218afb..a42d9ed8c 100644
--- a/go/filer/filer.go
+++ b/go/filer/filer.go
@@ -14,6 +14,6 @@ type Filer interface {
 	FindFile(filePath string) (fid string, err error)
 	ListDirectories(dirPath string) (dirs []DirectoryEntry, err error)
 	ListFiles(dirPath string, lastFileName string, limit int) (files []FileEntry, err error)
-	DeleteDirectory(dirPath string) (err error)
+	DeleteDirectory(dirPath string, recursive bool) (err error)
 	DeleteFile(filePath string) (fid string, err error)
 }
diff --git a/go/filer/filer_embedded.go b/go/filer/filer_embedded.go
index 12783bc37..2a13805e9 100644
--- a/go/filer/filer_embedded.go
+++ b/go/filer/filer_embedded.go
@@ -1,16 +1,20 @@
 package filer
 
 import (
+	"code.google.com/p/weed-fs/go/operation"
+	"errors"
 	"fmt"
 	"path/filepath"
+	"strings"
 )
 
 type FilerEmbedded struct {
+	master      string
 	directories *DirectoryManagerInMap
 	files       *FileListInLevelDb
 }
 
-func NewFilerEmbedded(dir string) (filer *FilerEmbedded, err error) {
+func NewFilerEmbedded(master string, dir string) (filer *FilerEmbedded, err error) {
 	dm, de := NewDirectoryManagerInMap(filepath.Join(dir, "dir.log"))
 	if de != nil {
 		return nil, de
@@ -20,6 +24,7 @@ func NewFilerEmbedded(dir string) (filer *FilerEmbedded, err error) {
 		return nil, fe
 	}
 	filer = &FilerEmbedded{
+		master:      master,
 		directories: dm,
 		files:       fl,
 	}
@@ -52,15 +57,46 @@ func (filer *FilerEmbedded) ListFiles(dirPath string, lastFileName string, limit
 	}
 	return filer.files.ListFiles(dirId, lastFileName, limit), nil
 }
-func (filer *FilerEmbedded) DeleteDirectory(dirPath string) (err error) {
+func (filer *FilerEmbedded) DeleteDirectory(dirPath string, recursive bool) (err error) {
 	dirId, e := filer.directories.FindDirectory(dirPath)
 	if e != nil {
 		return e
 	}
-	if len(filer.files.ListFiles(dirId, "", 1)) > 0 {
-		return fmt.Errorf("Fail to delete non-empty directory %s!", dirPath)
+	if sub_dirs, sub_err := filer.directories.ListDirectories(dirPath); sub_err == nil {
+		if len(sub_dirs) > 0 && !recursive {
+			return fmt.Errorf("Fail to delete directory %s: %d sub directories found!", dirPath, len(sub_dirs))
+		}
+		for _, sub := range sub_dirs {
+			if delete_sub_err := filer.DeleteDirectory(filepath.Join(dirPath, sub.Name), recursive); delete_sub_err != nil {
+				return delete_sub_err
+			}
+		}
 	}
-	return filer.directories.DeleteDirectory(dirPath)
+	list := filer.files.ListFiles(dirId, "", 100)
+	if len(list) != 0 && !recursive {
+		if !recursive {
+			return fmt.Errorf("Fail to delete non-empty directory %s!", dirPath)
+		}
+	}
+	for {
+		if len(list) == 0 {
+			return filer.directories.DeleteDirectory(dirPath)
+		}
+		var fids []string
+		for _, fileEntry := range list {
+			fids = append(fids, string(fileEntry.Id))
+		}
+		if result_list, delete_file_err := operation.DeleteFiles(filer.master, fids); delete_file_err != nil {
+			return delete_file_err
+		} else {
+			if len(result_list.Errors) > 0 {
+				return errors.New(strings.Join(result_list.Errors, "\n"))
+			}
+		}
+		lastFile := list[len(list)-1]
+		list = filer.files.ListFiles(dirId, lastFile.Name, 100)
+	}
+
 }
 func (filer *FilerEmbedded) DeleteFile(filePath string) (fid string, err error) {
 	dir, file := filepath.Split(filePath)