Add embedded HTTP proxy server for streaming HTTP triggers (fixes #781)

eng/ci/public-build.yml

@@ -17,6 +17,32 @@ pr:
     include:
       - dev
 
+# Pipeline-level parameters surfaced in the ADO "Run pipeline" dialog so we can
+# build a custom branch of azure-functions-java-additions (e.g. a fork) before
+# the worker's mvn build. Defaults to a no-op so normal PRs continue to resolve
+# the library from Maven Central.
+#
+# TEMPORARY: defaults are flipped to build the fork branch
+# (ahmedmuhsin/azure-functions-java-additions @ feat/http-response-bodystream)
+# so GitHub-triggered PR CI for #877 picks up the unpublished
+# `azure-functions-java-core-library:1.4.0-SNAPSHOT`. Revert to the upstream
+# defaults (buildAdditionsFromSource=false, Azure repo, dev branch) once the
+# companion PR (Azure/azure-functions-java-additions#55) is merged and a
+# matching library version is published to Maven Central.
+parameters:
+  - name: buildAdditionsFromSource
+    displayName: 'Build azure-functions-java-additions from source (instead of resolving from Maven Central)'
+    type: boolean
+    default: true
+  - name: additionsRepoUrl
+    displayName: 'Git URL for azure-functions-java-additions (used only when buildAdditionsFromSource is true)'
+    type: string
+    default: 'https://github.com/ahmedmuhsin/azure-functions-java-additions.git'
+  - name: additionsBranch
+    displayName: 'Branch of azure-functions-java-additions to build (used only when buildAdditionsFromSource is true)'
+    type: string
+    default: 'feat/http-response-bodystream'
+
 resources:
   repositories:
     - repository: 1es
@@ -51,24 +77,37 @@ extends:
       - stage: Build
         jobs:
           - template: /eng/ci/templates/jobs/build.yml@self
+            parameters:
+              buildAdditionsFromSource: ${{ parameters.buildAdditionsFromSource }}
+              additionsRepoUrl: ${{ parameters.additionsRepoUrl }}
+              additionsBranch: ${{ parameters.additionsBranch }}
 
       - stage: TestWindows
         dependsOn: []
         jobs:
           - template: /eng/ci/templates/jobs/run-emulated-tests-windows.yml@self
             parameters:
               poolName: 1es-pool-azfunc-public
+              buildAdditionsFromSource: ${{ parameters.buildAdditionsFromSource }}
+              additionsRepoUrl: ${{ parameters.additionsRepoUrl }}
+              additionsBranch: ${{ parameters.additionsBranch }}
 
       - stage: TestLinux
         dependsOn: []
         jobs:
           - template: /eng/ci/templates/jobs/run-emulated-tests-linux.yml@self
             parameters:
               poolName: 1es-pool-azfunc-public
+              buildAdditionsFromSource: ${{ parameters.buildAdditionsFromSource }}
+              additionsRepoUrl: ${{ parameters.additionsRepoUrl }}
+              additionsBranch: ${{ parameters.additionsBranch }}
 
       - stage: TestDocker
         dependsOn: []
         jobs:
           - template: /eng/ci/templates/jobs/run-docker-tests-linux.yml@self
             parameters:
-              poolName: 1es-pool-azfunc-public
+              poolName: 1es-pool-azfunc-public
+              buildAdditionsFromSource: ${{ parameters.buildAdditionsFromSource }}
+              additionsRepoUrl: ${{ parameters.additionsRepoUrl }}
+              additionsBranch: ${{ parameters.additionsBranch }}

eng/ci/templates/jobs/build.yml

@@ -1,3 +1,14 @@
+parameters:
+  - name: buildAdditionsFromSource
+    type: boolean
+    default: false
+  - name: additionsRepoUrl
+    type: string
+    default: 'https://github.com/Azure/azure-functions-java-additions.git'
+  - name: additionsBranch
+    type: string
+    default: 'dev'
+
 jobs:
   - job: "Build"
     displayName: 'Build java worker'
@@ -14,6 +25,10 @@ jobs:
     - pwsh: |
         java -version
       displayName: 'Check default java version'
+    - ${{ if eq(parameters.buildAdditionsFromSource, true) }}:
+      - pwsh: |
+          ./installAdditionsLocally.ps1 -AdditionsRepoUrl '${{ parameters.additionsRepoUrl }}' -AdditionsBranch '${{ parameters.additionsBranch }}'
+        displayName: 'Install azure-functions-java-additions from source'
     - pwsh: |
         mvn clean package
       displayName: 'Build java worker'

eng/ci/templates/jobs/run-docker-tests-linux.yml

@@ -2,6 +2,15 @@ parameters:
   - name: poolName
     type: string
     default: ''
+  - name: buildAdditionsFromSource
+    type: boolean
+    default: false
+  - name: additionsRepoUrl
+    type: string
+    default: 'https://github.com/Azure/azure-functions-java-additions.git'
+  - name: additionsBranch
+    type: string
+    default: 'dev'
 
 jobs:
   - job: "TestDocker"
@@ -51,6 +60,11 @@ jobs:
           pip install -e dockertests/azure-functions-test-kit
         displayName: 'Install Python dependencies'
 
+      - ${{ if eq(parameters.buildAdditionsFromSource, true) }}:
+        - pwsh: |
+            ./installAdditionsLocally.ps1 -AdditionsRepoUrl '${{ parameters.additionsRepoUrl }}' -AdditionsBranch '${{ parameters.additionsBranch }}'
+          displayName: 'Install azure-functions-java-additions from source'
+
       - pwsh: |
           ./package-pipeline.ps1 -outputDir 'java-worker' -skipNuget
         displayName: 'Package Java worker'

eng/ci/templates/jobs/run-emulated-tests-linux.yml

@@ -2,6 +2,15 @@ parameters:
   - name: poolName
     type: string
     default: ''
+  - name: buildAdditionsFromSource
+    type: boolean
+    default: false
+  - name: additionsRepoUrl
+    type: string
+    default: 'https://github.com/Azure/azure-functions-java-additions.git'
+  - name: additionsBranch
+    type: string
+    default: 'dev'
 
 jobs:
   - job: "TestLinux"
@@ -83,6 +92,10 @@ jobs:
           docker compose -f emulatedtests/utils/docker-compose.yml pull
           docker compose -f emulatedtests/utils/docker-compose.yml up -d
         displayName: 'Install Azurite and Start Emulators'
+      - ${{ if eq(parameters.buildAdditionsFromSource, true) }}:
+        - pwsh: |
+            ./installAdditionsLocally.ps1 -AdditionsRepoUrl '${{ parameters.additionsRepoUrl }}' -AdditionsBranch '${{ parameters.additionsBranch }}'
+          displayName: 'Install azure-functions-java-additions from source'
       - pwsh: |
           if ("$(isTag)"){
             $buildNumber="$(Build.SourceBranchName)"

eng/ci/templates/jobs/run-emulated-tests-windows.yml

@@ -2,6 +2,15 @@ parameters:
   - name: poolName
     type: string
     default: ''
+  - name: buildAdditionsFromSource
+    type: boolean
+    default: false
+  - name: additionsRepoUrl
+    type: string
+    default: 'https://github.com/Azure/azure-functions-java-additions.git'
+  - name: additionsBranch
+    type: string
+    default: 'dev'
 
 jobs:
   - job: "TestWindows"
@@ -78,6 +87,10 @@ jobs:
           mkdir azurite
           azurite --silent --location azurite --debug azurite\debug.log &
         displayName: 'Install and Run Azurite'
+      - ${{ if eq(parameters.buildAdditionsFromSource, true) }}:
+        - pwsh: |
+            ./installAdditionsLocally.ps1 -AdditionsRepoUrl '${{ parameters.additionsRepoUrl }}' -AdditionsBranch '${{ parameters.additionsBranch }}'
+          displayName: 'Install azure-functions-java-additions from source'
       - pwsh: |
           if ("$(isTag)"){
             $buildNumber="$(Build.SourceBranchName)"

installAdditionsLocally.ps1

@@ -1,28 +1,50 @@
-# Variables for first repository
-$repoUrl1 = 'https://github.com/Azure/azure-functions-java-additions.git'
-$branchName1 = 'dev'
-$repoName1 = 'azure-functions-java-additions'
+[CmdletBinding()]
+param(
+    [string]$AdditionsRepoUrl = 'https://github.com/Azure/azure-functions-java-additions.git',
+    [string]$AdditionsBranch  = 'dev'
+)
 
-# Clone the first repository
-git clone $repoUrl1
+$ErrorActionPreference = 'Stop'
 
-# Change directory to the cloned repository
-Set-Location $repoName1
+$repoName       = 'azure-functions-java-additions'
+$workerRoot     = $PSScriptRoot
+$cloneDir       = Join-Path $workerRoot $repoName
+$mvnBuildScript = Join-Path $workerRoot 'mvnBuildAdditions.bat'
 
-# Checkout the desired branch
-git checkout $branchName1
+Write-Host "Installing $repoName from $AdditionsRepoUrl (branch: $AdditionsBranch)"
+Write-Host "Clone destination: $cloneDir"
 
-# Detect OS and execute build accordingly
-if ($IsWindows) {
-    # Run the batch script (mvnBuild.bat)
-    & "..\mvnBuildAdditions.bat"
-} else {
-    # Extract and explicitly invoke the mvn command from mvnBuild.bat
-    $mvnCommand = Get-Content "../mvnBuildAdditions.bat" | Where-Object { $_ -match '^mvn\s+' }
-    if ($null -ne $mvnCommand) {
-        # Execute the extracted mvn command explicitly as a single line
-        bash -c "$mvnCommand"
-    } else {
-        Write-Error "No mvn command found in mvnBuild.bat."
+# Make the clone idempotent for re-runs.
+if (Test-Path $cloneDir) {
+    Write-Host "Removing existing $cloneDir"
+    Remove-Item -Path $cloneDir -Recurse -Force
+}
+
+Push-Location $workerRoot
+try {
+    git clone --branch $AdditionsBranch --single-branch $AdditionsRepoUrl
+    if ($LASTEXITCODE -ne 0) { throw "git clone failed for $AdditionsRepoUrl ($AdditionsBranch)" }
+
+    Push-Location $repoName
+    try {
+        if ($IsWindows) {
+            # Run the batch script (mvnBuildAdditions.bat)
+            & $mvnBuildScript
+            if ($LASTEXITCODE -ne 0) { throw "mvnBuildAdditions.bat failed" }
+        } else {
+            # Extract and explicitly invoke the mvn command from mvnBuildAdditions.bat
+            $mvnCommand = Get-Content $mvnBuildScript | Where-Object { $_ -match '^mvn\s+' }
+            if ($null -ne $mvnCommand) {
+                # Execute the extracted mvn command explicitly as a single line
+                bash -c "$mvnCommand"
+                if ($LASTEXITCODE -ne 0) { throw "mvn command failed" }
+            } else {
+                throw "No mvn command found in $mvnBuildScript"
+            }
+        }
+    } finally {
+        Pop-Location
     }
+} finally {
+    Pop-Location
 }

pom.xml

@@ -15,7 +15,7 @@
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <maven.compiler.source>1.8</maven.compiler.source>
     <maven.compiler.target>1.8</maven.compiler.target>
-    <azure.functions.java.core.library.version>1.3.0</azure.functions.java.core.library.version>
+    <azure.functions.java.core.library.version>1.4.0-SNAPSHOT</azure.functions.java.core.library.version>
     <azure.functions.java.spi>1.1.0</azure.functions.java.spi>
     <azure.functions.java.sdktypes>1.0.2</azure.functions.java.sdktypes>
     <azure.functions.java.library.version>2.2.0</azure.functions.java.library.version>

src/main/java/com/microsoft/azure/functions/worker/Constants.java

@@ -18,4 +18,11 @@ private Constants(){}
     public static final String JAVA_ENABLE_OPENTELEMETRY = "JAVA_ENABLE_OPENTELEMETRY";
     public static final String JAVA_APPLICATIONINSIGHTS_ENABLE_TELEMETRY = "JAVA_APPLICATIONINSIGHTS_ENABLE_TELEMETRY";
     public static final String JAVA_ENABLE_SDK_TYPES = "JAVA_ENABLE_SDK_TYPES";
+    /**
+     * If set to "true" (case-insensitive), the worker will NOT start the
+     * embedded HTTP proxy server and will NOT advertise the {@code HttpUri}
+     * capability. Useful as an escape hatch if the proxy path causes problems.
+     * Default: unset (proxy enabled).
+     */
+    public static final String FUNCTIONS_JAVA_DISABLE_HTTP_PROXY = "FUNCTIONS_JAVA_DISABLE_HTTP_PROXY";
 }

src/main/java/com/microsoft/azure/functions/worker/JavaWorkerClient.java

@@ -12,6 +12,9 @@
 
 import com.microsoft.azure.functions.worker.broker.*;
 import com.microsoft.azure.functions.worker.handler.*;
+import com.microsoft.azure.functions.worker.http.HttpInvocationCoordinator;
+import com.microsoft.azure.functions.worker.http.HttpProxyServer;
+import com.microsoft.azure.functions.worker.http.ProxyConfig;
 import com.microsoft.azure.functions.worker.reflect.*;
 import com.microsoft.azure.functions.rpc.messages.*;
 
@@ -36,19 +39,28 @@ public JavaWorkerClient(IApplication app) {
         this.peer = new AtomicReference<>(null);
         this.handlerSuppliers = new HashMap<>();
         this.classPathProvider = new FactoryClassLoader().createClassLoaderProvider();
-
+        this.httpInvocationCoordinator = new HttpInvocationCoordinator();
+        this.httpProxyServer = httpProxyEnabled() ? new HttpProxyServer(ProxyConfig.defaults()) : null;
+
         this.addHandlers();
     }
 
+    private static boolean httpProxyEnabled() {
+        String value = System.getenv(Constants.FUNCTIONS_JAVA_DISABLE_HTTP_PROXY);
+        return !Boolean.parseBoolean(value);
+    }
+
     @PostConstruct
     private void addHandlers() {
         JavaFunctionBroker broker = new JavaFunctionBroker(classPathProvider);
-
-        this.handlerSuppliers.put(StreamingMessage.ContentCase.WORKER_INIT_REQUEST, () -> new WorkerInitRequestHandler(broker));
+
+        this.handlerSuppliers.put(StreamingMessage.ContentCase.WORKER_INIT_REQUEST,
+            () -> new WorkerInitRequestHandler(broker, this.httpProxyServer, this.httpInvocationCoordinator));
         this.handlerSuppliers.put(StreamingMessage.ContentCase.WORKER_WARMUP_REQUEST, WorkerWarmupHandler::new);
         this.handlerSuppliers.put(StreamingMessage.ContentCase.FUNCTION_ENVIRONMENT_RELOAD_REQUEST, () -> new FunctionEnvironmentReloadRequestHandler(broker));
         this.handlerSuppliers.put(StreamingMessage.ContentCase.FUNCTION_LOAD_REQUEST, () -> new FunctionLoadRequestHandler(broker));
-        this.handlerSuppliers.put(StreamingMessage.ContentCase.INVOCATION_REQUEST, () -> new InvocationRequestHandler(broker));
+        this.handlerSuppliers.put(StreamingMessage.ContentCase.INVOCATION_REQUEST,
+            () -> new InvocationRequestHandler(broker, this.httpInvocationCoordinator));
         this.handlerSuppliers.put(StreamingMessage.ContentCase.WORKER_STATUS_REQUEST, WorkerStatusRequestHandler::new);
         this.handlerSuppliers.put(StreamingMessage.ContentCase.WORKER_TERMINATE, WorkerTerminateRequestHandler::new);
     }
@@ -68,6 +80,15 @@ void logToHost(LogRecord record, String invocationId) {
 
     @Override
     public void close() throws Exception {
+        // Stop accepting HTTP proxy requests before tearing down the gRPC peer
+        // so in-flight HTTP handlers can drain on completion futures cleanly.
+        if (this.httpProxyServer != null) {
+            try {
+                this.httpProxyServer.close();
+            } catch (Exception ex) {
+                logger.log(Level.WARNING, "Failed to close HTTP proxy server cleanly", ex);
+            }
+        }
         this.peer.get().close();
         this.peer.set(null);
         this.channel.shutdownNow();
@@ -143,6 +164,8 @@ private synchronized void send(String requestId, MessageHandler<?, ?> marshaller
     private final AtomicReference<StreamingMessagePeer> peer;
     private final Map<StreamingMessage.ContentCase, Supplier<MessageHandler<?, ?>>> handlerSuppliers;
     private final ClassLoaderProvider classPathProvider;
+    private final HttpInvocationCoordinator httpInvocationCoordinator;
+    private final HttpProxyServer httpProxyServer;
 
     /**
      * @param functionsUri Host endpoint URI, or null for legacy startup args that only provide host and port.

src/main/java/com/microsoft/azure/functions/worker/binding/BindingDataStore.java

@@ -125,6 +125,31 @@ public Optional<TypedData> getDataTargetTypedValue(String name) throws Exception
 		});
     }
 
+    /**
+     * Returns the raw, unserialized response body of the HTTP output target, or
+     * {@code null} if no HTTP output target is registered or its body is null.
+     *
+     * <p>Used by the HTTP proxy path to recover streaming bodies
+     * ({@link java.io.InputStream}, {@code HttpResponseMessage.IOConsumer}) that
+     * cannot be represented in a protobuf {@code TypedData} and must instead be
+     * written directly to the {@code HttpExchange} response stream.</p>
+     */
+    public Object getHttpResponseRawBody() {
+        if (this.promotedTargets == null) {
+            return null;
+        }
+        Map<String, DataTarget> promoted = this.targets.get(this.promotedTargets);
+        if (promoted == null) {
+            return null;
+        }
+        for (DataTarget target : promoted.values()) {
+            if (target instanceof RpcHttpDataTarget) {
+                return ((RpcHttpDataTarget) target).getBody();
+            }
+        }
+        return null;
+    }
+
     public Optional<BindingData> getOrAddDataTarget(UUID outputId, String name, Type target, boolean ignoreDefinition) {
         DataTarget output = null;
         if (this.isDataTargetValid(name, target)) {