4273: Added a new create method to pass offset and length of sub arra…

…y. (#4407)

* 4273: Added a new create method to pass offset and lenght of sub array.

* 4273: Fixed codeformat error.

* 4273: Rephrased a comment.

* 4273: Added a new integration test using the new createBlobWithSubArrayFromByteArray code snippet.

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

@@ -1442,6 +1442,30 @@ public static Builder newBuilder() {
    */
   Blob create(BlobInfo blobInfo, byte[] content, BlobTargetOption... options);
 
+  /**
+   * Creates a new blob with the sub array of the given byte array. Direct upload is used to upload
+   * {@code content}. For large content, {@link #writer} is recommended as it uses resumable upload.
+   * MD5 and CRC32C hashes of {@code content} are computed and used for validating transferred data.
+   * Accepts a userProject {@link BlobGetOption} option, which defines the project id to assign
+   * operational costs.
+   *
+   * <p>Example of creating a blob from a byte array.
+   *
+   * <pre>{@code
+   * String bucketName = "my_unique_bucket";
+   * String blobName = "my_blob_name";
+   * BlobId blobId = BlobId.of(bucketName, blobName);
+   * BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setContentType("text/plain").build();
+   * Blob blob = storage.create(blobInfo, "Hello, World!".getBytes(UTF_8), 7, 5);
+   * }</pre>
+   *
+   * @return a [@code Blob} with complete information
+   * @throws StorageException upon failure
+   * @see <a href="https://cloud.google.com/storage/docs/hashes-etags">Hashes and ETags</a>
+   */
+  Blob create(
+      BlobInfo blobInfo, byte[] content, int offset, int length, BlobTargetOption... options);
+
   /**
    * Creates a new blob. Direct upload is used to upload {@code content}. For large content, {@link
    * #writer} is recommended as it uses resumable upload. By default any md5 and crc32c values in

google-cloud-clients/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java

@@ -149,6 +149,22 @@ public Blob create(BlobInfo blobInfo, byte[] content, BlobTargetOption... option
     return internalCreate(updatedInfo, content, options);
   }
 
+  @Override
+  public Blob create(
+      BlobInfo blobInfo, byte[] content, int offset, int length, BlobTargetOption... options) {
+    content = firstNonNull(content, EMPTY_BYTE_ARRAY);
+    byte[] subContent = Arrays.copyOfRange(content, offset, offset + length);
+    BlobInfo updatedInfo =
+        blobInfo
+            .toBuilder()
+            .setMd5(BaseEncoding.base64().encode(Hashing.md5().hashBytes(subContent).asBytes()))
+            .setCrc32c(
+                BaseEncoding.base64()
+                    .encode(Ints.toByteArray(Hashing.crc32c().hashBytes(subContent).asInt())))
+            .build();
+    return internalCreate(updatedInfo, subContent, options);
+  }
+
   @Override
   @Deprecated
   public Blob create(BlobInfo blobInfo, InputStream content, BlobWriteOption... options) {

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

@@ -97,8 +97,11 @@ public class StorageImplTest {
   private static final String BLOB_NAME2 = "n2";
   private static final String BLOB_NAME3 = "n3";
   private static final byte[] BLOB_CONTENT = {0xD, 0xE, 0xA, 0xD};
+  private static final byte[] BLOB_SUB_CONTENT = {0xE, 0xA};
   private static final String CONTENT_MD5 = "O1R4G1HJSDUISJjoIYmVhQ==";
   private static final String CONTENT_CRC32C = "9N3EPQ==";
+  private static final String SUB_CONTENT_MD5 = "5e7c7CdasUiOn3BO560jPg==";
+  private static final String SUB_CONTENT_CRC32C = "bljNYA==";
   private static final int DEFAULT_CHUNK_SIZE = 2 * 1024 * 1024;
   private static final String BASE64_KEY = "JVzfVl8NLD9FjedFuStegjRfES5ll5zc59CIXw572OA=";
   private static final Key KEY =
@@ -445,6 +448,34 @@ public void testCreateBlob() throws IOException {
     assertEquals(-1, byteStream.read(streamBytes));
   }
 
+  @Test
+  public void testCreateBlobWithSubArrayFromByteArray() throws IOException {
+    Capture<ByteArrayInputStream> capturedStream = Capture.newInstance();
+    EasyMock.expect(
+            storageRpcMock.create(
+                EasyMock.eq(
+                    BLOB_INFO1
+                        .toBuilder()
+                        .setMd5(SUB_CONTENT_MD5)
+                        .setCrc32c(SUB_CONTENT_CRC32C)
+                        .build()
+                        .toPb()),
+                EasyMock.capture(capturedStream),
+                EasyMock.eq(EMPTY_RPC_OPTIONS)))
+        .andReturn(BLOB_INFO1.toPb());
+    EasyMock.replay(storageRpcMock);
+    initializeService();
+
+    Blob blob = storage.create(BLOB_INFO1, BLOB_CONTENT, 1, 2);
+
+    assertEquals(expectedBlob1, blob);
+    ByteArrayInputStream byteStream = capturedStream.getValue();
+    byte[] streamBytes = new byte[BLOB_SUB_CONTENT.length];
+    assertEquals(BLOB_SUB_CONTENT.length, byteStream.read(streamBytes));
+    assertArrayEquals(BLOB_SUB_CONTENT, streamBytes);
+    assertEquals(-1, byteStream.read(streamBytes));
+  }
+
   @Test
   public void testCreateBlobRetry() throws IOException {
     Capture<ByteArrayInputStream> capturedStream1 = Capture.newInstance();

google-cloud-examples/src/main/java/com/google/cloud/examples/storage/snippets/StorageSnippets.java

@@ -132,6 +132,20 @@ public Blob createBlobFromByteArray(String bucketName, String blobName) {
     return blob;
   }
 
+  /** Example of creating a blob with sub array from a byte array. */
+  // [TARGET create(BlobInfo, byte[], offset, length, BlobTargetOption...)]
+  // [VARIABLE "my_unique_bucket"]
+  // [VARIABLE "my_blob_name"]
+  public Blob createBlobWithSubArrayFromByteArray(
+      String bucketName, String blobName, int offset, int length) {
+    // [START createBlobWithSubArrayFromByteArray]
+    BlobId blobId = BlobId.of(bucketName, blobName);
+    BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setContentType("text/plain").build();
+    Blob blob = storage.create(blobInfo, "Hello, World!".getBytes(UTF_8), offset, length);
+    // [END createBlobWithSubArrayFromByteArray]
+    return blob;
+  }
+
   /** Example of creating a blob from an input stream. */
   // [TARGET create(BlobInfo, InputStream, BlobWriteOption...)]
   // [VARIABLE "my_unique_bucket"]

google-cloud-examples/src/test/java/com/google/cloud/examples/storage/snippets/ITStorageSnippets.java

@@ -206,6 +206,24 @@ public void testCreateCopyAndGetBlob() {
     copiedBlob.delete();
   }
 
+  @Test
+  public void testCreateCopyAndGetBlobFromSubArray() {
+    String blobName = "test-create-copy-get-blob-from-sub-array";
+    Blob blob = storageSnippets.createBlobWithSubArrayFromByteArray(BUCKET, blobName, 7, 1);
+    assertNotNull(blob);
+    Blob copiedBlob = storageSnippets.copyBlobInChunks(BUCKET, blobName, "copy-blob");
+    assertNotNull(copiedBlob);
+    try {
+      storageSnippets.getBlobFromIdWithMetageneration(BUCKET, blobName, -1);
+      fail("Expected StorageException to be thrown");
+    } catch (StorageException ex) {
+      // expected
+    }
+    assertTrue(
+        storageSnippets.deleteBlobFromIdWithGeneration(BUCKET, blobName, blob.getGeneration()));
+    copiedBlob.delete();
+  }
+
   @Test
   public void testCreateBlobFromInputStream() {
     Blob blob =