[core-http] use response status to determine whether body is stream (Azure#13192)

Our current code has the incorrect assumption that if any body mapper
of an operation spec is of `Stream` type then response body must be a
stream and `httpRequest.streamResponseBody` is set according to this
assumption. However, Storage issue Azure#12997 shows that the default
response of Download Blob operation does not have a stream body with
the following operation spec:

```typescript
  responses: {
    200: {
      bodyMapper: {
        serializedName: "parsedResponse",
        type: {
          name: "Stream"
        }
      },
      headersMapper: Mappers.BlobDownloadHeaders
    },
    ...
    default: {
      bodyMapper: Mappers.StorageError,
      headersMapper: Mappers.BlobDownloadHeaders
    }
  },
```
Treating response body as stream in the default case prevent us from
parsing the body which leads to missing `code` and `message` from the
error object.

This change fixes the issue by actually checking the body type of the
response corresponding to the response status code.

WebResource{Like}.streamResponseBody is deprecated 
because a boolean value is not enough to determine whether a response
is streaming or not. A list of streaming response status code
`streamResponseStatusCodes` is introduced and used instead.

ServiceClient no longer sets the deprecated
`streamResponsebody` property based on incorrect assumption. However
we still respect `streamResponseBody` in case it's set by users.

Core v2 removes `streamResponseBody`.

Author: jeremymeng

sdk/core/core-client/src/deserializationPolicy.ts

@@ -19,7 +19,6 @@ import {
   RequiredSerializerOptions
 } from "./interfaces";
 import { MapperTypeNames } from "./serializer";
-import { isStreamOperation } from "./interfaceHelpers";
 import { getOperationRequestInfo } from "./operationHelpers";
 
 const defaultJsonContentTypes = ["application/json", "text/json"];
@@ -243,7 +242,9 @@ function handleErrorResponse(
 
   const errorResponseSpec = responseSpec ?? operationSpec.responses.default;
 
-  const initialErrorMessage = isStreamOperation(operationSpec)
+  const initialErrorMessage = parsedResponse.request.streamResponseStatusCodes?.has(
+    parsedResponse.status
+  )
     ? `Unexpected status code: ${parsedResponse.status}`
     : (parsedResponse.bodyAsText as string);
 
@@ -318,7 +319,10 @@ async function parse(
   opts: RequiredSerializerOptions,
   parseXML?: (str: string, opts?: XmlOptions) => Promise<any>
 ): Promise<FullOperationResponse> {
-  if (!operationResponse.request.streamResponseBody && operationResponse.bodyAsText) {
+  if (
+    !operationResponse.request.streamResponseStatusCodes?.has(operationResponse.status) &&
+    operationResponse.bodyAsText
+  ) {
     const text = operationResponse.bodyAsText;
     const contentType: string = operationResponse.headers.get("Content-Type") || "";
     const contentComponents: string[] = !contentType

sdk/core/core-client/src/interfaceHelpers.ts

@@ -5,19 +5,21 @@ import { MapperTypeNames } from "./serializer";
 import { OperationSpec, OperationParameter } from "./interfaces";
 
 /**
+ * Gets the list of status codes for streaming responses.
  * @internal @hidden
  */
-export function isStreamOperation(operationSpec: OperationSpec): boolean {
+export function getStreamingResponseStatusCodes(operationSpec: OperationSpec): Set<number> {
+  const result = new Set<number>();
   for (const statusCode in operationSpec.responses) {
     const operationResponse = operationSpec.responses[statusCode];
     if (
       operationResponse.bodyMapper &&
       operationResponse.bodyMapper.type.name === MapperTypeNames.Stream
     ) {
-      return true;
+      result.add(Number(statusCode));
     }
   }
-  return false;
+  return result;
 }
 
 /**

sdk/core/core-client/src/serviceClient.ts

@@ -21,7 +21,7 @@ import {
   CompositeMapper,
   XmlOptions
 } from "./interfaces";
-import { isStreamOperation } from "./interfaceHelpers";
+import { getStreamingResponseStatusCodes } from "./interfaceHelpers";
 import { getRequestUrl } from "./urlHelpers";
 import { isPrimitiveType } from "./utils";
 import { deserializationPolicy, DeserializationPolicyOptions } from "./deserializationPolicy";
@@ -195,8 +195,8 @@ export class ServiceClient {
       }
     }
 
-    if (request.streamResponseBody === undefined) {
-      request.streamResponseBody = isStreamOperation(operationSpec);
+    if (request.streamResponseStatusCodes === undefined) {
+      request.streamResponseStatusCodes = getStreamingResponseStatusCodes(operationSpec);
     }
 
     try {

sdk/core/core-client/test/serviceClient.spec.ts

@@ -821,6 +821,86 @@ describe("ServiceClient", function() {
     }
   });
 
+  it("should deserialize non-streaming default response", async function() {
+    const StorageError: CompositeMapper = {
+      serializedName: "StorageError",
+      type: {
+        name: "Composite",
+        className: "StorageError",
+        modelProperties: {
+          message: {
+            xmlName: "Message",
+            serializedName: "Message",
+            type: {
+              name: "String"
+            }
+          },
+          code: {
+            xmlName: "Code",
+            serializedName: "Code",
+            type: {
+              name: "String"
+            }
+          }
+        }
+      }
+    };
+
+    const serializer = createSerializer(undefined, true);
+
+    const operationSpec: OperationSpec = {
+      httpMethod: "GET",
+      responses: {
+        200: {
+          bodyMapper: {
+            serializedName: "parsedResponse",
+            type: {
+              name: "Stream"
+            }
+          }
+        },
+        default: {
+          bodyMapper: StorageError
+        }
+      },
+      baseUrl: "https://example.com",
+      serializer
+    };
+
+    let request: OperationRequest = createPipelineRequest({ url: "https://example.com" });
+    const operationInfo = getOperationRequestInfo(request);
+    operationInfo.operationSpec = operationSpec;
+
+    const httpsClient: HttpsClient = {
+      sendRequest: (req) => {
+        request = req;
+        return Promise.resolve({
+          request,
+          status: 500,
+          headers: createHttpHeaders({
+            "Content-Type": "application/json"
+          }),
+          bodyAsText: `{ "Code": "BlobNotFound", "Message": "The specified blob does not exist." }`
+        });
+      }
+    };
+
+    const pipeline = createEmptyPipeline();
+    pipeline.addPolicy(deserializationPolicy());
+    const client = new ServiceClient({
+      httpsClient,
+      pipeline
+    });
+
+    try {
+      await client.sendOperationRequest({}, operationSpec);
+      assert.fail();
+    } catch (ex) {
+      assert.strictEqual(ex.code, "BlobNotFound");
+      assert.strictEqual(ex.message, "The specified blob does not exist.");
+    }
+  });
+
   it("should re-use the common instance of DefaultHttpClient", function() {
     const client = new ServiceClient();
     assert.strictEqual((client as any)._httpsClient, getCachedDefaultHttpsClient());

sdk/core/core-http/CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## 1.2.3 (Unreleased)
 
+- Fix an issue where non-streaming response body was treated as stream [PR 13192](https://github.com/Azure/azure-sdk-for-js/pull/13192)
 
 ## 1.2.2 (2021-01-07)
 

sdk/core/core-http/review/core-http.api.md

@@ -901,7 +901,7 @@ export class WebResource implements WebResourceLike {
         [key: string]: any;
     }, headers?: {
         [key: string]: any;
-    } | HttpHeadersLike, streamResponseBody?: boolean, withCredentials?: boolean, abortSignal?: AbortSignalLike, timeout?: number, onUploadProgress?: (progress: TransferProgressEvent) => void, onDownloadProgress?: (progress: TransferProgressEvent) => void, proxySettings?: ProxySettings, keepAlive?: boolean, decompressResponse?: boolean);
+    } | HttpHeadersLike, streamResponseBody?: boolean, withCredentials?: boolean, abortSignal?: AbortSignalLike, timeout?: number, onUploadProgress?: (progress: TransferProgressEvent) => void, onDownloadProgress?: (progress: TransferProgressEvent) => void, proxySettings?: ProxySettings, keepAlive?: boolean, decompressResponse?: boolean, streamResponseStatusCodes?: Set<number>);
     // (undocumented)
     abortSignal?: AbortSignalLike;
     // (undocumented)
@@ -932,7 +932,9 @@ export class WebResource implements WebResourceLike {
     requestId: string;
     shouldDeserialize?: boolean | ((response: HttpOperationResponse) => boolean);
     spanOptions?: SpanOptions;
+    // @deprecated (undocumented)
     streamResponseBody?: boolean;
+    streamResponseStatusCodes?: Set<number>;
     // (undocumented)
     timeout: number;
     // (undocumented)
@@ -965,7 +967,9 @@ export interface WebResourceLike {
     requestId: string;
     shouldDeserialize?: boolean | ((response: HttpOperationResponse) => boolean);
     spanOptions?: SpanOptions;
+    // @deprecated (undocumented)
     streamResponseBody?: boolean;
+    streamResponseStatusCodes?: Set<number>;
     timeout: number;
     url: string;
     validateRequestProperties(): void;

sdk/core/core-http/src/fetchHttpClient.ts

@@ -155,14 +155,19 @@ export abstract class FetchHttpClient implements HttpClient {
       const response: CommonResponse = await this.fetch(httpRequest.url, requestInit);
 
       const headers = parseHeaders(response.headers);
+
+      const streaming =
+        httpRequest.streamResponseStatusCodes?.has(response.status) ||
+        httpRequest.streamResponseBody;
+
       operationResponse = {
         headers: headers,
         request: httpRequest,
         status: response.status,
-        readableStreamBody: httpRequest.streamResponseBody
+        readableStreamBody: streaming
           ? ((response.body as unknown) as NodeJS.ReadableStream)
           : undefined,
-        bodyAsText: !httpRequest.streamResponseBody ? await response.text() : undefined
+        bodyAsText: !streaming ? await response.text() : undefined
       };
 
       const onDownloadProgress = httpRequest.onDownloadProgress;

sdk/core/core-http/src/operationSpec.ts

@@ -95,16 +95,19 @@ export interface OperationSpec {
   readonly responses: { [responseCode: string]: OperationResponse };
 }
 
-export function isStreamOperation(operationSpec: OperationSpec): boolean {
-  let result = false;
+/**
+ * Gets the list of status codes for streaming responses.
+ * @internal @hidden
+ */
+export function getStreamResponseStatusCodes(operationSpec: OperationSpec): Set<number> {
+  const result = new Set<number>();
   for (const statusCode in operationSpec.responses) {
-    const operationResponse: OperationResponse = operationSpec.responses[statusCode];
+    const operationResponse = operationSpec.responses[statusCode];
     if (
       operationResponse.bodyMapper &&
       operationResponse.bodyMapper.type.name === MapperType.Stream
     ) {
-      result = true;
-      break;
+      result.add(Number(statusCode));
     }
   }
   return result;

sdk/core/core-http/src/policies/deserializationPolicy.ts

@@ -3,7 +3,7 @@
 
 import { HttpOperationResponse } from "../httpOperationResponse";
 import { OperationResponse } from "../operationResponse";
-import { OperationSpec, isStreamOperation } from "../operationSpec";
+import { OperationSpec } from "../operationSpec";
 import { RestError } from "../restError";
 import { MapperType } from "../serializer";
 import { parseXML } from "../util/xml";
@@ -256,7 +256,10 @@ function handleErrorResponse(
   }
 
   const errorResponseSpec = responseSpec ?? operationSpec.responses.default;
-  const initialErrorMessage = isStreamOperation(operationSpec)
+  const streaming =
+    parsedResponse.request.streamResponseStatusCodes?.has(parsedResponse.status) ||
+    parsedResponse.request.streamResponseBody;
+  const initialErrorMessage = streaming
     ? `Unexpected status code: ${parsedResponse.status}`
     : (parsedResponse.bodyAsText as string);
 
@@ -341,7 +344,10 @@ function parse(
     return Promise.reject(e);
   };
 
-  if (!operationResponse.request.streamResponseBody && operationResponse.bodyAsText) {
+  const streaming =
+    operationResponse.request.streamResponseStatusCodes?.has(operationResponse.status) ||
+    operationResponse.request.streamResponseBody;
+  if (!streaming && operationResponse.bodyAsText) {
     const text = operationResponse.bodyAsText;
     const contentType: string = operationResponse.headers.get("Content-Type") || "";
     const contentComponents: string[] = !contentType

sdk/core/core-http/src/serviceClient.ts

@@ -13,7 +13,7 @@ import {
   OperationParameter,
   ParameterPath
 } from "./operationParameter";
-import { isStreamOperation, OperationSpec } from "./operationSpec";
+import { getStreamResponseStatusCodes, OperationSpec } from "./operationSpec";
 import {
   deserializationPolicy,
   DeserializationContentTypes,
@@ -504,8 +504,8 @@ export class ServiceClient {
 
       serializeRequestBody(this, httpRequest, operationArguments, operationSpec);
 
-      if (httpRequest.streamResponseBody === undefined || httpRequest.streamResponseBody === null) {
-        httpRequest.streamResponseBody = isStreamOperation(operationSpec);
+      if (httpRequest.streamResponseStatusCodes === undefined) {
+        httpRequest.streamResponseStatusCodes = getStreamResponseStatusCodes(operationSpec);
       }
 
       let rawResponse: HttpOperationResponse;