S3 API: fix DeleteMultipleObjectsHandler

fix https://github.com/chrislusf/seaweedfs/issues/1241

weed/s3api/filer_multipart.go

@@ -112,7 +112,7 @@ func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploa
 		},
 	}
 
-	if err = s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true, false, true); err != nil {
+	if err = s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, false, true); err != nil {
 		glog.V(1).Infof("completeMultipartUpload cleanup %s upload %s: %v", *input.Bucket, *input.UploadId, err)
 	}
 
@@ -127,7 +127,7 @@ func (s3a *S3ApiServer) abortMultipartUpload(input *s3.AbortMultipartUploadInput
 		return nil, ErrNoSuchUpload
 	}
 	if exists {
-		err = s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true, true, true)
+		err = s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true, true)
 	}
 	if err != nil {
 		glog.V(1).Infof("bucket %s remove upload %s: %v", *input.Bucket, *input.UploadId, err)

weed/s3api/filer_util.go

@@ -117,21 +117,13 @@ func (s3a *S3ApiServer) list(parentDirectoryPath, prefix, startFrom string, incl
 
 }
 
-func (s3a *S3ApiServer) rm(parentDirectoryPath, entryName string, isDirectory, isDeleteData, isRecursive bool) error {
+func (s3a *S3ApiServer) rm(parentDirectoryPath, entryName string, isDeleteData, isRecursive bool) error {
 
 	return s3a.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
-		request := &filer_pb.DeleteEntryRequest{
-			Directory:    parentDirectoryPath,
-			Name:         entryName,
-			IsDeleteData: isDeleteData,
-			IsRecursive:  isRecursive,
-		}
-
-		glog.V(1).Infof("delete entry %v/%v: %v", parentDirectoryPath, entryName, request)
-		if _, err := client.DeleteEntry(context.Background(), request); err != nil {
-			glog.V(0).Infof("delete entry %v: %v", request, err)
-			return fmt.Errorf("delete entry %s/%s: %v", parentDirectoryPath, entryName, err)
+		err := doDeleteEntry(client, parentDirectoryPath, entryName, isDeleteData, isRecursive)
+		if err != nil {
+			return err
 		}
 
 		return nil
@@ -139,57 +131,24 @@ func (s3a *S3ApiServer) rm(parentDirectoryPath, entryName string, isDirectory, i
 
 }
 
-func (s3a *S3ApiServer) streamRemove(quiet bool, fn func() (finished bool, parentDirectoryPath string, entryName string, isDeleteData, isRecursive bool), respFn func(err string)) error {
+func doDeleteEntry(client filer_pb.SeaweedFilerClient, parentDirectoryPath string, entryName string, isDeleteData bool, isRecursive bool) error {
+	request := &filer_pb.DeleteEntryRequest{
+		Directory:    parentDirectoryPath,
+		Name:         entryName,
+		IsDeleteData: isDeleteData,
+		IsRecursive:  isRecursive,
+	}
 
-	return s3a.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
-
-		stream, err := client.StreamDeleteEntries(context.Background())
-		if err != nil {
-			glog.V(0).Infof("stream delete entry: %v", err)
-			return fmt.Errorf("stream delete entry: %v", err)
+	glog.V(1).Infof("delete entry %v/%v: %v", parentDirectoryPath, entryName, request)
+	if resp, err := client.DeleteEntry(context.Background(), request); err != nil {
+		glog.V(0).Infof("delete entry %v: %v", request, err)
+		return fmt.Errorf("delete entry %s/%s: %v", parentDirectoryPath, entryName, err)
+	} else {
+		if resp.Error != "" {
+			return fmt.Errorf("delete entry %s/%s: %v", parentDirectoryPath, entryName, resp.Error)
 		}
-
-		waitc := make(chan struct{})
-		go func() {
-			for {
-				resp, err := stream.Recv()
-				if err == io.EOF {
-					// read done.
-					close(waitc)
-					return
-				}
-				if err != nil {
-					glog.V(0).Infof("streamRemove: %v", err)
-					return
-				}
-				respFn(resp.Error)
-			}
-		}()
-
-		for {
-			finished, parentDirectoryPath, entryName, isDeleteData, isRecursive := fn()
-			if finished {
-				break
-			}
-			err = stream.Send(&filer_pb.DeleteEntryRequest{
-				Directory:            parentDirectoryPath,
-				Name:                 entryName,
-				IsDeleteData:         isDeleteData,
-				IsRecursive:          isRecursive,
-				IgnoreRecursiveError: quiet,
-			})
-			if err != nil {
-				glog.V(0).Infof("streamRemove: %v", err)
-				break
-			}
-
-		}
-		stream.CloseSend()
-		<-waitc
-		return err
-
-	})
-
+	}
+	return nil
 }
 
 func (s3a *S3ApiServer) exists(parentDirectoryPath string, entryName string, isDirectory bool) (exists bool, err error) {

weed/s3api/s3api_bucket_handlers.go

@@ -94,7 +94,7 @@ func (s3a *S3ApiServer) DeleteBucketHandler(w http.ResponseWriter, r *http.Reque
 		return nil
 	})
 
-	err = s3a.rm(s3a.option.BucketsPath, bucket, true, false, true)
+	err = s3a.rm(s3a.option.BucketsPath, bucket, false, true)
 
 	if err != nil {
 		writeErrorResponse(w, ErrInternalError, r.URL)

weed/s3api/s3api_object_handlers.go

@@ -13,6 +13,7 @@ import (
 	"github.com/gorilla/mux"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/server"
 	"github.com/chrislusf/seaweedfs/weed/util"
 )
@@ -165,38 +166,32 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
 		return
 	}
 
-	var index int
-
 	var deletedObjects []ObjectIdentifier
 	var deleteErrors []DeleteError
-	s3a.streamRemove(deleteObjects.Quiet, func() (finished bool, parentDirectoryPath string, entryName string, isDeleteData, isRecursive bool) {
-		if index >= len(deleteObjects.Objects) {
-			finished = true
-			return
-		}
 
-		object := deleteObjects.Objects[index]
+	s3a.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
-		lastSeparator := strings.LastIndex(object.ObjectName, "/")
-		parentDirectoryPath, entryName, isDeleteData, isRecursive = "/", object.ObjectName, true, false
-		if lastSeparator > 0 && lastSeparator+1 < len(object.ObjectName) {
-			entryName = object.ObjectName[lastSeparator+1:]
-			parentDirectoryPath = "/" + object.ObjectName[:lastSeparator]
+		for _, object := range deleteObjects.Objects {
+			lastSeparator := strings.LastIndex(object.ObjectName, "/")
+			parentDirectoryPath, entryName, isDeleteData, isRecursive := "/", object.ObjectName, true, false
+			if lastSeparator > 0 && lastSeparator+1 < len(object.ObjectName) {
+				entryName = object.ObjectName[lastSeparator+1:]
+				parentDirectoryPath = "/" + object.ObjectName[:lastSeparator]
+			}
+			parentDirectoryPath = fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, parentDirectoryPath)
+
+			err := doDeleteEntry(client, parentDirectoryPath, entryName, isDeleteData, isRecursive)
+			if err == nil {
+				deletedObjects = append(deletedObjects, object)
+			} else {
+				deleteErrors = append(deleteErrors, DeleteError{
+					Code:    "",
+					Message: err.Error(),
+					Key:     object.ObjectName,
+				})
+			}
 		}
-		parentDirectoryPath = fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, parentDirectoryPath)
-		return
-	}, func(err string) {
-		object := deleteObjects.Objects[index]
-		if err == "" {
-			deletedObjects = append(deletedObjects, object)
-		} else {
-			deleteErrors = append(deleteErrors, DeleteError{
-				Code:    "",
-				Message: err,
-				Key:     object.ObjectName,
-			})
-		}
-		index++
+		return nil
 	})
 
 	deleteResp := DeleteObjectsResponse{}