Merge pull request Azure#238 from yojagad/AnotherDuplicateData

Account for removal of duplicate data from host

Author: yojagad

src/WorkerChannel.ts

@@ -5,6 +5,7 @@ import { IFunctionLoader } from './FunctionLoader';
 import { CreateContextAndInputs, LogCallback, ResultCallback } from './Context';
 import { IEventStream } from './GrpcService';
 import { toTypedData } from './converters';
+import { augmentTriggerMetadata } from './augmenters';
 import { systemError, systemLog } from './utils/Logger';
 import { InternalException } from './utils/InternalException';
 
@@ -82,8 +83,10 @@ export class WorkerChannel implements IWorkerChannel {
    * @param msg gRPC message content
    */
   public workerInitRequest(requestId: string, msg: rpc.WorkerInitRequest) {
-    let capabilitiesDictionary = {};
-    capabilitiesDictionary['RpcHttpBodyOnly'] = "true";
+    const capabilitiesDictionary = {
+      RpcHttpTriggerMetadataRemoved: "true",
+      RpcHttpBodyOnly: "true"
+    };
     this._eventStream.write({
       requestId: requestId,
       workerInitResponse: {
@@ -126,6 +129,9 @@ export class WorkerChannel implements IWorkerChannel {
    * @param msg gRPC message content
    */
   public invocationRequest(requestId: string, msg: rpc.InvocationRequest) {
+    // Repopulate triggerMetaData if http.
+    augmentTriggerMetadata(msg);
+
     let info = this._functionLoader.getInfo(<string>msg.functionId);
     let logCallback: LogCallback = (level, ...args) => {
       this.log({
@@ -267,4 +273,4 @@ export class WorkerChannel implements IWorkerChannel {
 
     return status;
   }
-}
+}

src/augmenters/InvocationRequestAugmenters.ts

@@ -0,0 +1,34 @@
+import { AzureFunctionsRpcMessages as rpc } from '../../azure-functions-language-worker-protobuf/src/rpc';
+
+/**
+ * Augments TriggerMetadata from invocation with $request.
+ * @param request IInvocationRequest object
+ */
+export function augmentTriggerMetadata(request: rpc.IInvocationRequest) {
+    var key: any, value: any;
+    if (request.inputData) {
+        request.inputData.forEach((element, _index, _array) => {
+        let elementKeys = Object.keys(element);
+        if (elementKeys) {
+          for (let val of elementKeys) {
+            if (element[val] && element[val].http) {
+                key = element["name"];
+                value = element[val];
+                break;
+              }
+            }
+          }
+      });
+    }
+    if(request && request.triggerMetadata)
+    {
+      if(key && (request.triggerMetadata[key] === undefined || request.triggerMetadata[key] === null))
+      {
+        request.triggerMetadata[key] = value;
+      }
+      if((request.triggerMetadata["$request"] === undefined || request.triggerMetadata["$request"] === null))
+      {
+        request.triggerMetadata["$request"] = value;
+      }
+    }
+}

src/augmenters/index.ts

@@ -0,0 +1 @@
+export * from './InvocationRequestAugmenters';

src/converters/BindingConverters.ts

@@ -24,7 +24,8 @@ export function getBindingDefinitions(info: FunctionInfo): BindingDefinition[] {
 export function getNormalizedBindingData(request: rpc.IInvocationRequest): Dict<any> {
   let bindingData: Dict<any> = {
     invocationId: request.invocationId
-  };
+  };  
+
   // node binding data is camel cased due to language convention
   if (request.triggerMetadata) {
     Object.assign(bindingData, convertKeysToCamelCase(request.triggerMetadata))

test/BindingConvertersTests.ts

@@ -20,6 +20,7 @@ describe('Binding Converters', () => {
             string: "Https://mock/"
         }
     };
+
     var request: rpc.IInvocationRequest = <rpc.IInvocationRequest> {
         triggerMetadata: triggerDataMock,
         invocationId: "12341"

test/WorkerChannelTests.ts

@@ -28,7 +28,8 @@ describe('WorkerChannel', () => {
       requestId: 'id',
       workerInitResponse: {
         capabilities: { 
-          'RpcHttpBodyOnly': "true" 
+              'RpcHttpBodyOnly': "true",
+              'RpcHttpTriggerMetadataRemoved': "true"
         },
         result: {
           status: rpc.StatusResult.Status.Success
@@ -161,13 +162,45 @@ describe('WorkerChannel', () => {
       name: 'test',
       outputBindings: {}
     })
+    
+    var triggerDataMock: { [k: string]: rpc.ITypedData } = {
+      "Headers": {
+          json: JSON.stringify({Connection: 'Keep-Alive'})
+      },
+      "Sys": {
+          json: JSON.stringify({MethodName: 'test-js', UtcNow: '2018', RandGuid: '3212'})
+        }
+    };
+
+    var inputDataValue = {
+      name: "req",
+      data: {
+          data: "http",
+          http: 
+          {
+              body:
+              {
+                  data: "string",
+                  body: "blahh"
+              },
+              rawBody:
+              {
+                  data: "string",
+                  body: "blahh"
+              }
+          }
+      } 
+    };
+
+    var actualInvocationRequest: rpc.IInvocationRequest = <rpc.IInvocationRequest> {
+      functionId: 'id',
+      invocationId: '1',
+      inputData: [inputDataValue],
+      triggerMetadata: triggerDataMock,
+    };
 
     stream.addTestMessage({
-      invocationRequest: {
-        functionId: 'id',
-        invocationId: '1',
-        inputData: []
-      }
+      invocationRequest: actualInvocationRequest
     });
 
     sinon.assert.calledWithMatch(stream.written, <rpc.IStreamingMessage> {
@@ -179,6 +212,10 @@ describe('WorkerChannel', () => {
         outputData: []
       }
     });
+
+    // triggerMedata will be augmented with inpuDataValue since "RpcHttpTriggerMetadataRemoved" capability is set to true and therefore not populated by the host.
+    expect(JSON.stringify(actualInvocationRequest.triggerMetadata!.$request)).to.equal(JSON.stringify(inputDataValue.data));
+    expect(JSON.stringify(actualInvocationRequest.triggerMetadata!.req)).to.equal(JSON.stringify(inputDataValue.data));
   });
 
   it ('throws for malformed messages', () => {