GH-264 Added initial support for Supplier and Consumer for Azure
Resolves #264
This commit is contained in:
@@ -24,6 +24,7 @@ import com.microsoft.azure.functions.ExecutionContext;
|
|||||||
import com.microsoft.azure.functions.OutputBinding;
|
import com.microsoft.azure.functions.OutputBinding;
|
||||||
import org.reactivestreams.Publisher;
|
import org.reactivestreams.Publisher;
|
||||||
import reactor.core.publisher.Flux;
|
import reactor.core.publisher.Flux;
|
||||||
|
import reactor.core.publisher.Mono;
|
||||||
|
|
||||||
import org.springframework.cloud.function.context.AbstractSpringFunctionAdapterInitializer;
|
import org.springframework.cloud.function.context.AbstractSpringFunctionAdapterInitializer;
|
||||||
|
|
||||||
@@ -52,7 +53,8 @@ public class AzureSpringBootRequestHandler<I, O> extends AbstractSpringFunctionA
|
|||||||
}
|
}
|
||||||
initialize(context);
|
initialize(context);
|
||||||
|
|
||||||
Publisher<?> events = extract(convertEvent(input));
|
Publisher<?> events = input == null ? Mono.empty() : extract(convertEvent(input));
|
||||||
|
|
||||||
Publisher<?> output = apply(events);
|
Publisher<?> output = apply(events);
|
||||||
return result(input, output);
|
return result(input, output);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -19,7 +19,9 @@ package org.springframework.cloud.function.adapter.azure;
|
|||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
import java.util.Arrays;
|
import java.util.Arrays;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
|
import java.util.function.Consumer;
|
||||||
import java.util.function.Function;
|
import java.util.function.Function;
|
||||||
|
import java.util.function.Supplier;
|
||||||
import java.util.stream.Collectors;
|
import java.util.stream.Collectors;
|
||||||
|
|
||||||
import com.microsoft.azure.functions.ExecutionContext;
|
import com.microsoft.azure.functions.ExecutionContext;
|
||||||
@@ -120,6 +122,26 @@ public class AzureSpringBootRequestHandlerTests {
|
|||||||
assertThat(bar).isNotNull();
|
assertThat(bar).isNotNull();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void supplierNonFluxBean() {
|
||||||
|
AzureSpringBootRequestHandler<Void, List<String>> handler = handler(NonFluxSupplierConfig.class);
|
||||||
|
List<String> result = handler.handleRequest(null, new TestExecutionContext("supplier"));
|
||||||
|
|
||||||
|
assertThat(result).isNotEmpty();
|
||||||
|
assertThat(result.toString()).isEqualTo("[foo1, foo2]");
|
||||||
|
}
|
||||||
|
|
||||||
|
private static String consumerResult;
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void consumerNonFluxBean() {
|
||||||
|
AzureSpringBootRequestHandler<String, Void> handler = handler(NonFluxConsumerConfig.class);
|
||||||
|
Object result = handler.handleRequest("foo1", new TestExecutionContext("consumer"));
|
||||||
|
|
||||||
|
assertThat(result).isNull();
|
||||||
|
assertThat(consumerResult).isEqualTo("foo1");
|
||||||
|
}
|
||||||
|
|
||||||
@After
|
@After
|
||||||
public void close() throws IOException {
|
public void close() throws IOException {
|
||||||
if (this.handler != null) {
|
if (this.handler != null) {
|
||||||
@@ -137,6 +159,26 @@ public class AzureSpringBootRequestHandlerTests {
|
|||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Configuration
|
||||||
|
protected static class NonFluxSupplierConfig {
|
||||||
|
|
||||||
|
@Bean
|
||||||
|
public Supplier<List<String>> supplier() {
|
||||||
|
return () -> Arrays.asList("foo1", "foo2");
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
@Configuration
|
||||||
|
protected static class NonFluxConsumerConfig {
|
||||||
|
|
||||||
|
@Bean
|
||||||
|
public Consumer<String> consumer() {
|
||||||
|
return (v) -> consumerResult = v;
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
@Configuration
|
@Configuration
|
||||||
protected static class BareConfig {
|
protected static class BareConfig {
|
||||||
|
|
||||||
|
|||||||
@@ -47,6 +47,7 @@ import org.springframework.context.ConfigurableApplicationContext;
|
|||||||
import org.springframework.context.support.GenericApplicationContext;
|
import org.springframework.context.support.GenericApplicationContext;
|
||||||
import org.springframework.util.Assert;
|
import org.springframework.util.Assert;
|
||||||
import org.springframework.util.ClassUtils;
|
import org.springframework.util.ClassUtils;
|
||||||
|
import org.springframework.util.CollectionUtils;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Base implementation for adapter initializers and request handlers.
|
* Base implementation for adapter initializers and request handlers.
|
||||||
@@ -212,7 +213,7 @@ public abstract class AbstractSpringFunctionAdapterInitializer<C> implements Clo
|
|||||||
}
|
}
|
||||||
@SuppressWarnings("unchecked")
|
@SuppressWarnings("unchecked")
|
||||||
O value = (O) result;
|
O value = (O) result;
|
||||||
return value;
|
return CollectionUtils.isEmpty(result) ? null : value;
|
||||||
}
|
}
|
||||||
|
|
||||||
private boolean isSingleInput(Function<?, ?> function, Object input) {
|
private boolean isSingleInput(Function<?, ?> function, Object input) {
|
||||||
@@ -332,7 +333,7 @@ public abstract class AbstractSpringFunctionAdapterInitializer<C> implements Clo
|
|||||||
|
|
||||||
name = resolveName(Consumer.class, targetContext);
|
name = resolveName(Consumer.class, targetContext);
|
||||||
if (context.containsBean(name) && context.getBean(name) instanceof Consumer) {
|
if (context.containsBean(name) && context.getBean(name) instanceof Consumer) {
|
||||||
this.consumer = getAndInstrumentFromContext(name);
|
this.function = getAndInstrumentFromContext(name); // FluxConsumer or any other consumer wrapper is a Function
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user