feat: add support of startOffset and endOffset

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

@@ -1032,6 +1032,28 @@ public static BlobListOption delimiter(String delimiter) {
       return new BlobListOption(StorageRpc.Option.DELIMITER, delimiter);
     }
 
+    /**
+     * Returns an option to set a startOffset to filter results to objects whose names are
+     * lexicographically equal to or after startOffset. If endOffset is also set, the objects listed
+     * have names between startOffset (inclusive) and endOffset (exclusive).
+     *
+     * @param startOffset startOffset to filter the results
+     */
+    public static BlobListOption startOffset(String startOffset) {
+      return new BlobListOption(StorageRpc.Option.START_OFF_SET, startOffset);
+    }
+
+    /**
+     * Returns an option to set a endOffset to filter results to objects whose names are
+     * lexicographically before endOffset. If startOffset is also set, the objects listed have names
+     * between startOffset (inclusive) and endOffset (exclusive).
+     *
+     * @param endOffset endOffset to filter the results
+     */
+    public static BlobListOption endOffset(String endOffset) {
+      return new BlobListOption(StorageRpc.Option.END_OFF_SET, endOffset);
+    }
+
     /**
      * Returns an option to define the billing user project. This option is required by buckets with
      * `requester_pays` flag enabled to assign operation costs.

google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java

@@ -350,6 +350,8 @@ public Tuple<String, Iterable<StorageObject>> list(final String bucket, Map<Opti
               .setProjection(DEFAULT_PROJECTION)
               .setVersions(Option.VERSIONS.getBoolean(options))
               .setDelimiter(Option.DELIMITER.getString(options))
+              .setStartOffset(Option.START_OFF_SET.getString(options))
+              .setEndOffset(Option.END_OFF_SET.getString(options))
               .setPrefix(Option.PREFIX.getString(options))
               .setMaxResults(Option.MAX_RESULTS.getLong(options))
               .setPageToken(Option.PAGE_TOKEN.getString(options))

google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/StorageRpc.java

@@ -58,6 +58,8 @@ enum Option {
     MAX_RESULTS("maxResults"),
     PAGE_TOKEN("pageToken"),
     DELIMITER("delimiter"),
+    START_OFF_SET("startOffset"),
+    END_OFF_SET("endOffset"),
     VERSIONS("versions"),
     FIELDS("fields"),
     CUSTOMER_SUPPLIED_KEY("customerSuppliedKey"),

google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplTest.java

@@ -665,6 +665,30 @@ public void testListBlobsDelimiter() {
     assertArrayEquals(blobList.toArray(), Iterables.toArray(page.getValues(), Blob.class));
   }
 
+  @Test
+  public void testListBlobsWithOffset() {
+    String cursor = "cursor";
+    String startOffset = "startOffset";
+    String endOffset = "endOffset";
+    Map<StorageRpc.Option, ?> options =
+        ImmutableMap.of(
+            StorageRpc.Option.START_OFF_SET, startOffset, StorageRpc.Option.END_OFF_SET, endOffset);
+    ImmutableList<BlobInfo> blobInfoList = ImmutableList.of(BLOB_INFO1, BLOB_INFO2);
+    Tuple<String, Iterable<com.google.api.services.storage.model.StorageObject>> result =
+        Tuple.of(cursor, Iterables.transform(blobInfoList, BlobInfo.INFO_TO_PB_FUNCTION));
+    EasyMock.expect(storageRpcMock.list(BUCKET_NAME1, options)).andReturn(result);
+    EasyMock.replay(storageRpcMock);
+    initializeService();
+    ImmutableList<Blob> blobList = ImmutableList.of(expectedBlob1, expectedBlob2);
+    Page<Blob> page =
+        storage.list(
+            BUCKET_NAME1,
+            Storage.BlobListOption.startOffset(startOffset),
+            Storage.BlobListOption.endOffset(endOffset));
+    assertEquals(cursor, page.getNextPageToken());
+    assertArrayEquals(blobList.toArray(), Iterables.toArray(page.getValues(), Blob.class));
+  }
+
   @Test
   public void testUpdateBucket() {
     BucketInfo updatedBucketInfo = BUCKET_INFO1.toBuilder().setIndexPage("some-page").build();

google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITStorageTest.java

@@ -993,6 +993,53 @@ public void testListBlobsVersioned() throws ExecutionException, InterruptedExcep
     }
   }
 
+  @Test
+  public void testListBlobsWithOffset() throws ExecutionException, InterruptedException {
+    String bucketName = RemoteStorageHelper.generateBucketName();
+    Bucket bucket =
+        storage.create(BucketInfo.newBuilder(bucketName).setVersioningEnabled(true).build());
+    try {
+      List<String> blobNames =
+          ImmutableList.of("startOffset_blob1", "startOffset_blob2", "blob3_endOffset");
+      BlobInfo blob1 =
+          BlobInfo.newBuilder(bucket, blobNames.get(0)).setContentType(CONTENT_TYPE).build();
+      BlobInfo blob2 =
+          BlobInfo.newBuilder(bucket, blobNames.get(1)).setContentType(CONTENT_TYPE).build();
+      BlobInfo blob3 =
+          BlobInfo.newBuilder(bucket, blobNames.get(2)).setContentType(CONTENT_TYPE).build();
+
+      Blob remoteBlob1 = storage.create(blob1);
+      Blob remoteBlob2 = storage.create(blob2);
+      Blob remoteBlob3 = storage.create(blob3);
+      assertNotNull(remoteBlob1);
+      assertNotNull(remoteBlob2);
+      assertNotNull(remoteBlob3);
+
+      // Listing blobs without BlobListOptions.
+      Page<Blob> page1 = storage.list(bucketName);
+      assertEquals(3, Iterators.size(page1.iterateAll().iterator()));
+
+      // Listing blobs with startOffset.
+      Page<Blob> page2 =
+          storage.list(bucketName, Storage.BlobListOption.startOffset("startOffset"));
+      assertEquals(2, Iterators.size(page2.iterateAll().iterator()));
+
+      // Listing blobs with endOffset.
+      Page<Blob> page3 = storage.list(bucketName, Storage.BlobListOption.endOffset("endOffset"));
+      assertEquals(1, Iterators.size(page3.iterateAll().iterator()));
+
+      // Listing blobs with startOffset and endOffset.
+      Page<Blob> page4 =
+          storage.list(
+              bucketName,
+              Storage.BlobListOption.startOffset("startOffset"),
+              Storage.BlobListOption.endOffset("endOffset"));
+      assertEquals(0, Iterators.size(page4.iterateAll().iterator()));
+    } finally {
+      RemoteStorageHelper.forceDelete(storage, bucketName, 5, TimeUnit.SECONDS);
+    }
+  }
+
   @Test(timeout = 5000)
   public void testListBlobsCurrentDirectory() throws InterruptedException {
     String directoryName = "test-list-blobs-current-directory/";