bodge block blob upload.

kasobol-msft · Jan 26, 2022 · 3430b02 · 3430b02
1 parent 707049c
commit 3430b02
Show file tree

Hide file tree

Showing 3 changed files with 56 additions and 14 deletions.
diff --git a/...zure-storage-blob/src/main/java/com/azure/storage/blob/implementation/BlockBlobsImpl.java b/...zure-storage-blob/src/main/java/com/azure/storage/blob/implementation/BlockBlobsImpl.java
@@ -18,6 +18,8 @@
 import com.azure.core.annotation.ServiceMethod;
 import com.azure.core.annotation.UnexpectedResponseExceptionType;
 import com.azure.core.http.rest.RestProxy;
+import com.azure.core.syncasync.MultiResult;
+import com.azure.core.syncasync.SingleResult;
 import com.azure.core.util.Base64Util;
 import com.azure.core.util.Context;
 import com.azure.core.util.DateTimeRfc1123;
@@ -107,6 +109,45 @@ Mono<BlockBlobsUploadResponse> upload(
                 @HeaderParam("Accept") String accept,
                 Context context);
 
+        @Put("/{containerName}/{blob}")
+        @ExpectedResponses({201})
+        @UnexpectedResponseExceptionType(BlobStorageException.class)
+        SingleResult<BlockBlobsUploadResponse> upload(
+            @HostParam("url") String url,
+            @PathParam("containerName") String containerName,
+            @PathParam("blob") String blob,
+            @HeaderParam("x-ms-blob-type") String blobType,
+            @QueryParam("timeout") Integer timeout,
+            @HeaderParam("Content-MD5") String transactionalContentMD5,
+            @HeaderParam("Content-Length") long contentLength,
+            @HeaderParam("x-ms-blob-content-type") String contentType,
+            @HeaderParam("x-ms-blob-content-encoding") String contentEncoding,
+            @HeaderParam("x-ms-blob-content-language") String contentLanguage,
+            @HeaderParam("x-ms-blob-content-md5") String contentMd5,
+            @HeaderParam("x-ms-blob-cache-control") String cacheControl,
+            @HeaderParam("x-ms-meta-") Map<String, String> metadata,
+            @HeaderParam("x-ms-lease-id") String leaseId,
+            @HeaderParam("x-ms-blob-content-disposition") String contentDisposition,
+            @HeaderParam("x-ms-encryption-key") String encryptionKey,
+            @HeaderParam("x-ms-encryption-key-sha256") String encryptionKeySha256,
+            @HeaderParam("x-ms-encryption-algorithm") EncryptionAlgorithmType encryptionAlgorithm,
+            @HeaderParam("x-ms-encryption-scope") String encryptionScope,
+            @HeaderParam("x-ms-access-tier") AccessTier tier,
+            @HeaderParam("If-Modified-Since") DateTimeRfc1123 ifModifiedSince,
+            @HeaderParam("If-Unmodified-Since") DateTimeRfc1123 ifUnmodifiedSince,
+            @HeaderParam("If-Match") String ifMatch,
+            @HeaderParam("If-None-Match") String ifNoneMatch,
+            @HeaderParam("x-ms-if-tags") String ifTags,
+            @HeaderParam("x-ms-version") String version,
+            @HeaderParam("x-ms-client-request-id") String requestId,
+            @HeaderParam("x-ms-tags") String blobTagsString,
+            @HeaderParam("x-ms-immutability-policy-until-date") DateTimeRfc1123 immutabilityPolicyExpiry,
+            @HeaderParam("x-ms-immutability-policy-mode") BlobImmutabilityPolicyMode immutabilityPolicyMode,
+            @HeaderParam("x-ms-legal-hold") Boolean legalHold,
+            @BodyParam("application/octet-stream") MultiResult<ByteBuffer> body,
+            @HeaderParam("Accept") String accept,
+            Context context);
+
         @Put("/{containerName}/{blob}")
         @ExpectedResponses({201})
         @UnexpectedResponseExceptionType(BlobStorageException.class)
@@ -308,11 +349,11 @@ Mono<BlockBlobsGetBlockListResponse> getBlockList(
      * @return the completion.
      */
     @ServiceMethod(returns = ReturnType.SINGLE)
-    public Mono<BlockBlobsUploadResponse> uploadWithResponseAsync(
+    public SingleResult<BlockBlobsUploadResponse> uploadWithResponseAsync(
             String containerName,
             String blob,
             long contentLength,
-            Flux<ByteBuffer> body,
+            MultiResult<ByteBuffer> body,
             Integer timeout,
             byte[] transactionalContentMD5,
             Map<String, String> metadata,

diff --git a/...e-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlockBlobAsyncClient.java b/...e-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlockBlobAsyncClient.java
@@ -9,6 +9,8 @@
 import com.azure.core.http.HttpPipeline;
 import com.azure.core.http.rest.Response;
 import com.azure.core.http.rest.SimpleResponse;
+import com.azure.core.syncasync.SingleResult;
+import com.azure.core.syncasync.SyncAsyncMultiResult;
 import com.azure.core.util.Context;
 import com.azure.core.util.FluxUtil;
 import com.azure.core.util.logging.ClientLogger;
@@ -337,18 +339,22 @@ public Mono<Response<BlockBlobItem>> uploadWithResponse(Flux<ByteBuffer> data, l
     @ServiceMethod(returns = ReturnType.SINGLE)
     public Mono<Response<BlockBlobItem>> uploadWithResponse(BlockBlobSimpleUploadOptions options) {
         try {
-            return withContext(context -> uploadWithResponse(options, context));
+            return withContext(context -> uploadWithResponse(options, context).getAsync());
         } catch (RuntimeException ex) {
             return monoError(logger, ex);
         }
     }
 
-    Mono<Response<BlockBlobItem>> uploadWithResponse(BlockBlobSimpleUploadOptions options, Context context) {
+    SingleResult<Response<BlockBlobItem>> uploadWithResponse(BlockBlobSimpleUploadOptions options, Context context) {
         StorageImplUtils.assertNotNull("options", options);
-        Flux<ByteBuffer> data = options.getDataFlux() == null ? Utility.convertStreamToByteBuffer(
-            options.getDataStream(), options.getLength(), BlobAsyncClient.BLOB_DEFAULT_UPLOAD_BLOCK_SIZE, true)
-            .subscribeOn(Schedulers.boundedElastic())
-            : options.getDataFlux();
+        SyncAsyncMultiResult<ByteBuffer> data = new SyncAsyncMultiResult<ByteBuffer>(
+            () -> options.getDataFlux() == null ? FluxUtil.getSyncBody(options.getDataStream(), 1024 * 64)
+                : options.getDataFlux().toStream(),
+            ()-> options.getDataFlux() == null ? Utility.convertStreamToByteBuffer(
+                    options.getDataStream(), options.getLength(), BlobAsyncClient.BLOB_DEFAULT_UPLOAD_BLOCK_SIZE, true)
+                .subscribeOn(Schedulers.boundedElastic())
+                : options.getDataFlux()
+        );
         BlobRequestConditions requestConditions = options.getRequestConditions() == null ? new BlobRequestConditions()
             : options.getRequestConditions();
         context = context == null ? Context.NONE : context;

diff --git a/.../azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlockBlobClient.java b/.../azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlockBlobClient.java
@@ -416,12 +416,7 @@ public Response<BlockBlobItem> uploadWithResponse(InputStream data, long length,
     public Response<BlockBlobItem> uploadWithResponse(BlockBlobSimpleUploadOptions options, Duration timeout,
         Context context) {
         StorageImplUtils.assertNotNull("options", options);
-        Mono<Response<BlockBlobItem>> upload = client.uploadWithResponse(options, context);
-        try {
-            return blockWithOptionalTimeout(upload, timeout);
-        } catch (UncheckedIOException e) {
-            throw logger.logExceptionAsError(e);
-        }
+        return client.uploadWithResponse(options, context).getSync(timeout);
     }
 
     /**