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/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,28 @@ public void testListBlobsDelimiter() {
     assertArrayEquals(blobList.toArray(), Iterables.toArray(page.getValues(), Blob.class));
   }
 
+  @Test
+  public void testListBlobsWithOffset() {
+    String cursor = "cursor";
+    Map<StorageRpc.Option, ?> options =
+        ImmutableMap.of(
+            StorageRpc.Option.START_OFF_SET, BLOB_NAME1, StorageRpc.Option.END_OFF_SET, BLOB_NAME2);
+    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(BLOB_NAME1),
+            Storage.BlobListOption.endOffset(BLOB_NAME2));
+    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,56 @@ 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 {
+      String[] blobNames = {
+        "test-list-blobs-start-offset-blob1",
+        "test-list-blobs-start-offset-blob2",
+        "test-list-blobs-end-offset-blob3"
+      };
+      BlobInfo blob1 =
+          BlobInfo.newBuilder(bucket, blobNames[0]).setContentType(CONTENT_TYPE).build();
+      BlobInfo blob2 =
+          BlobInfo.newBuilder(bucket, blobNames[1]).setContentType(CONTENT_TYPE).build();
+      BlobInfo blob3 =
+          BlobInfo.newBuilder(bucket, blobNames[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);
+      Page<Blob> page =
+          storage.list(
+              bucketName,
+              Storage.BlobListOption.startOffset("test-list-blobs-start-offset-blob"),
+              Storage.BlobListOption.endOffset("test-list-blobs-end-offset-blob3"));
+      // Listing blobs is eventually consistent, we loop until the list is of the expected size.
+      while (Iterators.size(page.iterateAll().iterator()) != 3) {
+        page =
+            storage.list(
+                bucketName,
+                Storage.BlobListOption.startOffset("test-list-blobs-start-offset-blob"),
+                Storage.BlobListOption.endOffset("test-list-blobs-end-offset-blob3"));
+      }
+      Set<String> blobSet = ImmutableSet.of(blobNames[0], blobNames[1], blobNames[2]);
+      Iterator<Blob> iterator = page.iterateAll().iterator();
+      while (iterator.hasNext()) {
+        Blob remoteBlob = iterator.next();
+        assertEquals(bucketName, remoteBlob.getBucket());
+        assertTrue(blobSet.contains(remoteBlob.getName()));
+        assertNotNull(remoteBlob.getGeneration());
+      }
+    } finally {
+      RemoteStorageHelper.forceDelete(storage, bucketName, 5, TimeUnit.SECONDS);
+    }
+  }
+
   @Test(timeout = 5000)
   public void testListBlobsCurrentDirectory() throws InterruptedException {
     String directoryName = "test-list-blobs-current-directory/";