func()

in cmd/zc_traverser_s3.go [66:186]
88 lines of code
18 McCabe index (conditional complexity)

func (t *s3Traverser) Traverse(preprocessor objectMorpher, processor objectProcessor, filters []ObjectFilter) (err error) {
	invalidAzureBlobName := func(objectKey string) bool {
		/* S3 object name is invalid if it ends with period or
		   one of virtual directories in path ends with period.
		   This list is not exhaustive
		*/
		return strings.HasSuffix(objectKey, ".") ||
			strings.Contains(objectKey, "./")
	}
	invalidNameErrorMsg := "Skipping S3 object %s, as it is not a valid Blob name. Rename the object and retry the transfer"
	// Check if resource is a single object.
	if t.s3URLParts.IsObjectSyntactically() && !t.s3URLParts.IsDirectorySyntactically() && !t.s3URLParts.IsBucketSyntactically() {
		objectPath := strings.Split(t.s3URLParts.ObjectKey, "/")
		objectName := objectPath[len(objectPath)-1]

		oi, err := t.s3Client.StatObject(t.s3URLParts.BucketName, t.s3URLParts.ObjectKey, minio.StatObjectOptions{})
		if invalidAzureBlobName(t.s3URLParts.ObjectKey) {
			WarnStdoutAndScanningLog(fmt.Sprintf(invalidNameErrorMsg, t.s3URLParts.ObjectKey))
			return common.EAzError.InvalidBlobName()
		}

		// If we actually got object properties, process them.
		// Otherwise, treat it as a directory.
		// According to IsDirectorySyntactically, objects and folders can share names
		if err == nil {
			// We had to statObject anyway, get ALL the info.
			oie := common.ObjectInfoExtension{ObjectInfo: oi}
			storedObject := newStoredObject(
				preprocessor,
				objectName,
				"",
				common.EEntityType.File(),
				oi.LastModified,
				oi.Size,
				&oie,
				noBlobProps,
				oie.NewCommonMetadata(),
				t.s3URLParts.BucketName)

			err = processIfPassedFilters(
				filters,
				storedObject,
				processor)
			_, err = getProcessingError(err)
			if err != nil {
				return err
			}

			return nil
		}
	}

	// Append a trailing slash if it is missing.
	if !strings.HasSuffix(t.s3URLParts.ObjectKey, "/") && t.s3URLParts.ObjectKey != "" {
		t.s3URLParts.ObjectKey += "/"
	}

	// Ignore *s in URLs and treat them as normal characters
	// This is because * is both a valid URL path character and a valid portion of an object key in S3.
	searchPrefix := t.s3URLParts.ObjectKey

	// It's a bucket or virtual directory.
	for objectInfo := range t.s3Client.ListObjectsV2(t.s3URLParts.BucketName, searchPrefix, t.recursive, t.ctx.Done()) {
		if objectInfo.Err != nil {
			return fmt.Errorf("cannot list objects, %v", objectInfo.Err)
		}

		if objectInfo.StorageClass == "" {
			// Directories are the only objects without storage classes.
			continue
		}

		if invalidAzureBlobName(objectInfo.Key) {
			//Throw a warning on console and continue
			WarnStdoutAndScanningLog(fmt.Sprintf(invalidNameErrorMsg, objectInfo.Key))
			continue
		}

		objectPath := strings.Split(objectInfo.Key, "/")
		objectName := objectPath[len(objectPath)-1]

		// re-join the unescaped path.
		relativePath := strings.TrimPrefix(objectInfo.Key, searchPrefix)

		if strings.HasSuffix(relativePath, "/") {
			// If a file has a suffix of /, it's still treated as a folder.
			// Thus, akin to the old code. skip it.
			continue
		}

		// default to empty props, but retrieve real ones if required
		oie := common.ObjectInfoExtension{ObjectInfo: minio.ObjectInfo{}}
		if t.getProperties {
			oi, err := t.s3Client.StatObject(t.s3URLParts.BucketName, objectInfo.Key, minio.StatObjectOptions{})
			if err != nil {
				return err
			}
			oie = common.ObjectInfoExtension{ObjectInfo: oi}
		}
		storedObject := newStoredObject(
			preprocessor,
			objectName,
			relativePath,
			common.EEntityType.File(),
			objectInfo.LastModified,
			objectInfo.Size,
			&oie,
			noBlobProps,
			oie.NewCommonMetadata(),
			t.s3URLParts.BucketName)

		err = processIfPassedFilters(filters,
			storedObject,
			processor)
		_, err = getProcessingError(err)
		if err != nil {
			return
		}
	}
	return
}