Storage: Give users the ability to disable gzip content encoding to increase throughput (#4170)

* 3822: Added BlobTargetOption for user to set disabledGzipContent true

* 3822: Changed method comment

* 3822: Changed method comment with more explanation

* 3822: Fixed comments after review. Added unit test

* 3822: Fixed formatting errors in Storage class

* 3822: Fixed formatting errors in StorageImplTest class

Author: andrey-qlogic

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

@@ -346,6 +346,14 @@ public static BlobTargetOption metagenerationNotMatch() {
       return new BlobTargetOption(StorageRpc.Option.IF_METAGENERATION_NOT_MATCH);
     }
 
+    /**
+     * Returns an option for blob's data disabledGzipContent. If this option is used, the request
+     * will create a blob with disableGzipContent; at present, this is only for upload.
+     */
+    public static BlobTargetOption disableGzipContent() {
+      return new BlobTargetOption(StorageRpc.Option.IF_DISABLE_GZIP_CONTENT, true);
+    }
+
     /**
      * Returns an option to set a customer-supplied AES256 key for server-side encryption of the
      * blob.

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

@@ -287,6 +287,9 @@ public StorageObject create(
                   storageObject,
                   new InputStreamContent(storageObject.getContentType(), content));
       insert.getMediaHttpUploader().setDirectUploadEnabled(true);
+      Boolean disableGzipContent = Option.IF_DISABLE_GZIP_CONTENT.getBoolean(options);
+      if (disableGzipContent != null)
+        insert.getMediaHttpUploader().setDisableGZipContent(disableGzipContent);
       setEncryptionHeaders(insert.getRequestHeaders(), ENCRYPTION_KEY_PREFIX, options);
       return insert
           .setProjection(DEFAULT_PROJECTION)

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

@@ -48,6 +48,7 @@ enum Option {
     IF_SOURCE_METAGENERATION_NOT_MATCH("ifSourceMetagenerationNotMatch"),
     IF_SOURCE_GENERATION_MATCH("ifSourceGenerationMatch"),
     IF_SOURCE_GENERATION_NOT_MATCH("ifSourceGenerationNotMatch"),
+    IF_DISABLE_GZIP_CONTENT("disableGzipContent"),
     PREFIX("prefix"),
     MAX_RESULTS("maxResults"),
     PAGE_TOKEN("pageToken"),

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

@@ -154,6 +154,8 @@ public class StorageImplTest {
   private static final BlobTargetOption BLOB_TARGET_GENERATION = BlobTargetOption.generationMatch();
   private static final BlobTargetOption BLOB_TARGET_METAGENERATION =
       BlobTargetOption.metagenerationMatch();
+  private static final BlobTargetOption BLOB_TARGET_DISABLE_GZIP_CONTENT =
+      BlobTargetOption.disableGzipContent();
   private static final BlobTargetOption BLOB_TARGET_NOT_EXIST = BlobTargetOption.doesNotExist();
   private static final BlobTargetOption BLOB_TARGET_PREDEFINED_ACL =
       BlobTargetOption.predefinedAcl(Storage.PredefinedAcl.PRIVATE);
@@ -162,6 +164,8 @@ public class StorageImplTest {
           StorageRpc.Option.IF_METAGENERATION_MATCH, BLOB_INFO1.getMetageneration(),
           StorageRpc.Option.IF_GENERATION_MATCH, 0L,
           StorageRpc.Option.PREDEFINED_ACL, BUCKET_TARGET_PREDEFINED_ACL.getValue());
+  private static final Map<StorageRpc.Option, ?> BLOB_TARGET_OPTIONS_CREATE_DISABLE_GZIP_CONTENT =
+      ImmutableMap.of(StorageRpc.Option.IF_DISABLE_GZIP_CONTENT, true);
   private static final Map<StorageRpc.Option, ?> BLOB_TARGET_OPTIONS_UPDATE =
       ImmutableMap.of(
           StorageRpc.Option.IF_METAGENERATION_MATCH, BLOB_INFO1.getMetageneration(),
@@ -545,6 +549,32 @@ public void testCreateBlobWithOptions() throws IOException {
     assertEquals(-1, byteStream.read(streamBytes));
   }
 
+  @Test
+  public void testCreateBlobWithDisabledGzipContent() throws IOException {
+    Capture<ByteArrayInputStream> capturedStream = Capture.newInstance();
+    EasyMock.expect(
+            storageRpcMock.create(
+                EasyMock.eq(
+                    BLOB_INFO1
+                        .toBuilder()
+                        .setMd5(CONTENT_MD5)
+                        .setCrc32c(CONTENT_CRC32C)
+                        .build()
+                        .toPb()),
+                EasyMock.capture(capturedStream),
+                EasyMock.eq(BLOB_TARGET_OPTIONS_CREATE_DISABLE_GZIP_CONTENT)))
+        .andReturn(BLOB_INFO1.toPb());
+    EasyMock.replay(storageRpcMock);
+    initializeService();
+    Blob blob = storage.create(BLOB_INFO1, BLOB_CONTENT, BLOB_TARGET_DISABLE_GZIP_CONTENT);
+    assertEquals(expectedBlob1, blob);
+    ByteArrayInputStream byteStream = capturedStream.getValue();
+    byte[] streamBytes = new byte[BLOB_CONTENT.length];
+    assertEquals(BLOB_CONTENT.length, byteStream.read(streamBytes));
+    assertArrayEquals(BLOB_CONTENT, streamBytes);
+    assertEquals(-1, byteStream.read(streamBytes));
+  }
+
   @Test
   public void testCreateBlobWithEncryptionKey() throws IOException {
     Capture<ByteArrayInputStream> capturedStream = Capture.newInstance();