Storage - Content Validation Decoder Implementation with pipeline policy

sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/implementation/util/BuilderHelper.java

@@ -39,6 +39,7 @@
 import com.azure.storage.common.policy.ResponseValidationPolicyBuilder;
 import com.azure.storage.common.policy.ScrubEtagPolicy;
 import com.azure.storage.common.policy.StorageBearerTokenChallengeAuthorizationPolicy;
+import com.azure.storage.common.policy.StorageContentValidationDecoderPolicy;
 import com.azure.storage.common.policy.StorageContentValidationPolicy;
 import com.azure.storage.common.policy.StorageSharedKeyCredentialPolicy;
 
@@ -97,6 +98,8 @@ public static HttpPipeline buildPipeline(StorageSharedKeyCredential storageShare
         // Closest to API goes first, closest to wire goes last.
         List<HttpPipelinePolicy> policies = new ArrayList<>();
 
+        policies.add(new StorageContentValidationDecoderPolicy());
+
         policies.add(getUserAgentPolicy(configuration, logOptions, clientOptions));
         policies.add(new RequestIdPolicy());
 

sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/implementation/util/DownloadValidationUtils.java

@@ -0,0 +1,51 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.storage.blob.implementation.util;
+
+import com.azure.core.util.Context;
+import com.azure.storage.common.ContentValidationAlgorithm;
+import com.azure.storage.common.implementation.Constants;
+import com.azure.storage.common.implementation.contentvalidation.ContentValidationModeResolver;
+
+/**
+ * Centralizes download content validation decisions based on {@link ContentValidationAlgorithm}.
+ * <p>
+ * Mirrors the pattern established by {@link ContentValidationModeResolver} for uploads.
+ * <p>
+ * RESERVED FOR INTERNAL USE.
+ */
+public final class DownloadValidationUtils {
+
+    private DownloadValidationUtils() {
+    }
+
+    /**
+     * Whether the algorithm requires structured message decoding (CRC64 / AUTO).
+     */
+    public static boolean isStructuredMessageAlgorithm(ContentValidationAlgorithm algorithm) {
+        return ContentValidationModeResolver.isCrc64OrAuto(algorithm);
+    }
+
+    /**
+     * Resolves the effective algorithm, defaulting null to NONE.
+     */
+    public static ContentValidationAlgorithm resolveAlgorithm(ContentValidationAlgorithm algorithm) {
+        return algorithm != null ? algorithm : ContentValidationAlgorithm.NONE;
+    }
+
+    /**
+     * Adds structured message decoding context key when CRC64/AUTO validation is active.
+     *
+     * @param context The base context to augment. Null is treated as {@link Context#NONE}.
+     * @param algorithm The resolved checksum algorithm.
+     * @return The augmented context.
+     */
+    public static Context applyStructuredMessageContext(Context context, ContentValidationAlgorithm algorithm) {
+        Context base = context == null ? Context.NONE : context;
+        if (!isStructuredMessageAlgorithm(algorithm)) {
+            return base;
+        }
+        return base.addData(Constants.STRUCTURED_MESSAGE_DECODING_CONTEXT_KEY, true);
+    }
+}

sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobAsyncClientBase.java

@@ -45,6 +45,7 @@
 import com.azure.storage.blob.implementation.util.BlobRequestConditionProperty;
 import com.azure.storage.blob.implementation.util.BlobSasImplUtil;
 import com.azure.storage.blob.implementation.util.ChunkedDownloadUtils;
+import com.azure.storage.blob.implementation.util.DownloadValidationUtils;
 import com.azure.storage.blob.implementation.util.ModelHelper;
 import com.azure.storage.blob.models.AccessTier;
 import com.azure.storage.blob.models.BlobBeginCopySourceRequestConditions;
@@ -1261,18 +1262,25 @@ Mono<BlobDownloadAsyncResponse> downloadStreamWithResponseInternal(BlobRange ran
         BlobRequestConditions requestConditions, boolean getRangeContentMd5,
         ContentValidationAlgorithm contentValidationAlgorithm, Context context) {
         BlobRange finalRange = range == null ? new BlobRange(0) : range;
-        Boolean getMD5 = getRangeContentMd5 ? getRangeContentMd5 : null;
+
+        final ContentValidationAlgorithm algorithm
+            = DownloadValidationUtils.resolveAlgorithm(contentValidationAlgorithm);
+        final boolean isStructuredMessageEnabled = DownloadValidationUtils.isStructuredMessageAlgorithm(algorithm);
+
+        final Boolean finalGetMD5 = (!isStructuredMessageEnabled && getRangeContentMd5) ? true : null;
+
         BlobRequestConditions finalRequestConditions
             = requestConditions == null ? new BlobRequestConditions() : requestConditions;
         DownloadRetryOptions finalOptions = (options == null) ? new DownloadRetryOptions() : options;
 
-        // The first range should eagerly convert headers as they'll be used to create response types.
-        Context firstRangeContext = context == null
+        final Context baseContext = context == null
             ? new Context("azure-eagerly-convert-headers", true)
             : context.addData("azure-eagerly-convert-headers", true);
 
-        return downloadRange(finalRange, finalRequestConditions, finalRequestConditions.getIfMatch(), getMD5,
-            firstRangeContext).map(response -> {
+        final Context downloadContext = DownloadValidationUtils.applyStructuredMessageContext(baseContext, algorithm);
+
+        return downloadRange(finalRange, finalRequestConditions, finalRequestConditions.getIfMatch(), finalGetMD5,
+            downloadContext).map(response -> {
                 BlobsDownloadHeaders blobsDownloadHeaders = new BlobsDownloadHeaders(response.getHeaders());
                 String eTag = blobsDownloadHeaders.getETag();
                 BlobDownloadHeaders blobDownloadHeaders = ModelHelper.populateBlobDownloadHeaders(blobsDownloadHeaders,
@@ -1298,31 +1306,33 @@ Mono<BlobDownloadAsyncResponse> downloadStreamWithResponseInternal(BlobRange ran
                         return Mono.error(throwable);
                     }
 
-                    long newCount = finalCount - offset;
-
-                    /*
-                     * It's possible that the network stream will throw an error after emitting all data but before
-                     * completing. Issuing a retry at this stage would leave the download in a bad state with
-                     * incorrect count and offset values. Because we have read the intended amount of data, we can
-                     * ignore the error at the end of the stream.
-                     */
-                    if (newCount == 0) {
-                        LOGGER.warning("Exception encountered in ReliableDownload after all data read from the network "
-                            + "but before stream signaled completion. Returning success as all data was downloaded. "
-                            + "Exception message: " + throwable.getMessage());
-                        return Mono.empty();
-                    }
-
                     try {
-                        return downloadRange(new BlobRange(initialOffset + offset, newCount), finalRequestConditions,
-                            eTag, getMD5, context);
+                        long newCount = finalCount - offset;
+
+                        /*
+                         * It's possible that the network stream will throw an error after emitting all data but
+                         * before completing. Issuing a retry at this stage would leave the download in a bad
+                         * state with incorrect count and offset values. Because we have read the intended amount
+                         * of data, we can ignore the error at the end of the stream.
+                         */
+                        if (newCount == 0) {
+                            LOGGER.warning(
+                                "Exception encountered in ReliableDownload after all data read from the network "
+                                    + "but before stream signaled completion. Returning success as all data was "
+                                    + "downloaded. Exception message: " + throwable.getMessage());
+                            return Mono.empty();
+                        }
+
+                        BlobRange retryRange = new BlobRange(initialOffset + offset, newCount);
+                        return downloadRange(retryRange, finalRequestConditions, eTag, finalGetMD5, downloadContext);
                     } catch (Exception e) {
                         return Mono.error(e);
                     }
                 };
 
                 return BlobDownloadAsyncResponseConstructorProxy.create(response, onDownloadErrorResume, finalOptions);
             });
+
     }
 
     private Mono<StreamResponse> downloadRange(BlobRange range, BlobRequestConditions requestConditions, String eTag,