Http stream support

package.json

@@ -41,12 +41,14 @@
         "watch": "webpack --watch --mode development"
     },
     "dependencies": {
+        "cookie": "^0.6.0",
         "long": "^4.0.0",
         "undici": "^5.13.0"
     },
     "devDependencies": {
         "@types/chai": "^4.2.22",
         "@types/chai-as-promised": "^7.1.5",
+        "@types/cookie": "^0.6.0",
         "@types/fs-extra": "^9.0.13",
         "@types/long": "^4.0.2",
         "@types/minimist": "^1.2.2",

src/InvocationContext.ts

@@ -10,6 +10,7 @@ import {
     TraceContext,
     TriggerMetadata,
 } from '@azure/functions';
+import { fallbackLogHandler } from './utils/fallbackLogHandler';
 
 export class InvocationContext implements types.InvocationContext {
     invocationId: string;
@@ -92,26 +93,3 @@ class InvocationContextExtraOutputs implements types.InvocationContextExtraOutpu
         this.#outputs[name] = value;
     }
 }
-
-function fallbackLogHandler(level: types.LogLevel, ...args: unknown[]): void {
-    switch (level) {
-        case 'trace':
-            console.trace(...args);
-            break;
-        case 'debug':
-            console.debug(...args);
-            break;
-        case 'information':
-            console.info(...args);
-            break;
-        case 'warning':
-            console.warn(...args);
-            break;
-        case 'critical':
-        case 'error':
-            console.error(...args);
-            break;
-        default:
-            console.log(...args);
-    }
-}

src/InvocationModel.ts

@@ -21,7 +21,10 @@ import { toCamelCaseValue } from './converters/toCamelCase';
 import { toRpcHttp } from './converters/toRpcHttp';
 import { toRpcTypedData } from './converters/toRpcTypedData';
 import { AzFuncSystemError } from './errors';
+import { waitForProxyRequest } from './http/httpProxy';
+import { HttpRequest } from './http/HttpRequest';
 import { InvocationContext } from './InvocationContext';
+import { isHttpStreamEnabled } from './setup';
 import { isHttpTrigger, isTimerTrigger, isTrigger } from './utils/isTrigger';
 import { isDefined, nonNullProp, nonNullValue } from './utils/nonNull';
 
@@ -61,7 +64,6 @@ export class InvocationModel implements coreTypes.InvocationModel {
         if (req.inputData) {
             for (const binding of req.inputData) {
                 const bindingName = nonNullProp(binding, 'name');
-                let input: unknown = fromRpcTypedData(binding.data);
 
                 const rpcBinding = this.#bindings[bindingName];
                 if (!rpcBinding) {
@@ -72,6 +74,15 @@ export class InvocationModel implements coreTypes.InvocationModel {
                     );
                 }
                 const bindingType = rpcBinding.type;
+
+                let input: unknown;
+                if (isHttpTrigger(bindingType) && isHttpStreamEnabled()) {
+                    const proxyRequest = await waitForProxyRequest(this.#coreCtx.invocationId);
+                    input = new HttpRequest({ ...binding.data?.http, proxyRequest });
+                } else {
+                    input = fromRpcTypedData(binding.data);
+                }
+
                 if (isTimerTrigger(bindingType)) {
                     input = toCamelCaseValue(input);
                 }
@@ -107,10 +118,14 @@ export class InvocationModel implements coreTypes.InvocationModel {
         for (const [name, binding] of Object.entries(this.#bindings)) {
             if (binding.direction === 'out') {
                 if (name === returnBindingKey) {
-                    response.returnValue = await this.#convertOutput(binding, result);
+                    response.returnValue = await this.#convertOutput(context.invocationId, binding, result);
                     usedReturnValue = true;
                 } else {
-                    const outputValue = await this.#convertOutput(binding, context.extraOutputs.get(name));
+                    const outputValue = await this.#convertOutput(
+                        context.invocationId,
+                        binding,
+                        context.extraOutputs.get(name)
+                    );
                     if (isDefined(outputValue)) {
                         response.outputData.push({ name, data: outputValue });
                     }
@@ -129,9 +144,13 @@ export class InvocationModel implements coreTypes.InvocationModel {
         return response;
     }
 
-    async #convertOutput(binding: RpcBindingInfo, value: unknown): Promise<RpcTypedData | null | undefined> {
+    async #convertOutput(
+        invocationId: string,
+        binding: RpcBindingInfo,
+        value: unknown
+    ): Promise<RpcTypedData | null | undefined> {
         if (binding.type?.toLowerCase() === 'http') {
-            return toRpcHttp(value);
+            return toRpcHttp(invocationId, value);
         } else {
             return toRpcTypedData(value);
         }

src/ProgrammingModel.ts

@@ -0,0 +1,29 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License.
+
+import * as coreTypes from '@azure/functions-core';
+import { CoreInvocationContext, WorkerCapabilities } from '@azure/functions-core';
+import { version } from './constants';
+import { setupHttpProxy } from './http/httpProxy';
+import { InvocationModel } from './InvocationModel';
+import { isHttpStreamEnabled, lockSetup } from './setup';
+
+export class ProgrammingModel implements coreTypes.ProgrammingModel {
+    name = '@azure/functions';
+    version = version;
+
+    getInvocationModel(coreCtx: CoreInvocationContext): InvocationModel {
+        return new InvocationModel(coreCtx);
+    }
+
+    async getCapabilities(capabilities: WorkerCapabilities): Promise<WorkerCapabilities> {
+        lockSetup();
+
+        if (isHttpStreamEnabled()) {
+            const httpUri = await setupHttpProxy();
+            capabilities.HttpUri = httpUri;
+        }
+
+        return capabilities;
+    }
+}

src/addBindingName.ts

@@ -1,10 +1,6 @@
 // Copyright (c) .NET Foundation. All rights reserved.
 // Licensed under the MIT License.
 
-export { HttpRequest } from './http/HttpRequest';
-export { HttpResponse } from './http/HttpResponse';
-export { InvocationContext } from './InvocationContext';
-
 const bindingCounts: Record<string, number> = {};
 /**
  * If the host spawns multiple workers, it expects the metadata (including binding name) to be the same across workers.

src/app.ts

@@ -22,27 +22,20 @@ import {
     WarmupFunctionOptions,
 } from '@azure/functions';
 import * as coreTypes from '@azure/functions-core';
-import { CoreInvocationContext, FunctionCallback } from '@azure/functions-core';
-import { returnBindingKey, version } from './constants';
+import { FunctionCallback } from '@azure/functions-core';
+import { returnBindingKey } from './constants';
 import { toRpcDuration } from './converters/toRpcDuration';
-import { InvocationModel } from './InvocationModel';
 import * as output from './output';
+import { ProgrammingModel } from './ProgrammingModel';
 import * as trigger from './trigger';
 import { isTrigger } from './utils/isTrigger';
 import { tryGetCoreApiLazy } from './utils/tryGetCoreApiLazy';
 
 export * as hook from './hooks/registerHook';
+export { setup } from './setup';
 
-class ProgrammingModel implements coreTypes.ProgrammingModel {
-    name = '@azure/functions';
-    version = version;
-    getInvocationModel(coreCtx: CoreInvocationContext): InvocationModel {
-        return new InvocationModel(coreCtx);
-    }
-}
-
-let hasSetup = false;
-function setup() {
+let hasSetModel = false;
+function setProgrammingModel() {
     const coreApi = tryGetCoreApiLazy();
     if (!coreApi) {
         console.warn(
@@ -51,7 +44,7 @@ function setup() {
     } else {
         coreApi.setProgrammingModel(new ProgrammingModel());
     }
-    hasSetup = true;
+    hasSetModel = true;
 }
 
 function convertToHttpOptions(
@@ -141,8 +134,8 @@ export function warmup(name: string, options: WarmupFunctionOptions): void {
 }
 
 export function generic(name: string, options: GenericFunctionOptions): void {
-    if (!hasSetup) {
-        setup();
+    if (!hasSetModel) {
+        setProgrammingModel();
     }
 
     const bindings: Record<string, coreTypes.RpcBindingInfo> = {};

src/converters/toRpcHttp.ts

@@ -3,11 +3,13 @@
 
 import { RpcHttpData, RpcTypedData } from '@azure/functions-core';
 import { AzFuncSystemError } from '../errors';
+import { sendProxyResponse } from '../http/httpProxy';
 import { HttpResponse } from '../http/HttpResponse';
+import { isHttpStreamEnabled } from '../setup';
 import { toRpcHttpCookie } from './toRpcHttpCookie';
 import { toRpcTypedData } from './toRpcTypedData';
 
-export async function toRpcHttp(data: unknown): Promise<RpcTypedData | null | undefined> {
+export async function toRpcHttp(invocationId: string, data: unknown): Promise<RpcTypedData | null | undefined> {
     if (data === null || data === undefined) {
         return data;
     } else if (typeof data !== 'object') {
@@ -17,6 +19,12 @@ export async function toRpcHttp(data: unknown): Promise<RpcTypedData | null | un
     }
 
     const response = data instanceof HttpResponse ? data : new HttpResponse(data);
+    if (isHttpStreamEnabled()) {
+        // send http data over http proxy instead of rpc
+        await sendProxyResponse(invocationId, response);
+        return;
+    }
+
     const rpcResponse: RpcHttpData = {};
     rpcResponse.statusCode = response.status.toString();