retry for any request without a streaming payload #550

Author: ehennum

lib/documents.js

@@ -15,7 +15,6 @@
  */
 'use strict';
 
-
 var requester  = require('./requester.js');
 var mlutil     = require('./mlutil.js');
 var Operation  = require('./operation.js');
@@ -869,15 +868,17 @@ function writeStreamImpl(document, categories) {
   requestOptions.path = mlutil.databaseParam(connectionParams, endpoint, sep);
   mlutil.addTxidHeaders(requestOptions, txid);
 
-  // TODO: treat as chunked single document if no properties
-  var requestPartList = [];
-  addDocumentParts(requestPartList, document, true);
-
   var operation = new Operation(
       'write document stream', this.client, requestOptions, 'chunkedMultipart', 'single'
       );
-  operation.uri               = document.uri;
-  operation.requestDocument   = requestPartList;
+  operation.isReplayable = false;
+  operation.uri = document.uri;
+
+  // TODO: treat as chunked single document if no properties
+  var requestPartList = [];
+  addDocumentParts(operation, requestPartList, document, true);
+  operation.requestDocument = requestPartList;
+
   operation.multipartBoundary = mlutil.multipartBoundary;
   operation.errorTransform    = uriErrorTransform;
 
@@ -1098,8 +1099,8 @@ function writeMetadata(document, categories) {
 /** @ignore */
 function writeContent(contentOnly, document, requestParams, categories, requestType) {
   /*jshint validthis:true */
-  var content    = document.content;
-  var hasContent = (content != null);
+  var content     = document.content;
+  var hasContent  = (content != null);
 
   var endpoint = '/v1/documents';
 
@@ -1215,7 +1216,9 @@ function writeContent(contentOnly, document, requestParams, categories, requestT
     operation.categories = categories;
   }
   if (hasContent) {
-    operation.requestBody = mlutil.marshal(content);
+    operation.requestBody = mlutil.marshal(content, operation);
+  } else {
+    operation.isReplayable = false;
   }
   operation.outputTransform = singleWriteOutputTransform;
   operation.contentOnly     = (contentOnly === true);
@@ -1229,11 +1232,6 @@ function writeDocumentList(contentOnly, documents, requestParams, categories) {
     documents = documents.sort(compareDocuments);
   }
 
-  var requestPartList = [];
-  for (var i=0; i < documents.length; i++) {
-    addDocumentParts(requestPartList, documents[i], false);
-  }
-
   var endpoint = '/v1/documents';
 
   var txid = getTxid(requestParams);
@@ -1266,8 +1264,15 @@ function writeDocumentList(contentOnly, documents, requestParams, categories) {
   if (Array.isArray(categories) && categories.length > 0) {
     operation.categories = categories;
   }
+
   operation.multipartBoundary = multipartBoundary;
-  operation.requestPartList   = requestPartList;
+
+  var requestPartList = [];
+  for (var i=0; i < documents.length; i++) {
+    addDocumentParts(operation, requestPartList, documents[i], false);
+  }
+  operation.requestPartList = requestPartList;
+
   operation.errorTransform    = uriListErrorTransform;
   if (contentOnly === true) {
     operation.subdata = ['documents', 'uri'];
@@ -1425,7 +1430,7 @@ function addWriteConfig(document, hasUri, content, headers, sep) {
   return writeConfig;
 }
 /** @ignore */
-function addDocumentParts(partList, document, isContentOptional) {
+function addDocumentParts(operation, partList, document, isContentOptional) {
   var uri    = document.uri;
   var hasUri = (uri != null);
 
@@ -1464,7 +1469,7 @@ function addDocumentParts(partList, document, isContentOptional) {
     headers['Content-Disposition'] = disposition+'; category=content';
 
     if (hasContent) {
-      part.content = mlutil.marshal(content);
+      part.content = mlutil.marshal(content, operation);
     }
 
     partList.push(part);

lib/endpoint-proxy.js

@@ -56,6 +56,16 @@ function multiAtomicRequest(client, funcdef, args) {
 
   checkArgNames(funcdef, args, requestHeaders);
 
+  const connectionParams = client.connectionParams;
+  const requestOptions = mlutil.copyProperties(connectionParams);
+  requestOptions.method = 'POST';
+  requestOptions.headers = requestHeaders;
+  requestOptions.path = mlutil.databaseParam(connectionParams, endpoint, '?');
+
+  const operation = new Operation(
+      'call to ' + endpoint, client, requestOptions, 'single', funcdef.returnKind()
+  );
+
   const requestPartList = [];
   for (const [paramName, paramdef] of funcdef.paramdefs()) {
     const paramArgs = args[paramName];
@@ -65,25 +75,15 @@ function multiAtomicRequest(client, funcdef, args) {
 
     if (Array.isArray(paramArgs)) {
       for (const paramArg of paramArgs) {
-        requestPartList.push(paramName + '=' + encodeURIComponent(mlutil.marshal(paramArg)));
+        requestPartList.push(paramName + '=' + encodeURIComponent(mlutil.marshal(paramArg, operation)));
       }
     } else {
-      requestPartList.push(paramName + '=' + encodeURIComponent(mlutil.marshal(paramArgs)));
+      requestPartList.push(paramName + '=' + encodeURIComponent(mlutil.marshal(paramArgs, operation)));
     }
   }
+  operation.requestBody = requestPartList.join('&');
 
   const outputTransform = specifyOutputTransform(funcdef, requestHeaders);
-
-  const connectionParams = client.connectionParams;
-  const requestOptions = mlutil.copyProperties(connectionParams);
-  requestOptions.method = 'POST';
-  requestOptions.headers = requestHeaders;
-  requestOptions.path = mlutil.databaseParam(connectionParams, endpoint, '?');
-
-  const operation = new Operation(
-      'call to ' + endpoint, client, requestOptions, 'single', funcdef.returnKind()
-  );
-  operation.requestBody = mlutil.marshal(requestPartList.join('&'));
   addOutputTransform(funcdef, operation, outputTransform);
 
   return startRequest(funcdef, operation);
@@ -99,6 +99,17 @@ function multiNodeRequest(client, funcdef, args) {
 
   checkArgNames(funcdef, args, requestHeaders);
 
+  const connectionParams = client.connectionParams;
+  const requestOptions = mlutil.copyProperties(connectionParams);
+  requestOptions.method = 'POST';
+  requestOptions.headers = requestHeaders;
+  requestOptions.path = mlutil.databaseParam(connectionParams, endpoint, '?');
+
+  const operation = new Operation(
+      'call to ' + endpoint, client, requestOptions, 'multipart', funcdef.returnKind()
+  );
+  operation.multipartBoundary = multipartBoundary;
+
   const requestPartList = [];
   for (const [paramName, paramdef] of funcdef.paramdefs()) {
     const paramArgs = args[paramName];
@@ -115,30 +126,19 @@ function multiNodeRequest(client, funcdef, args) {
       for (const paramArg of paramArgs) {
         requestPartList.push({
           headers: headers,
-          content: mlutil.marshal(paramArg)
+          content: mlutil.marshal(paramArg, operation)
         });
       }
     } else {
       requestPartList.push({
         headers: headers,
-        content: mlutil.marshal(paramArgs)
+        content: mlutil.marshal(paramArgs, operation)
       });
     }
   }
+  operation.requestPartList = requestPartList;
 
   const outputTransform = specifyOutputTransform(funcdef, requestHeaders);
-
-  const connectionParams = client.connectionParams;
-  const requestOptions = mlutil.copyProperties(connectionParams);
-  requestOptions.method = 'POST';
-  requestOptions.headers = requestHeaders;
-  requestOptions.path = mlutil.databaseParam(connectionParams, endpoint, '?');
-
-  const operation = new Operation(
-      'call to ' + endpoint, client, requestOptions, 'multipart', funcdef.returnKind()
-  );
-  operation.multipartBoundary = multipartBoundary;
-  operation.requestPartList = requestPartList;
   addOutputTransform(funcdef, operation, outputTransform);
 
   return startRequest(funcdef, operation);

lib/graphs.js

@@ -391,7 +391,9 @@ function changeGraph(action, isStreaming, args) {
   if (graphType === 'named') {
     operation.uri = uri;
   }
-  if (!isStreaming) {
+  if (isStreaming) {
+    operation.isReplayable = false;
+  } else {
     operation.requestBody = data;
   }
   operation.outputTransform = emptyOutputTransform;

lib/mlutil.js

@@ -16,7 +16,6 @@
 'use strict';
 var util = require("util");
 
-
 var multipartBoundary = 'MLBOUND_' + Date.UTC(2014,12,31);
 
 // Normalize arguments by returning them as an array.
@@ -213,7 +212,7 @@ function identify(arg, withValues) {
   }
 }
 
-function marshal(data) {
+function marshal(data, state) {
   if (data == null) {
     return null;
   } else if (typeof data === 'string' || data instanceof String) {
@@ -222,6 +221,9 @@ function marshal(data) {
     return data;
   // readable stream might not inherit from ReadableStream
   } else if (typeof data._read === 'function') {
+    if (state !== void 0 && state.isReplayable === true) {
+      state.isReplayable = false;
+    }
     return data;
   } else if (typeof data === 'object' && data !== null) {
     if (Array.isArray(data)) {

lib/operation.js

@@ -26,13 +26,16 @@ function Operation(name, client, options, requestType, responseType) {
   this.logger             = client.getLogger();
   this.options            = options;
   this.requestType        = requestType;
+  this.isReplayable       = true;
   this.responseType       = responseType;
   this.validStatusCodes   = null;
   this.inlineAsDocument   = true;
   this.errorTransform     = null;
   this.error              = null;
   this.outputTransform    = null;
   this.subdata            = null;
+  this.authenticator      = null;
+  this.inputSender        = null;
   this.startedResponse    = false;
   this.done               = false;
   this.outputMode         = 'none';
@@ -48,6 +51,8 @@ function Operation(name, client, options, requestType, responseType) {
   this.nextMetadataBuffer = null;
   this.timestamp          = null;
   this.complexValues      = null;
+  this.retryAttempt       = 0;
+  this.retryDuration      = 0;
 }
 Operation.prototype.STREAM_MODES_CHUNKED_OBJECT_SEQUENCE =
   {chunked: true, object: true, sequence: true};