Add option to recursively delete a folder.

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)
 }

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)

go/weed/weed_server/filer_server.go

@@ -21,7 +21,7 @@ func NewFilerServer(r *http.ServeMux, port int, master string, dir string, colle
 		port:       ":" + strconv.Itoa(port),
 	}
 
-	if fs.filer, err = filer.NewFilerEmbedded(dir); err != nil {
+	if fs.filer, err = filer.NewFilerEmbedded(master, dir); err != nil {
 		glog.Fatal("Can not start filer in dir", dir, ": ", err.Error())
 		return
 	}

go/weed/weed_server/filer_server_handlers.go

@@ -171,11 +171,14 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// curl -X DELETE http://localhost:8888/path/to
+// curl -X DELETE http://localhost:8888/path/to?recursive=true
 func (fs *FilerServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
 	var err error
 	var fid string
 	if strings.HasSuffix(r.URL.Path, "/") {
-		err = fs.filer.DeleteDirectory(r.URL.Path)
+		isRecursive := r.FormValue("recursive") == "true"
+		err = fs.filer.DeleteDirectory(r.URL.Path, isRecursive)
 	} else {
 		fid, err = fs.filer.DeleteFile(r.URL.Path)
 		if err == nil {