Provides a hook for custom function invokers to extend/modify the function results handling. + * + * @param rawInputs the inputs passed in from the Azure runtime + * @param functionInputs the actual inputs used for the function invocation; may be + * {@link #enhanceInputIfNecessary different} from the {@literal rawInputs} + * @param functionResult the result from the function invocation + * @param function the invoked function + * @param executionContext the Azure execution context + * @return the possibly modified function results + */ + @SuppressWarnings("unchecked") + protected O postProcessImperativeFunctionResult(I rawInputs, Object functionInputs, Object functionResult, + FunctionInvocationWrapper function, ExecutionContext executionContext + ) { + return (O) this.convertOutputIfNecessary(rawInputs, functionResult); + } + + /** + * Post-processes the result from a reactive function invocation before returning it to the Azure + * runtime. The default behavior is to delegate to {@link #postProcessMonoFunctionResult} or + * {@link #postProcessFluxFunctionResult} based on the result type. + * + *
Provides a hook for custom function invokers to extend/modify the function results handling. + * + * @param rawInputs the inputs passed in from the Azure runtime + * @param functionInputs the actual inputs used for the function invocation; may be + * {@link #enhanceInputIfNecessary different} from the {@literal rawInputs} + * @param functionResult the result from the function invocation + * @param function the invoked function + * @param executionContext the Azure execution context + * @return the possibly modified function results + */ + protected O postProcessReactiveFunctionResult(I rawInputs, Object functionInputs, Publisher functionResult, + FunctionInvocationWrapper function, ExecutionContext executionContext + ) { + if (FunctionTypeUtils.isMono(function.getOutputType())) { + return postProcessMonoFunctionResult(rawInputs, functionInputs, Mono.from(functionResult), function, executionContext); + } + return postProcessFluxFunctionResult(rawInputs, functionInputs, Flux.from(functionResult), function, executionContext); + } + + /** + * Post-processes the {@code Mono} result from a reactive function invocation before returning it to the Azure + * runtime. The default behavior is to {@link Mono#blockOptional()} and {@link #convertOutputIfNecessary possibly convert} the result. + * + *
Provides a hook for custom function invokers to extend/modify the function results handling. + * + * @param rawInputs the inputs passed in from the Azure runtime + * @param functionInputs the actual inputs used for the function invocation; may be + * {@link #enhanceInputIfNecessary different} from the {@literal rawInputs} + * @param functionResult the Mono result from the function invocation + * @param function the invoked function + * @param executionContext the Azure execution context + * @return the possibly modified function results + */ + @SuppressWarnings("unchecked") + protected O postProcessMonoFunctionResult(I rawInputs, Object functionInputs, Mono functionResult, + FunctionInvocationWrapper function, ExecutionContext executionContext + ) { + return (O) this.convertOutputIfNecessary(rawInputs, functionResult.blockOptional().get()); + } + + /** + * Post-processes the {@code Flux} result from a reactive function invocation before returning it to the Azure + * runtime. The default behavior is to {@link Flux#toIterable() block} and {@link #convertOutputIfNecessary possibly convert} the results. + * + *
Provides a hook for custom function invokers to extend/modify the function results handling.
+ *
+ * @param rawInputs the inputs passed in from the Azure runtime
+ * @param functionInputs the actual inputs used for the function invocation; may be
+ * {@link #enhanceInputIfNecessary different} from the {@literal rawInputs}
+ * @param functionResult the Mono result from the function invocation
+ * @param function the invoked function
+ * @param executionContext the Azure execution context
+ * @return the possibly modified function results
+ */
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ protected O postProcessFluxFunctionResult(I rawInputs, Object functionInputs, Flux functionResult,
+ FunctionInvocationWrapper function, ExecutionContext executionContext
+ ) {
+ List resultList = new ArrayList<>();
+ for (Object resultItem : functionResult.toIterable()) {
+ if (resultItem instanceof Collection) {
+ resultList.addAll((Collection) resultItem);
}
else {
- List resultList = new ArrayList<>();
- for (Object resultItem : Flux.from((Publisher) output).toIterable()) {
- if (resultItem instanceof Collection) {
- resultList.addAll((Collection) resultItem);
- }
- else {
- if (!function.isSupplier() && Collection.class.isAssignableFrom(FunctionTypeUtils.getRawType(function.getInputType()))
- && !Collection.class.isAssignableFrom(FunctionTypeUtils.getRawType(function.getOutputType()))) {
- return (O) this.convertOutputIfNecessary(input, resultItem);
- }
- else {
- resultList.add(resultItem);
- }
- }
+ if (!function.isSupplier() && Collection.class.isAssignableFrom(FunctionTypeUtils.getRawType(function.getInputType()))
+ && !Collection.class.isAssignableFrom(FunctionTypeUtils.getRawType(function.getOutputType()))) {
+ return (O) this.convertOutputIfNecessary(rawInputs, resultItem);
+ }
+ else {
+ resultList.add(resultItem);
}
- return (O) this.convertOutputIfNecessary(input, resultList);
}
}
- return (O) this.convertOutputIfNecessary(input, output);
+ return (O) this.convertOutputIfNecessary(rawInputs, resultList);
}
@SuppressWarnings({ "unchecked", "rawtypes" })
diff --git a/spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/main/resources/META-INF/services/com.microsoft.azure.functions.spi.inject.FunctionInstanceInjector b/spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/main/resources/META-INF/services/com.microsoft.azure.functions.spi.inject.FunctionInstanceInjector
new file mode 100644
index 000000000..247910e32
--- /dev/null
+++ b/spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/main/resources/META-INF/services/com.microsoft.azure.functions.spi.inject.FunctionInstanceInjector
@@ -0,0 +1 @@
+org.springframework.cloud.function.adapter.azure.AzureFunctionInstanceInjector
\ No newline at end of file
diff --git a/spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/test/java/org/springframework/cloud/function/adapter/azure/CustomFunctionInvokerTests.java b/spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/test/java/org/springframework/cloud/function/adapter/azure/CustomFunctionInvokerTests.java
new file mode 100644
index 000000000..3d39543ec
--- /dev/null
+++ b/spring-cloud-function-adapters/spring-cloud-function-adapter-azure/src/test/java/org/springframework/cloud/function/adapter/azure/CustomFunctionInvokerTests.java
@@ -0,0 +1,200 @@
+/*
+ * Copyright 2022-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.function.adapter.azure;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import com.microsoft.azure.functions.ExecutionContext;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.support.GenericMessage;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.util.Lists.list;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.same;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+
+/**
+ * Unit tests for {@link FunctionInvoker} custom result handling.
+ *
+ * @author Chris Bono
+ */
+class CustomFunctionInvokerTests {
+
+ private FunctionInvoker currentInvoker;
+
+ @AfterEach
+ void closeCurrentInvoker() {
+ if (this.currentInvoker != null) {
+ this.currentInvoker.close();
+ }
+ }
+
+ /**
+ * Verifies custom result handling and proper post-process callback invocation for an imperative function.
+ */
+ @Test
+ void customImperativeResultHandling() {
+ FunctionInvoker
- *
- * @param name - name of the function
- * @param code - code of the function
- * @param resultTypeParameterizations - result types
- * @return a factory instance
- */
- public final CompiledFunctionFactory