Body can no longer be a stream (Azure#198)

src/request/RequestHandler.ts

@@ -1,11 +1,10 @@
 import { Agent, OutgoingHttpHeaders } from "http";
 import { RequestOptions } from "https"; // TYPES ONLY
 import * as querystring from "querystring";
-import { Stream } from "stream";
 import { Constants, IHeaders } from "..";
 import { ConnectionPolicy } from "../documents";
 import { GlobalEndpointManager } from "../globalEndpointManager";
-import { Body, RetryUtility } from "../retry";
+import { RetryUtility } from "../retry";
 import { bodyFromData, createRequestObject, parse, Response } from "./request";
 import { RequestContext } from "./RequestContext";
 
@@ -19,7 +18,7 @@ export class RequestHandler {
   public static async createRequestObjectStub(
     connectionPolicy: ConnectionPolicy,
     requestOptions: RequestOptions,
-    body: Body
+    body: Buffer
   ) {
     return createRequestObject(connectionPolicy, requestOptions, body);
   }
@@ -32,7 +31,7 @@ export class RequestHandler {
    * @param {string} method - the http request method ( 'get', 'post', 'put', .. etc ).
    * @param {String} hostname - The base url for the endpoint.
    * @param {string} path - the path of the requesed resource.
-   * @param {Object} data - the request body. It can be either string, buffer, stream or undefined.
+   * @param {Object} data - the request body. It can be either string, buffer, or undefined.
    * @param {Object} queryParams - query parameters for the request.
    * @param {Object} headers - specific headers for the request.
    * @param {function} callback - the callback that will be called when the response is retrieved and processed.
@@ -44,7 +43,7 @@ export class RequestHandler {
     method: string,
     hostname: string,
     request: RequestContext,
-    data: string | Buffer | Stream,
+    data: string | Buffer,
     queryParams: any, // TODO: any query params types
     headers: IHeaders
   ): Promise<Response<any>> {
@@ -57,27 +56,23 @@ export class RequestHandler {
       if (!body) {
         return {
           result: {
-            message: "parameter data must be a javascript object, string, Buffer, or stream"
+            message: "parameter data must be a javascript object, string, or Buffer"
           },
           headers: undefined
         };
       }
     }
 
     let buffer;
-    let stream: Stream;
     if (body) {
       if (Buffer.isBuffer(body)) {
         buffer = body;
-      } else if ((body as Stream).pipe) {
-        // it is a stream
-        stream = body;
       } else if (typeof body === "string") {
         buffer = Buffer.from(body, "utf8");
       } else {
         return {
           result: {
-            message: "body must be string, Buffer, or stream"
+            message: "body must be string or Buffer"
           },
           headers: undefined
         };
@@ -103,16 +98,7 @@ export class RequestHandler {
       requestOptions.headers[Constants.HttpHeaders.ContentLength] = buffer.length;
       return RetryUtility.execute(
         globalEndpointManager,
-        { buffer, stream: null },
-        this.createRequestObjectStub,
-        connectionPolicy,
-        requestOptions,
-        request
-      );
-    } else if (stream) {
-      return RetryUtility.execute(
-        globalEndpointManager,
-        { buffer: null, stream },
+        buffer,
         this.createRequestObjectStub,
         connectionPolicy,
         requestOptions,
@@ -121,7 +107,7 @@ export class RequestHandler {
     } else {
       return RetryUtility.execute(
         globalEndpointManager,
-        { buffer: null, stream: null },
+        null,
         this.createRequestObjectStub,
         connectionPolicy,
         requestOptions,

src/request/request.ts

@@ -7,7 +7,6 @@ import * as url from "url";
 import { Constants, Helper } from "../common";
 import { ConnectionPolicy, MediaReadMode } from "../documents";
 import { IHeaders } from "../queryExecutionContext";
-import { Body } from "../retry";
 
 import { ErrorResponse } from "./ErrorResponse";
 export { ErrorResponse }; // Should refactor this out
@@ -56,7 +55,7 @@ export function parse(urlString: string) {
 export function createRequestObject(
   connectionPolicy: ConnectionPolicy,
   requestOptions: https.RequestOptions,
-  body: Body
+  body: Buffer
 ): Promise<Response<any>> {
   return new Promise<Response<any>>((resolve, reject) => {
     function onTimeout() {
@@ -117,10 +116,8 @@ export function createRequestObject(
 
     httpsRequest.once("error", reject);
 
-    if (body["stream"] !== null) {
-      body["stream"].pipe(httpsRequest);
-    } else if (body["buffer"] !== null) {
-      httpsRequest.write(body["buffer"]);
+    if (body) {
+      httpsRequest.write(body);
       httpsRequest.end();
     } else {
       httpsRequest.end();

src/retry/retryUtility.ts

@@ -1,5 +1,4 @@
 import { RequestOptions } from "https";
-import { Stream } from "stream";
 import * as url from "url";
 import { EndpointDiscoveryRetryPolicy, ResourceThrottleRetryPolicy, SessionRetryPolicy } from ".";
 import { Constants, Helper, StatusCodes, SubStatusCodes } from "../common";
@@ -12,34 +11,27 @@ import { DefaultRetryPolicy } from "./defaultRetryPolicy";
 import { IRetryPolicy } from "./IRetryPolicy";
 import { RetryContext } from "./RetryContext";
 
-/** @hidden */
-export interface Body {
-  buffer?: Buffer;
-  stream?: Stream;
-}
-
 /** @hidden */
 export type CreateRequestObjectStubFunction = (
   connectionPolicy: ConnectionPolicy,
   requestOptions: RequestOptions,
-  body: Body
+  body: Buffer
 ) => Promise<Response<any>>; // TODO: any response
 
 /** @hidden */
 export class RetryUtility {
   /**
    * Executes the retry policy for the created request object.
    * @param {object} globalEndpointManager - an instance of GlobalEndpointManager class.
-   * @param {object} body - a dictionary containing 'buffer' and 'stream' keys to hold corresponding buffer or\
-   *  stream body, null otherwise.
+   * @param {object} body - request body. A buffer or a string.
    * @param {function} createRequestObjectStub - stub function that creates the request object.
    * @param {object} connectionPolicy - an instance of ConnectionPolicy that has the connection configs.
    * @param {RequestOptions} requestOptions - The request options.
    * @param {function} callback - the callback that will be called when the request is finished executing.
    */
   public static async execute(
     globalEndpointManager: GlobalEndpointManager,
-    body: Body,
+    body: Buffer,
     createRequestObjectFunc: CreateRequestObjectStubFunction,
     connectionPolicy: ConnectionPolicy,
     requestOptions: RequestOptions,
@@ -74,8 +66,7 @@ export class RetryUtility {
 
   /**
    * Applies the retry policy for the created request object.
-   * @param {object} body - a dictionary containing 'buffer' and 'stream' keys to hold corresponding buffer or \
-   * stream body, null otherwise.
+   * @param {object} body - request body. A buffer or a string.
    * @param {function} createRequestObjectFunc - function that creates the request object.
    * @param {object} connectionPolicy - an instance of ConnectionPolicy that has the connection configs.
    * @param {RequestOptions} requestOptions - The request options.
@@ -85,7 +76,7 @@ export class RetryUtility {
    * @param {function} callback - the callback that will be called when the response is retrieved and processed.
    */
   public static async apply(
-    body: Body,
+    body: Buffer,
     createRequestObjectFunc: CreateRequestObjectStubFunction,
     connectionPolicy: ConnectionPolicy,
     requestOptions: RequestOptions,