Merge pull request #914 from harshavardhana/enhance-filtering-further

Enhance listing further, this time handle cases related to common pre…

Makefile

@@ -35,8 +35,8 @@ lint:
 
 cyclo:
 	@echo "Running $@:"
-	@GO15VENDOREXPERIMENT=1 gocyclo -over 50 *.go
-	@GO15VENDOREXPERIMENT=1 gocyclo -over 50 pkg
+	@GO15VENDOREXPERIMENT=1 gocyclo -over 60 *.go
+	@GO15VENDOREXPERIMENT=1 gocyclo -over 60 pkg
 
 build: getdeps verifiers
 	@echo "Installing minio:" #@GO15VENDOREXPERIMENT=1 deadcode

pkg/fs/fs-common.go

@@ -33,13 +33,13 @@ type Metadata struct {
 
 // sortUnique sort a slice in lexical order, removing duplicate elements
 func sortUnique(objects []string) []string {
-	objectMap := make(map[string]string)
-	for _, v := range objects {
-		objectMap[v] = v
+	results := []string{}
+	seen := make(map[string]string)
+	for _, val := range objects {
+		if _, ok := seen[val]; !ok {
+			results = append(results, val)
+			seen[val] = val
 		}
-	var results []string
-	for k := range objectMap {
-		results = append(results, k)
 	}
 	sort.Strings(results)
 	return results

pkg/fs/fs-filter.go

@@ -40,6 +40,10 @@ func (fs API) filterObjects(bucket string, content contentInfo, resources Bucket
 				if err != nil {
 					return ObjectMetadata{}, resources, err.Trace()
 				}
+				if metadata.Mode.IsDir() {
+					resources.CommonPrefixes = append(resources.CommonPrefixes, name+resources.Delimiter)
+					return ObjectMetadata{}, resources, nil
+				}
 			case delimitedName == content.FileInfo.Name():
 				// Use resources.Prefix to filter out delimited files
 				metadata, err = getMetadata(fs.path, bucket, name)
@@ -63,6 +67,10 @@ func (fs API) filterObjects(bucket string, content contentInfo, resources Bucket
 			if err != nil {
 				return ObjectMetadata{}, resources, err.Trace()
 			}
+			if metadata.Mode.IsDir() {
+				resources.CommonPrefixes = append(resources.CommonPrefixes, name+resources.Delimiter)
+				return ObjectMetadata{}, resources, nil
+			}
 		case delimitedName == content.FileInfo.Name():
 			metadata, err = getMetadata(fs.path, bucket, name)
 			if err != nil {

pkg/fs/fs.go

@@ -17,6 +17,7 @@
 package fs
 
 import (
+	"errors"
 	"os"
 	"runtime"
 	"strings"
@@ -93,13 +94,20 @@ func (fs API) DeleteBucket(bucket string) *probe.Error {
 	if _, err := os.Stat(bucketDir); os.IsNotExist(err) {
 		return probe.NewError(BucketNotFound{Bucket: bucket})
 	}
-	files, err := ioutil.ReadDir(bucketDir)
-	if err != nil {
-		return probe.NewError(err)
+	var errNotEmpty = errors.New("Directory Not empty")
+	isDirNotEmpty := func(fp string, fl os.FileInfo, err error) error {
+		if fl.Mode().IsRegular() || fl.Mode()&os.ModeSymlink == os.ModeSymlink {
+			return errNotEmpty
 		}
-	if len(files) > 0 {
+		return ErrSkipDir
+	}
+	err := WalkUnsorted(bucketDir, isDirNotEmpty)
+	if err != nil {
+		if err == errNotEmpty {
 			return probe.NewError(BucketNotEmpty{Bucket: bucket})
 		}
+		return probe.NewError(err)
+	}
 	if err := os.Remove(bucketDir); err != nil {
 		return probe.NewError(err)
 	}
@@ -128,7 +136,6 @@ func (fs API) ListBuckets() ([]BucketMetadata, *probe.Error) {
 				continue
 			}
 		}
-
 		metadata := BucketMetadata{
 			Name:    file.Name(),
 			Created: file.ModTime(),
@@ -263,8 +270,44 @@ func (fs API) ListObjects(bucket string, resources BucketResourcesMetadata) ([]O
 		}
 	}
 
-	// If delimiter is supplied make sure that paging doesn't go deep, treat it as simple directory listing.
-	if resources.Delimiter != "" {
+	// if delimiter is supplied and not prefix then we are the very top level, list everything and move on.
+	if resources.Delimiter != "" && resources.Prefix == "" {
+		files, err := ioutil.ReadDir(rootPrefix)
+		if err != nil {
+			if os.IsNotExist(err) {
+				return nil, resources, probe.NewError(BucketNotFound{Bucket: bucket})
+			}
+			return nil, resources, probe.NewError(err)
+		}
+		for _, fl := range files {
+			p.files = append(p.files, contentInfo{
+				Prefix:   fl.Name(),
+				Size:     fl.Size(),
+				Mode:     fl.Mode(),
+				ModTime:  fl.ModTime(),
+				FileInfo: fl,
+			})
+		}
+	}
+
+	// If delimiter and prefix is supplied make sure that paging doesn't go deep, treat it as simple directory listing.
+	if resources.Delimiter != "" && resources.Prefix != "" {
+		if !strings.HasSuffix(resources.Prefix, resources.Delimiter) {
+			fl, err := os.Stat(filepath.Join(rootPrefix, resources.Prefix))
+			if err != nil {
+				if os.IsNotExist(err) {
+					return nil, resources, probe.NewError(ObjectNotFound{Bucket: bucket, Object: resources.Prefix})
+				}
+				return nil, resources, probe.NewError(err)
+			}
+			p.files = append(p.files, contentInfo{
+				Prefix:   resources.Prefix,
+				Size:     fl.Size(),
+				Mode:     os.ModeDir,
+				ModTime:  fl.ModTime(),
+				FileInfo: fl,
+			})
+		} else {
 			files, err := ioutil.ReadDir(filepath.Join(rootPrefix, resources.Prefix))
 			if err != nil {
 				if os.IsNotExist(err) {
@@ -285,7 +328,9 @@ func (fs API) ListObjects(bucket string, resources BucketResourcesMetadata) ([]O
 					FileInfo: fl,
 				})
 			}
-	} else {
+		}
+	}
+	if resources.Delimiter == "" {
 		var files []contentInfo
 		getAllFiles := func(fp string, fl os.FileInfo, err error) error {
 			// If any error return back quickly

pkg/fs/walk.go

@@ -33,9 +33,28 @@ func Walk(root string, walkFn WalkFunc) error {
 	return walk(root, info, walkFn)
 }
 
+// WalkUnsorted walks the file tree rooted at root, calling walkFn for each file or
+// directory in the tree, including root.
+func WalkUnsorted(root string, walkFn WalkFunc) error {
+	info, err := os.Lstat(root)
+	if err != nil {
+		return walkFn(root, nil, err)
+	}
+	return walk(root, info, walkFn)
+}
+
 // readDirNames reads the directory named by dirname and returns
 // a sorted list of directory entries.
 func readDirNames(dirname string) ([]string, error) {
+	names, err := readDirUnsortedNames(dirname)
+	if err != nil {
+		return nil, err
+	}
+	sort.Strings(names)
+	return names, nil
+}
+
+func readDirUnsortedNames(dirname string) ([]string, error) {
 	f, err := os.Open(dirname)
 	if err != nil {
 		return nil, err
@@ -45,7 +64,6 @@ func readDirNames(dirname string) ([]string, error) {
 	if err != nil {
 		return nil, err
 	}
-	sort.Strings(names)
 	return names, nil
 }
 
@@ -66,6 +84,46 @@ var ErrSkipDir = errors.New("skip this directory")
 // as an error by any function.
 var ErrSkipFile = errors.New("skip this file")
 
+func walkUnsorted(path string, info os.FileInfo, walkFn WalkFunc) error {
+	err := walkFn(path, info, nil)
+	if err != nil {
+		if info.Mode().IsDir() && err == ErrSkipDir {
+			return nil
+		}
+		if info.Mode().IsRegular() && err == ErrSkipFile {
+			return nil
+		}
+		return err
+	}
+
+	if !info.IsDir() {
+		return nil
+	}
+
+	names, err := readDirUnsortedNames(path)
+	if err != nil {
+		return walkFn(path, info, err)
+	}
+	for _, name := range names {
+		filename := filepath.Join(path, name)
+		fileInfo, err := os.Lstat(filename)
+		if err != nil {
+			if err := walkFn(filename, fileInfo, err); err != nil && err != ErrSkipDir && err != ErrSkipFile {
+				return err
+			}
+		} else {
+			err = walk(filename, fileInfo, walkFn)
+			if err != nil {
+				if err == ErrSkipDir || err == ErrSkipFile {
+					return nil
+				}
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 // walk recursively descends path, calling w.
 func walk(path string, info os.FileInfo, walkFn WalkFunc) error {
 	err := walkFn(path, info, nil)