SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1458 added support for MessageingGateway to Source

Browse Source 
Added support for wiring MessagingGateway as Source to provide the same function support as for the Supplier

Resolves #1458
This commit is contained in:
Oleg Zhurakousky
2018-08-29 05:31:11 -05:00
parent 93959dd9f0
commit b7a7470e92
5 changed files with 106 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
spring-cloud-stream/pom.xml
View File

@@ -72,6 +72,18 @@
<artifactId>spring-boot-autoconfigure-processor</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-http</artifactId>
<version>5.1.0.M2</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>

4
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionCatalogWrapper.java
View File

@@ -21,6 +21,7 @@ import org.springframework.util.Assert;
/**
* @author David Turanski
* @author Oleg Zhurakousky
*
* @since 2.1
**/
@@ -44,4 +45,7 @@ class FunctionCatalogWrapper {
return lookup(null, name);
}
<T> boolean contains(Class<T> functionType, String name) {
return catalog.lookup(functionType, name) != null;
}
}

7
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
View File

@@ -16,6 +16,8 @@
package org.springframework.cloud.stream.function;
import java.util.function.Supplier;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
@@ -52,7 +54,6 @@ public class FunctionConfiguration {
public IntegrationFlowFunctionSupport functionSupport(FunctionCatalogWrapper functionCatalog,
FunctionInspector functionInspector, CompositeMessageConverterFactory messageConverterFactory,
StreamFunctionProperties functionProperties) {
return new IntegrationFlowFunctionSupport(functionCatalog, functionInspector, messageConverterFactory,
functionProperties);
}
@@ -76,7 +77,9 @@ public class FunctionConfiguration {
return functionSupport.integrationFlowForFunction(sink.input(), null).get();
}
else if (source != null) {
return functionSupport.integrationFlowFromNamedSupplier().channel(this.source.output()).get();
return functionSupport.containsFunction(Supplier.class)
? functionSupport.integrationFlowFromNamedSupplier().channel(this.source.output()).get()
: null;
}
throw new UnsupportedOperationException("Bindings other then Source, Processor and Sink are not currently supported");
}

45
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java
View File

@@ -26,10 +26,12 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.function.context.FunctionCatalog;
import org.springframework.cloud.function.context.FunctionType;
import org.springframework.cloud.function.context.catalog.FunctionInspector;
import org.springframework.cloud.function.core.FluxSupplier;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.integration.dsl.IntegrationFlowBuilder;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.messaging.Message;
@@ -45,7 +47,7 @@ import org.springframework.util.StringUtils;
*
* @since 2.1
*/
public class IntegrationFlowFunctionSupport {
class IntegrationFlowFunctionSupport {
private final FunctionCatalogWrapper functionCatalog;
@@ -64,7 +66,7 @@ public class IntegrationFlowFunctionSupport {
* @param messageConverterFactory
* @param functionProperties
*/
public IntegrationFlowFunctionSupport(FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector,
IntegrationFlowFunctionSupport(FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector,
CompositeMessageConverterFactory messageConverterFactory, StreamFunctionProperties functionProperties) {
Assert.notNull(functionCatalog, "'functionCatalog' must not be null");
@@ -77,8 +79,21 @@ public class IntegrationFlowFunctionSupport {
this.functionProperties = functionProperties;
}
/**
* Determines if function specified via 'spring.cloud.stream.function.definition'
* property can be located in {@link FunctionCatalog}
*
* @param typeOfFunction must be Supplier, Function or Consumer
* @return
*/
public <T> boolean containsFunction(Class<T> typeOfFunction) {
return StringUtils.hasText(this.functionProperties.getDefinition())
&& this.functionCatalog.contains(typeOfFunction, this.functionProperties.getDefinition());
}
public FunctionType getCurrentFunctionType() {
return functionInspector.getRegistration(functionCatalog.lookup(this.functionProperties.getDefinition())).getType();
FunctionType functionType = functionInspector.getRegistration(functionCatalog.lookup(this.functionProperties.getDefinition())).getType();
return functionType;
}
/**
@@ -162,6 +177,23 @@ public class IntegrationFlowFunctionSupport {
return false;
}
public <I,O> boolean andThenFunction(FluxMessageChannel fluxChannel, MessageChannel outputChannel) {
if (StringUtils.hasText(this.functionProperties.getDefinition())) {
FunctionInvoker<I,O> functionInvoker =
new FunctionInvoker<>(this.functionProperties.getDefinition(), this.functionCatalog,
this.functionInspector, this.messageConverterFactory, this.errorChannel);
if (outputChannel != null) {
subscribeToInput(functionInvoker, fluxChannel, outputChannel::send);
}
else {
subscribeToInput(functionInvoker, fluxChannel, null);
}
return true;
}
return false;
}
private <O> Mono<Void> subscribeToOutput(Consumer<Message<O>> outputProcessor,
Publisher<Message<O>> outputPublisher) {
@@ -171,11 +203,12 @@ public class IntegrationFlowFunctionSupport {
return output.then();
}
private <I,O> void subscribeToInput(FunctionInvoker<I,O> functionInvoker, Publisher<Message<I>> publisher,
@SuppressWarnings("unchecked")
private <I,O> void subscribeToInput(FunctionInvoker<I,O> functionInvoker, Publisher<?> publisher,
Consumer<Message<O>> outputProcessor) {
Flux<Message<I>> inputPublisher = Flux.from(publisher);
subscribeToOutput(outputProcessor, functionInvoker.apply(inputPublisher)).subscribe();
Flux<?> inputPublisher = Flux.from(publisher);
subscribeToOutput(outputProcessor, functionInvoker.apply((Flux<Message<I>>) inputPublisher)).subscribe();
}
}

46
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java
View File

@@ -24,9 +24,11 @@ import java.util.function.Supplier;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.test.web.client.TestRestTemplate;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.binder.test.InputDestination;
import org.springframework.cloud.stream.binder.test.OutputDestination;
@@ -37,7 +39,12 @@ import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.http.HttpMethod;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.http.dsl.Http;
import org.springframework.integration.http.dsl.HttpRequestHandlerEndpointSpec;
import org.springframework.integration.http.inbound.HttpRequestHandlingEndpointSupport;
import org.springframework.messaging.Message;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
@@ -94,6 +101,19 @@ public class GreenfieldFunctionEnableBindingTests {
}
}
@Test
public void testHttpEndpoint() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(HttpInboundEndpoint.class)).web(
WebApplicationType.SERVLET).run("--spring.cloud.stream.function.definition=upperCase", "--spring.jmx.enabled=false")) {
TestRestTemplate restTemplate = new TestRestTemplate();
restTemplate.postForLocation("http://localhost:8080", "hello");
OutputDestination target = context.getBean(OutputDestination.class);
assertThat(target.receive(10000).getPayload()).isEqualTo("HELLO".getBytes(StandardCharsets.UTF_8));
}
}
@EnableAutoConfiguration
@EnableBinding(Source.class)
@@ -128,4 +148,30 @@ public class GreenfieldFunctionEnableBindingTests {
};
}
}
@EnableAutoConfiguration
@EnableBinding(Source.class)
public static class HttpInboundEndpoint {
@Autowired
private Source source;
@Bean
public Function<String, String> upperCase() {
return s -> s.toUpperCase();
}
@Bean
public HttpRequestHandlingEndpointSupport doFoo(IntegrationFlowFunctionSupport functionSupport) {
FluxMessageChannel fluxChannel = new FluxMessageChannel();
HttpRequestHandlerEndpointSpec httpRequestHandler = Http
.inboundChannelAdapter("/*")
.requestMapping(requestMapping -> requestMapping.methods(HttpMethod.POST)
.consumes("*/*"))
.requestChannel(fluxChannel);
functionSupport.andThenFunction(fluxChannel, source.output());
return httpRequestHandler.get();
}
}
}