Remove BlobListOption.recursive option and fix delimiter handling

- Add BlobListOption.currentDirectory option that sets the '/' delimiter
- Add isDirectory method to BlobInfo objects
- Change StorageRpc.list(bucket) method to add prefixes to the list of storage objects
- Add unit and integration tests

gcloud-java-storage/src/main/java/com/google/gcloud/spi/DefaultStorageRpc.java

@@ -57,8 +57,10 @@
 import com.google.api.services.storage.model.ComposeRequest.SourceObjects.ObjectPreconditions;
 import com.google.api.services.storage.model.Objects;
 import com.google.api.services.storage.model.StorageObject;
+import com.google.common.base.Function;
 import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.gcloud.storage.StorageException;
@@ -67,6 +69,7 @@
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.math.BigInteger;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
@@ -151,7 +154,7 @@ public Tuple<String, Iterable<Bucket>> list(Map<Option, ?> options) {
   }
 
   @Override
-  public Tuple<String, Iterable<StorageObject>> list(String bucket, Map<Option, ?> options) {
+  public Tuple<String, Iterable<StorageObject>> list(final String bucket, Map<Option, ?> options) {
     try {
       Objects objects = storage.objects()
           .list(bucket)
@@ -163,8 +166,20 @@ public Tuple<String, Iterable<StorageObject>> list(String bucket, Map<Option, ?>
           .setPageToken(PAGE_TOKEN.getString(options))
           .setFields(FIELDS.getString(options))
           .execute();
-      return Tuple.<String, Iterable<StorageObject>>of(
-          objects.getNextPageToken(), objects.getItems());
+      Iterable<StorageObject> storageObjects = Iterables.concat(
+          objects.getItems() != null ? objects.getItems() : ImmutableList.<StorageObject>of(),
+          objects.getPrefixes() != null
+              ? Lists.transform(objects.getPrefixes(), new Function<String, StorageObject>() {
+                @Override
+                public StorageObject apply(String prefix) {
+                  return new StorageObject()
+                      .set("isDirectory", true)
+                      .setBucket(bucket)
+                      .setName(prefix)
+                      .setSize(BigInteger.ZERO);
+                }
+              }) : ImmutableList.<StorageObject>of());
+      return Tuple.of(objects.getNextPageToken(), storageObjects);
     } catch (IOException ex) {
       throw translate(ex);
     }

gcloud-java-storage/src/main/java/com/google/gcloud/spi/StorageRpc.java

@@ -344,7 +344,7 @@ void write(String uploadId, byte[] toWrite, int toWriteOffset, long destOffset,
   /**
    * Continues rewriting on an already open rewrite channel.
    *
-   * @throws StorageException
+   * @throws StorageException upon failure
    */
   RewriteResponse continueRewrite(RewriteResponse previousResponse);
 }

gcloud-java-storage/src/main/java/com/google/gcloud/storage/Blob.java

@@ -290,6 +290,12 @@ Builder updateTime(Long updateTime) {
       return this;
     }
 
+    @Override
+    Builder isDirectory(boolean isDirectory) {
+      infoBuilder.isDirectory(isDirectory);
+      return this;
+    }
+
     @Override
     public Blob build() {
       return new Blob(storage, infoBuilder);

gcloud-java-storage/src/main/java/com/google/gcloud/storage/BlobInfo.java

@@ -78,6 +78,7 @@ public StorageObject apply(BlobInfo blobInfo) {
   private final String contentDisposition;
   private final String contentLanguage;
   private final Integer componentCount;
+  private final boolean isDirectory;
 
   /**
    * This class is meant for internal use only. Users are discouraged from using this class.
@@ -187,6 +188,8 @@ public abstract static class Builder {
 
     abstract Builder updateTime(Long updateTime);
 
+    abstract Builder isDirectory(boolean isDirectory);
+
     /**
      * Creates a {@code BlobInfo} object.
      */
@@ -215,6 +218,7 @@ static final class BuilderImpl extends Builder {
     private Long metageneration;
     private Long deleteTime;
     private Long updateTime;
+    private Boolean isDirectory;
 
     BuilderImpl(BlobId blobId) {
       this.blobId = blobId;
@@ -241,6 +245,7 @@ static final class BuilderImpl extends Builder {
       metageneration = blobInfo.metageneration;
       deleteTime = blobInfo.deleteTime;
       updateTime = blobInfo.updateTime;
+      isDirectory = blobInfo.isDirectory;
     }
 
     @Override
@@ -364,6 +369,12 @@ Builder updateTime(Long updateTime) {
       return this;
     }
 
+    @Override
+    Builder isDirectory(boolean isDirectory) {
+      this.isDirectory = isDirectory;
+      return this;
+    }
+
     @Override
     public BlobInfo build() {
       checkNotNull(blobId);
@@ -392,6 +403,7 @@ public BlobInfo build() {
     metageneration = builder.metageneration;
     deleteTime = builder.deleteTime;
     updateTime = builder.updateTime;
+    isDirectory = firstNonNull(builder.isDirectory, Boolean.FALSE);
   }
 
   /**
@@ -588,6 +600,18 @@ public Long updateTime() {
     return updateTime;
   }
 
+  /**
+   * Returns {@code true} if the current blob represents a directory. This can only happen if the
+   * blob is returned by {@link Storage#list(String, Storage.BlobListOption...)} when the
+   * {@link Storage.BlobListOption#currentDirectory()} option is used. If {@code true} only
+   * {@link #blobId()} and {@link #size()} are set for the current blob: {@link BlobId#name()} ends
+   * with the '/' character, {@link BlobId#generation()} returns {@code null} and {@link #size()} is
+   * {@code 0}.
+   */
+  public boolean isDirectory() {
+    return isDirectory;
+  }
+
   /**
    * Returns a builder for the current blob.
    */
@@ -761,6 +785,9 @@ public Acl apply(ObjectAccessControl objectAccessControl) {
         }
       }));
     }
+    if (storageObject.get("isDirectory") != null) {
+      builder.isDirectory(Boolean.TRUE);
+    }
     return builder.build();
   }
 }

gcloud-java-storage/src/main/java/com/google/gcloud/storage/Storage.java

@@ -694,10 +694,17 @@ public static BlobListOption prefix(String prefix) {
     }
 
     /**
-     * Returns an option to specify whether blob listing should include subdirectories or not.
+     * If specified, results are returned in a directory-like mode. Blobs whose names, aside from
+     * a possible {@link #prefix(String)}, do not contain the '/' delimiter are returned as is.
+     * Blobs whose names, aside from a possible {@link #prefix(String)}, contain the '/' delimiter,
+     * will have their name truncated after the delimiter and will be returned as {@link Blob}
+     * objects where only {@link Blob#blobId()}, {@link Blob#size()} and {@link Blob#isDirectory()}
+     * are set. For such directory blobs, ({@link BlobId#generation()} returns {@code null}),
+     * {@link Blob#size()} returns {@code 0} while {@link Blob#isDirectory()} returns {@code true}.
+     * Duplicate directory blobs are omitted.
      */
-    public static BlobListOption recursive(boolean recursive) {
-      return new BlobListOption(StorageRpc.Option.DELIMITER, recursive);
+    public static BlobListOption currentDirectory() {
+      return new BlobListOption(StorageRpc.Option.DELIMITER, true);
     }
 
     /**
@@ -1289,7 +1296,8 @@ private static void checkContentType(BlobInfo blobInfo) throws IllegalArgumentEx
   Page<Bucket> list(BucketListOption... options);
 
   /**
-   * Lists the bucket's blobs.
+   * Lists the bucket's blobs. If the {@link BlobListOption#currentDirectory()} option is provided,
+   * results are returned in a directory-like mode.
    *
    * @throws StorageException upon failure
    */

gcloud-java-storage/src/main/java/com/google/gcloud/storage/StorageImpl.java

@@ -76,6 +76,7 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   private static final byte[] EMPTY_BYTE_ARRAY = {};
   private static final String EMPTY_BYTE_ARRAY_MD5 = "1B2M2Y8AsgTpgAmY7PhCfg==";
   private static final String EMPTY_BYTE_ARRAY_CRC32C = "AAAAAA==";
+  private static final String PATH_DELIMITER = "/";
 
   private static final Function<Tuple<Storage, Boolean>, Boolean> DELETE_FUNCTION =
       new Function<Tuple<Storage, Boolean>, Boolean>() {
@@ -669,7 +670,7 @@ private static <T> void addToOptionMap(StorageRpc.Option getOption, StorageRpc.O
     }
     Boolean value = (Boolean) temp.remove(DELIMITER);
     if (Boolean.TRUE.equals(value)) {
-      temp.put(DELIMITER, options().pathDelimiter());
+      temp.put(DELIMITER, PATH_DELIMITER);
     }
     if (useAsSource) {
       addToOptionMap(IF_GENERATION_MATCH, IF_SOURCE_GENERATION_MATCH, generation, temp);

gcloud-java-storage/src/main/java/com/google/gcloud/storage/StorageOptions.java

@@ -16,24 +16,19 @@
 
 package com.google.gcloud.storage;
 
-import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableSet;
 import com.google.gcloud.ServiceOptions;
 import com.google.gcloud.spi.DefaultStorageRpc;
 import com.google.gcloud.spi.StorageRpc;
 import com.google.gcloud.spi.StorageRpcFactory;
 
-import java.util.Objects;
 import java.util.Set;
 
 public class StorageOptions extends ServiceOptions<Storage, StorageRpc, StorageOptions> {
 
   private static final long serialVersionUID = -7804860602287801084L;
   private static final String GCS_SCOPE = "https://www.googleapis.com/auth/devstorage.full_control";
   private static final Set<String> SCOPES = ImmutableSet.of(GCS_SCOPE);
-  private static final String DEFAULT_PATH_DELIMITER = "/";
-
-  private final String pathDelimiter;
 
   public static class DefaultStorageFactory implements StorageFactory {
 
@@ -58,24 +53,10 @@ public StorageRpc create(StorageOptions options) {
   public static class Builder extends
       ServiceOptions.Builder<Storage, StorageRpc, StorageOptions, Builder> {
 
-    private String pathDelimiter;
-
     private Builder() {}
 
     private Builder(StorageOptions options) {
       super(options);
-      pathDelimiter = options.pathDelimiter;
-    }
-
-    /**
-     * Sets the path delimiter for the storage service.
-     *
-     * @param pathDelimiter the path delimiter to set
-     * @return the builder
-     */
-    public Builder pathDelimiter(String pathDelimiter) {
-      this.pathDelimiter = pathDelimiter;
-      return this;
     }
 
     @Override
@@ -86,7 +67,6 @@ public StorageOptions build() {
 
   private StorageOptions(Builder builder) {
     super(StorageFactory.class, StorageRpcFactory.class, builder);
-    pathDelimiter = MoreObjects.firstNonNull(builder.pathDelimiter, DEFAULT_PATH_DELIMITER);
   }
 
   @SuppressWarnings("unchecked")
@@ -106,13 +86,6 @@ protected Set<String> scopes() {
     return SCOPES;
   }
 
-  /**
-   * Returns the storage service's path delimiter.
-   */
-  public String pathDelimiter() {
-    return pathDelimiter;
-  }
-
   /**
    * Returns a default {@code StorageOptions} instance.
    */
@@ -128,7 +101,7 @@ public Builder toBuilder() {
 
   @Override
   public int hashCode() {
-    return baseHashCode() ^ Objects.hash(pathDelimiter);
+    return baseHashCode();
   }
 
   @Override
@@ -137,7 +110,7 @@ public boolean equals(Object obj) {
       return false;
     }
     StorageOptions other = (StorageOptions) obj;
-    return baseEquals(other) && Objects.equals(pathDelimiter, other.pathDelimiter);
+    return baseEquals(other);
   }
 
   public static Builder builder() {

gcloud-java-storage/src/test/java/com/google/gcloud/storage/BlobInfoTest.java

@@ -20,14 +20,19 @@
 import static com.google.gcloud.storage.Acl.Role.READER;
 import static com.google.gcloud.storage.Acl.Role.WRITER;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
 
+import com.google.api.services.storage.model.StorageObject;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.gcloud.storage.Acl.Project;
 import com.google.gcloud.storage.Acl.User;
 
 import org.junit.Test;
 
+import java.math.BigInteger;
 import java.util.List;
 import java.util.Map;
 
@@ -76,6 +81,10 @@ public class BlobInfoTest {
       .size(SIZE)
       .updateTime(UPDATE_TIME)
       .build();
+  private static final BlobInfo DIRECTORY_INFO = BlobInfo.builder("b", "n/")
+      .size(0L)
+      .isDirectory(true)
+      .build();
 
   @Test
   public void testToBuilder() {
@@ -118,6 +127,30 @@ public void testBuilder() {
     assertEquals(SELF_LINK, BLOB_INFO.selfLink());
     assertEquals(SIZE, BLOB_INFO.size());
     assertEquals(UPDATE_TIME, BLOB_INFO.updateTime());
+    assertFalse(BLOB_INFO.isDirectory());
+    assertEquals("b", DIRECTORY_INFO.bucket());
+    assertEquals("n/", DIRECTORY_INFO.name());
+    assertNull(DIRECTORY_INFO.acl());
+    assertNull(DIRECTORY_INFO.componentCount());
+    assertNull(DIRECTORY_INFO.contentType());
+    assertNull(DIRECTORY_INFO.cacheControl());
+    assertNull(DIRECTORY_INFO.contentDisposition());
+    assertNull(DIRECTORY_INFO.contentEncoding());
+    assertNull(DIRECTORY_INFO.contentLanguage());
+    assertNull(DIRECTORY_INFO.crc32c());
+    assertNull(DIRECTORY_INFO.deleteTime());
+    assertNull(DIRECTORY_INFO.etag());
+    assertNull(DIRECTORY_INFO.generation());
+    assertNull(DIRECTORY_INFO.id());
+    assertNull(DIRECTORY_INFO.md5());
+    assertNull(DIRECTORY_INFO.mediaLink());
+    assertNull(DIRECTORY_INFO.metadata());
+    assertNull(DIRECTORY_INFO.metageneration());
+    assertNull(DIRECTORY_INFO.owner());
+    assertNull(DIRECTORY_INFO.selfLink());
+    assertEquals(0L, (long) DIRECTORY_INFO.size());
+    assertNull(DIRECTORY_INFO.updateTime());
+    assertTrue(DIRECTORY_INFO.isDirectory());
   }
 
   private void compareBlobs(BlobInfo expected, BlobInfo value) {
@@ -151,6 +184,35 @@ public void testToPbAndFromPb() {
     compareBlobs(BLOB_INFO, BlobInfo.fromPb(BLOB_INFO.toPb()));
     BlobInfo blobInfo = BlobInfo.builder(BlobId.of("b", "n")).build();
     compareBlobs(blobInfo, BlobInfo.fromPb(blobInfo.toPb()));
+    StorageObject object = new StorageObject()
+        .setName("n/")
+        .setBucket("b")
+        .setSize(BigInteger.ZERO)
+        .set("isDirectory", true);
+    blobInfo = BlobInfo.fromPb(object);
+    assertEquals("b", blobInfo.bucket());
+    assertEquals("n/", blobInfo.name());
+    assertNull(blobInfo.acl());
+    assertNull(blobInfo.componentCount());
+    assertNull(blobInfo.contentType());
+    assertNull(blobInfo.cacheControl());
+    assertNull(blobInfo.contentDisposition());
+    assertNull(blobInfo.contentEncoding());
+    assertNull(blobInfo.contentLanguage());
+    assertNull(blobInfo.crc32c());
+    assertNull(blobInfo.deleteTime());
+    assertNull(blobInfo.etag());
+    assertNull(blobInfo.generation());
+    assertNull(blobInfo.id());
+    assertNull(blobInfo.md5());
+    assertNull(blobInfo.mediaLink());
+    assertNull(blobInfo.metadata());
+    assertNull(blobInfo.metageneration());
+    assertNull(blobInfo.owner());
+    assertNull(blobInfo.selfLink());
+    assertEquals(0L, (long) blobInfo.size());
+    assertNull(blobInfo.updateTime());
+    assertTrue(blobInfo.isDirectory());
   }
 
   @Test