From 0be5f14766901be1b5f7ae517f74111209e999ba Mon Sep 17 00:00:00 2001 From: Dave Syer Date: Fri, 16 Mar 2018 11:10:54 -0400 Subject: [PATCH] Provide new config options for streams enabling composition User can switch off source or sink behaviour (the default is to bind to input and output streams), and then configure the name of a supplier (for a source) or consumer (for a sink). --- docs/src/main/asciidoc/getting-started.adoc | 10 +- .../main/asciidoc/spring-cloud-function.adoc | 55 +++++ .../config/StreamAutoConfiguration.java | 217 ++++++++++++++++-- .../config/StreamConfigurationProperties.java | 146 ++++++++++-- .../StreamListeningConsumerInvoker.java | 194 ++++++++++++++++ .../SupplierInvokingMessageProducer.java | 19 +- ...reamingExplicitProcessorEnabledTests.java} | 15 +- ...PojoStreamingExplicitSinkEnabledTests.java | 122 ++++++++++ ...joStreamingExplicitSourceEnabledTests.java | 120 ++++++++++ .../cloud/function/web/DefaultRouteTests.java | 78 +++++++ 10 files changed, 921 insertions(+), 55 deletions(-) create mode 100644 spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/config/StreamListeningConsumerInvoker.java rename spring-cloud-function-stream/src/test/java/org/springframework/cloud/function/stream/mixed/{PojoStreamingExplicitEndpointTests.java => PojoStreamingExplicitProcessorEnabledTests.java} (86%) create mode 100644 spring-cloud-function-stream/src/test/java/org/springframework/cloud/function/stream/mixed/PojoStreamingExplicitSinkEnabledTests.java create mode 100644 spring-cloud-function-stream/src/test/java/org/springframework/cloud/function/stream/mixed/PojoStreamingExplicitSourceEnabledTests.java create mode 100644 spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/DefaultRouteTests.java diff --git a/docs/src/main/asciidoc/getting-started.adoc b/docs/src/main/asciidoc/getting-started.adoc index 76670b13b..bd4b5d22e 100644 --- a/docs/src/main/asciidoc/getting-started.adoc +++ b/docs/src/main/asciidoc/getting-started.adoc @@ -58,13 +58,13 @@ one-per-jar. It's up to the developer to choose. An app with multiple functions can be deployed multiple times in different "personalities", exposing different functions over different physical transports. -== Deploying a Packaged Function - -TBD: describe the deployer app. - == Dynamic Compilation -To run these examples, change into the `scripts` directory: +There is a sample app that uses the function compiler to create a +function from a configuration property. The vanilla "function-sample" +also has that feature. And there are some examples that you can run to +see the compilation happening at run time. To run these examples, +change into the `scripts` directory: ---- cd scripts diff --git a/docs/src/main/asciidoc/spring-cloud-function.adoc b/docs/src/main/asciidoc/spring-cloud-function.adoc index 5c62a6824..d20a5ea50 100644 --- a/docs/src/main/asciidoc/spring-cloud-function.adoc +++ b/docs/src/main/asciidoc/spring-cloud-function.adoc @@ -18,3 +18,58 @@ include::https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/maste == Getting Started include::getting-started.adoc[] + +== Standalone Web Applications + +The `spring-cloud-function-web` module has autoconfiguration that +activates when it is included in a Spring Boot web application (with +MVC support). There is also a `spring-cloud-starter-function-web` to +collect all the optional dependnecies in case you just want a simple +getting started experience. + +With the web configurations activated your app will have an MVC +endpoint (on "/" by default, but configurable with +`spring.cloud.function.web.path`) that can be used to access the +functions in the application context. The supported content types are +plain text and JSON. + +|=== +| Method | Path | Request | Response | Status + +| GET | /{supplier} | - | Items from the named supplier | 200 OK +| POST | /{consumer} | JSON object or text | Mirrors input and pushes request body into consumer | 202 Accepted +| POST | /{consumer} | JSON array or text with new lines | Mirrors input and pushes body into consumer one by one | 202 Accepted +| POST | /{function} | JSON object or text | The result of applying the named function | 200 OK +| POST | /{function} | JSON array or text with new lines | The result of applying the named function | 200 OK +| GET | /{function}/{item} | - | Convert the item into an object and return the result of applying the function | 200 OK + +|=== + +As the table above shows the behaviour of the endpoint depends on the method and also the type of incoming request data. When the incoming data is single valued, and the target function is declared as obviously single valued (i.e. not returning a collection or `Flux`), then the response will also contain a single value. For multi-valued responses the client can ask for a server-sent event stream by sending `Accept: text/event-stream". If there is only one function (consumer etc.) then the name in the path is optional. Composite functions can be addressed using pipes or commas to separate function names (pipes are legal in URL paths, but a bit awkward to type on the command line). + +Functions and consumers that are declared with input and output in `Message` will see the request headers on the input messages, and the output message headers will be converted to HTTP headers. + +== Standalone Streaming Applications + +To send or receive messages from a broker (such as RabbitMQ or Kafka) you can use the `spring-cloud-function-stream` adapter. Add the adapter to your classpath along with the appropriate binder from Spring Cloud Stream. The adapter will bind to the message broker as a `Processor` (input and output streams) unless the user explicitly disables one or the other using `spring.cloud.function.stream.{source,sink}.enabled=false`. + +An incoming message is routed to a function (or consumer). If there is only one, then the choice is obvious. If there are multiple functions that can accept an incoming message, the message is inspected to see if there is a `stream_routekey` header containing the name of a function. The header is also added to outgoing messages from a supplier. Messages with no route key can be routed exclusively to a function or consumer by specifying `spring.cloud.function.stream.{processor,sink}.name`. A single supplier can be chosen for output messages (if more than one is available) using the `spring.cloud.function.stream.supplier.name`. Routing headers or function names can be composed using a comma or pipe separated name. + +NOTE: some binders will fail on startup if the message broker is not available and the function catalog contains suppliers that immediately produce messages when accessed. You can switch off the automatic publishing from suppliers on startup using the `spring.cloud.function.strean.supplier.enabled=false` flag. + +== Serverless Platform Adapters + +As well as being able to run as a standalone process, a Spring Cloud +Function application can be adapted to run one of the existing +servlerless platforms. In the project there are adapters for +https://github.com/spring-cloud/spring-cloud-function/tree/master/spring-cloud-function-adapters/spring-cloud-function-adapter-aws[AWS +Lambda], +https://github.com/spring-cloud/spring-cloud-function/tree/master/spring-cloud-function-adapters/spring-cloud-function-adapter-azure[Azure], +and +https://github.com/spring-cloud/spring-cloud-function/tree/master/spring-cloud-function-adapters/spring-cloud-function-adapter-openwhisk[Apache +OpenWhisk]. The Oracle Fn platform has its own Spring Cloud Function adapter. + +== Deploying a Packaged Function + +TBD: describe the deployer app. + diff --git a/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/config/StreamAutoConfiguration.java b/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/config/StreamAutoConfiguration.java index 64c833bc8..044c89835 100644 --- a/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/config/StreamAutoConfiguration.java +++ b/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/config/StreamAutoConfiguration.java @@ -17,9 +17,11 @@ package org.springframework.cloud.function.stream.config; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.condition.ConditionOutcome; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.autoconfigure.condition.SpringBootCondition; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.function.context.FunctionCatalog; import org.springframework.cloud.function.context.catalog.FunctionInspector; @@ -27,40 +29,223 @@ import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.cloud.stream.messaging.Processor; +import org.springframework.cloud.stream.messaging.Sink; +import org.springframework.cloud.stream.messaging.Source; import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.ConditionContext; +import org.springframework.context.annotation.Conditional; +import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Lazy; +import org.springframework.core.type.AnnotatedTypeMetadata; /** * @author Mark Fisher * @author Marius Bogoevici */ +@Configuration @EnableConfigurationProperties(StreamConfigurationProperties.class) @ConditionalOnClass(Binder.class) @ConditionalOnBean(FunctionCatalog.class) @ConditionalOnProperty(name = "spring.cloud.stream.enabled", havingValue = "true", matchIfMissing = true) -@EnableBinding(Processor.class) public class StreamAutoConfiguration { - @Autowired - private StreamConfigurationProperties properties; - - @Bean + @Configuration // Because of the underlying behaviour of Spring AMQP etc., sources do not start // up and fail gracefully if the broker is down. So we need a flag to be able to // switch this off and stop the app failing on startup. - // TODO: find a slicker way to do it (e.g. backoff if the broker is down) - @ConditionalOnProperty(name = "spring.cloud.function.stream.supplier.enabled", havingValue = "true", matchIfMissing = true) - public SupplierInvokingMessageProducer supplierInvoker( - FunctionCatalog registry) { - return new SupplierInvokingMessageProducer(registry); + @ConditionalOnProperty(name = "spring.cloud.function.stream.source.enabled", havingValue = "true", matchIfMissing = true) + protected static class SourceConfiguration { + + @Autowired + private StreamConfigurationProperties properties; + + @Bean + public SupplierInvokingMessageProducer supplierInvoker( + FunctionCatalog registry) { + return new SupplierInvokingMessageProducer(registry, properties.getSource().getName()); + } + } - @Bean - public StreamListeningFunctionInvoker functionInvoker(FunctionCatalog registry, - FunctionInspector functionInspector, - @Lazy CompositeMessageConverterFactory compositeMessageConverterFactory) { - return new StreamListeningFunctionInvoker(registry, functionInspector, - compositeMessageConverterFactory, properties.getDefaultRoute()); + @Configuration + @ConditionalOnProperty(name = "spring.cloud.function.stream.processor.enabled", havingValue = "true", matchIfMissing = true) + @Conditional(SourceAndSinkCondition.class) + protected static class ProcessorConfiguration { + + @Autowired + private StreamConfigurationProperties properties; + + @Bean + public StreamListeningFunctionInvoker functionInvoker(FunctionCatalog registry, + FunctionInspector functionInspector, + @Lazy CompositeMessageConverterFactory compositeMessageConverterFactory) { + return new StreamListeningFunctionInvoker(registry, functionInspector, + compositeMessageConverterFactory, properties.getDefaultRoute()); + } + + } + + @Configuration + @Conditional(SinkOnlyCondition.class) + protected static class SinkConfiguration { + + @Autowired + private StreamConfigurationProperties properties; + + public SinkConfiguration() { + } + + @Bean + public StreamListeningConsumerInvoker consumerInvoker(FunctionCatalog registry, + FunctionInspector functionInspector, + @Lazy CompositeMessageConverterFactory compositeMessageConverterFactory) { + return new StreamListeningConsumerInvoker(registry, functionInspector, + compositeMessageConverterFactory, properties.getSink().getName()); + } + + } + + @Configuration + @EnableBinding(Processor.class) + @Conditional(ProcessorCondition.class) + protected class ProcessorBindingConfiguration { + } + + @Configuration + @EnableBinding(Source.class) + @Conditional(SourceCondition.class) + protected class SourceBindingConfiguration { + } + + @Configuration + @EnableBinding(Sink.class) + @Conditional(SinkCondition.class) + protected class SinkBindingConfiguration { + } + + private static class SinkOnlyCondition extends SpringBootCondition { + private SourceAndSinkCondition processor = new SourceAndSinkCondition(); + + private SinkCondition sink = new SinkCondition(); + + @Override + public ConditionOutcome getMatchOutcome(ConditionContext context, + AnnotatedTypeMetadata metadata) { + if (processor.matches(context, metadata)) { + return ConditionOutcome.noMatch("Source is provided by Processor"); + } + if (sink.matches(context, metadata)) { + return ConditionOutcome.match("Sink is explicitly enabled"); + } + return ConditionOutcome.noMatch("Sink is not enabled and not available through Processor"); + } + } + + private static class SourceAndSinkCondition extends SpringBootCondition { + private SourceCondition source = new SourceCondition(); + + private SinkCondition sink = new SinkCondition(); + + private ProcessorCondition processor = new ProcessorCondition(); + + @Override + public ConditionOutcome getMatchOutcome(ConditionContext context, + AnnotatedTypeMetadata metadata) { + if (processor.matches(context, metadata)) { + return ConditionOutcome.match("Processor is bound"); + } + if (sink.matches(context, metadata) && source.matches(context, metadata)) { + return ConditionOutcome.match("Both Source and Sink are bound"); + } + return ConditionOutcome.noMatch("Both Source and Sink are not bound"); + } + } + + private static class ProcessorCondition extends SpringBootCondition { + + private SourceCondition source = new SourceCondition(); + + private SinkCondition sink = new SinkCondition(); + + @Override + public ConditionOutcome getMatchOutcome(ConditionContext context, + AnnotatedTypeMetadata metadata) { + return (!source.matches(context, metadata) + && !sink.matches(context, metadata)) + ? ConditionOutcome.match( + "Neither source nor sink is explicitly disabled") + : ConditionOutcome.noMatch( + "Either sink or source was explicitly disabled"); + } + + } + + private static class SourceCondition extends SpringBootCondition { + + @Override + public ConditionOutcome getMatchOutcome(ConditionContext context, + AnnotatedTypeMetadata metadata) { + Boolean enabled = context.getEnvironment().getProperty( + "spring.cloud.function.stream.source.enabled", Boolean.class); + Boolean sink = context.getEnvironment().getProperty( + "spring.cloud.function.stream.sink.enabled", Boolean.class, true); + if (enabled != null && enabled) { + if (!sink) { + return ConditionOutcome + .match("Source explicitly enabled and sink disabled"); + } + else { + return ConditionOutcome + .noMatch("Source explicitly enabled and sink enabled"); + } + } + if (enabled == null) { + if (!sink) { + return ConditionOutcome + .match("Source implicitly enabled and sink disabled"); + } + else { + return ConditionOutcome + .noMatch("Source not explicitly enabled and sink enabled"); + } + } + return ConditionOutcome.noMatch("Source explicitly disabled"); + } + + } + + private static class SinkCondition extends SpringBootCondition { + + @Override + public ConditionOutcome getMatchOutcome(ConditionContext context, + AnnotatedTypeMetadata metadata) { + Boolean enabled = context.getEnvironment().getProperty( + "spring.cloud.function.stream.sink.enabled", Boolean.class); + Boolean source = context.getEnvironment().getProperty( + "spring.cloud.function.stream.source.enabled", Boolean.class, true); + if (enabled != null && enabled) { + if (!source) { + return ConditionOutcome + .match("Sink explicitly enabled and source disabled"); + } + else { + return ConditionOutcome + .noMatch("Sink explicitly enabled and source enabled"); + } + } + if (enabled == null) { + if (!source) { + return ConditionOutcome + .match("Sink implicitly enabled and source disabled"); + } + else { + return ConditionOutcome + .noMatch("Sink not explicitly enabled and source enabled"); + } + } + return ConditionOutcome.noMatch("Sink explicitly disabled"); + } + } } diff --git a/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/config/StreamConfigurationProperties.java b/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/config/StreamConfigurationProperties.java index a913bbd37..7a92c4536 100644 --- a/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/config/StreamConfigurationProperties.java +++ b/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/config/StreamConfigurationProperties.java @@ -24,33 +24,135 @@ import org.springframework.boot.context.properties.ConfigurationProperties; @ConfigurationProperties(prefix = "spring.cloud.function.stream") public class StreamConfigurationProperties { - /** - * The default route for a message if more than one is available and no explicit route - * key is provided. - */ - private String defaultRoute; + private Source source = new Source(); - /** - * Interval to be used for the Duration (in milliseconds) of a non-Flux producing - * Supplier. Default is 0, which means the Supplier will only be invoked once. - */ - private long interval = 0L; + private Sink sink = new Sink(); + + private Processor processor = new Processor(); public static final String ROUTE_KEY = "stream_routekey"; + public Sink getSink() { + return this.sink; + } + + public Source getSource() { + return this.source; + } + + public Processor getProcessor() { + return this.processor; + } + + public static class Sink { + + /** + * The name of a single consumer to wire up to the input channel. Default is null, + * which means all consumers are bound. + */ + private String name; + + /** + * Flag to be able to switch off binding consumers to input streams. + */ + private boolean enabled; + + public String getName() { + return this.name; + } + + public void setName(String name) { + this.name = name; + } + + public boolean isEnabled() { + return this.enabled; + } + + public void setEnabled(boolean enabled) { + this.enabled = enabled; + } + } + + public static class Source { + + /** + * The name of a single supplier to wire up to the output channel. Default is + * null, which means all suppliers are bound. + */ + private String name; + + /** + * Flag to be able to switch off binding suppliers to output streams. Because of + * the underlying behaviour of Spring AMQP etc., sources do not start up and fail + * gracefully if the broker is down. So this flag is needed to control the + * behaviour if you know the broker is not available and there are suppliers. + */ + private boolean enabled; + + /** + * Interval to be used for the Duration (in milliseconds) of a non-Flux producing + * Supplier. Default is 0, which means the Supplier will only be invoked once. + */ + private long interval = 0L; + + public long getInterval() { + return interval; + } + + public void setInterval(long interval) { + this.interval = interval; + } + + public boolean isEnabled() { + return this.enabled; + } + + public void setEnabled(boolean enabled) { + this.enabled = enabled; + } + + public String getName() { + return this.name; + } + + public void setName(String name) { + this.name = name; + } + } + + public static class Processor { + + /** + * The name of a single processor to wire up to the input and output channels. + * Default is null, which means all functions are bound. + */ + private String name; + + /** + * Flag to be able to switch off binding consumers to input streams. + */ + private boolean enabled; + + public String getName() { + return this.name; + } + + public void setName(String name) { + this.name = name; + } + + public boolean isEnabled() { + return this.enabled; + } + + public void setEnabled(boolean enabled) { + this.enabled = enabled; + } + } + public String getDefaultRoute() { - return defaultRoute; + return processor.getName()!=null ? processor.getName() : sink.getName(); } - public void setDefaultRoute(String defaultRoute) { - this.defaultRoute = defaultRoute; - } - - public long getInterval() { - return interval; - } - - public void setInterval(long interval) { - this.interval = interval; - } } diff --git a/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/config/StreamListeningConsumerInvoker.java b/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/config/StreamListeningConsumerInvoker.java new file mode 100644 index 000000000..71267b9b5 --- /dev/null +++ b/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/config/StreamListeningConsumerInvoker.java @@ -0,0 +1,194 @@ +/* + * Copyright 2016 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 + * + * http://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.stream.config; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Consumer; +import java.util.function.Function; + +import org.springframework.beans.factory.SmartInitializingSingleton; +import org.springframework.cloud.function.context.FunctionCatalog; +import org.springframework.cloud.function.context.catalog.FunctionInspector; +import org.springframework.cloud.function.context.message.MessageUtils; +import org.springframework.cloud.stream.annotation.Input; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; +import org.springframework.cloud.stream.messaging.Processor; +import org.springframework.messaging.Message; +import org.springframework.messaging.converter.MessageConverter; + +import reactor.core.publisher.Flux; + +/** + * @author Dave Syer + */ +public class StreamListeningConsumerInvoker implements SmartInitializingSingleton { + + private final FunctionInspector functionInspector; + + private final FunctionCatalog functionCatalog; + + private final CompositeMessageConverterFactory converterFactory; + + private MessageConverter converter; + + private final String defaultRoute; + + private final Map processors = new HashMap<>(); + + private int count = -1; + + private static final FluxMessageProcessor NOENDPOINT = flux -> Flux.empty(); + + private static final Object UNCONVERTED = new Object(); + + public StreamListeningConsumerInvoker(FunctionCatalog functionCatalog, + FunctionInspector functionInspector, + CompositeMessageConverterFactory converterFactory, String defaultRoute) { + this.functionCatalog = functionCatalog; + this.functionInspector = functionInspector; + this.converterFactory = converterFactory; + this.defaultRoute = defaultRoute; + } + + @Override + public void afterSingletonsInstantiated() { + this.converter = this.converterFactory.getMessageConverterForAllRegistered(); + } + + @StreamListener + public void handle(@Input(Processor.INPUT) Flux> input) { + input.groupBy(this::select).flatMap(group -> group.key().process(group)).subscribe(); + } + + private Flux> consumer(String name, Flux> flux) { + Consumer consumer = functionCatalog.lookup(Consumer.class, name); + consumer.accept(flux.map(message -> convertInput(consumer).apply(message)) + .filter(transformed -> transformed != UNCONVERTED)); + return Flux.empty(); + } + + private Flux> balance(List names, Flux> flux) { + if (names.isEmpty()) { + return Flux.empty(); + } + String name = choose(names); + if (functionCatalog.lookup(Consumer.class, name) != null) { + return consumer(name, flux); + } + return Flux.empty(); + } + + private synchronized String choose(List names) { + if (++count >= names.size() || count < 0) { + count = 0; + } + return names.get(count); + } + + private FluxMessageProcessor select(Message input) { + String name = null; + if (input.getHeaders().containsKey(StreamConfigurationProperties.ROUTE_KEY)) { + String key = (String) input.getHeaders() + .get(StreamConfigurationProperties.ROUTE_KEY); + name = stash(key); + } + if (name == null && defaultRoute != null) { + name = stash(defaultRoute); + } + if (name == null) { + Set names = new LinkedHashSet<>( + functionCatalog.getNames(Consumer.class)); + List matches = new ArrayList<>(); + if (names.size() == 1) { + String key = names.iterator().next(); + name = stash(key); + } + else { + for (String candidate : names) { + Object function = functionCatalog.lookup(Consumer.class, candidate); + if (function == null) { + continue; + } + Class inputType = functionInspector.getInputType(function); + Object value = this.converter.fromMessage(input, inputType); + if (value != null && inputType.isInstance(value)) { + matches.add(candidate); + } + } + if (matches.size() == 1) { + name = stash(matches.iterator().next()); + } + else { + // TODO: do we really want this? Or maybe warn that it is happening? + return flux -> balance(matches, flux); + } + } + } + if (name == null) { + return NOENDPOINT; + } + return processors.get(name); + } + + private String stash(String key) { + if (functionCatalog.lookup(Consumer.class, key) != null) { + if (!processors.containsKey(key)) { + processors.put(key, flux -> consumer(key, flux)); + } + return key; + } + return null; + } + + private Function, Object> convertInput(Object function) { + Class inputType = functionInspector.getInputType(function); + return m -> { + if (functionInspector.isMessage(function)) { + return MessageUtils.create(function, convertPayload(inputType, m), + m.getHeaders()); + } + else { + return convertPayload(inputType, m); + } + }; + } + + private Object convertPayload(Class inputType, Message m) { + Object result; + if (inputType.isAssignableFrom(m.getPayload().getClass())) { + result = m.getPayload(); + } + else { + result = this.converter.fromMessage(m, inputType); + } + if (result == null) { + result = UNCONVERTED; + } + return result; + } + + interface FluxMessageProcessor { + Flux> process(Flux> flux); + } + +} diff --git a/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/config/SupplierInvokingMessageProducer.java b/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/config/SupplierInvokingMessageProducer.java index 0ac92d432..668238a64 100644 --- a/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/config/SupplierInvokingMessageProducer.java +++ b/spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/config/SupplierInvokingMessageProducer.java @@ -28,6 +28,7 @@ import org.springframework.cloud.stream.messaging.Source; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.messaging.Message; import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.StringUtils; import reactor.core.Disposable; import reactor.core.publisher.Flux; @@ -44,15 +45,24 @@ public class SupplierInvokingMessageProducer extends MessageProducerSupport { private final Map disposables = new HashMap<>(); - public SupplierInvokingMessageProducer(FunctionCatalog registry) { + private String defaultRoute; + + public SupplierInvokingMessageProducer(FunctionCatalog registry, + String defaultRoute) { this.functionCatalog = registry; + this.defaultRoute = defaultRoute; this.setOutputChannelName(Source.OUTPUT); } @Override protected void doStart() { - for (String name : functionCatalog.getNames(Supplier.class)) { - start(name); + if (StringUtils.hasText(this.defaultRoute)) { + start(this.defaultRoute); + } + else { + for (String name : functionCatalog.getNames(Supplier.class)) { + start(name); + } } } @@ -83,7 +93,8 @@ public class SupplierInvokingMessageProducer extends MessageProducerSupport { if (!disposables.containsKey(name)) { synchronized (disposables) { if (!disposables.containsKey(name)) { - Supplier> supplier = functionCatalog.lookup(Supplier.class, name); + Supplier> supplier = functionCatalog.lookup(Supplier.class, + name); if (supplier != null) { suppliers.add(name); disposables.put(name, diff --git a/spring-cloud-function-stream/src/test/java/org/springframework/cloud/function/stream/mixed/PojoStreamingExplicitEndpointTests.java b/spring-cloud-function-stream/src/test/java/org/springframework/cloud/function/stream/mixed/PojoStreamingExplicitProcessorEnabledTests.java similarity index 86% rename from spring-cloud-function-stream/src/test/java/org/springframework/cloud/function/stream/mixed/PojoStreamingExplicitEndpointTests.java rename to spring-cloud-function-stream/src/test/java/org/springframework/cloud/function/stream/mixed/PojoStreamingExplicitProcessorEnabledTests.java index 4d4741e40..e89210bf4 100644 --- a/spring-cloud-function-stream/src/test/java/org/springframework/cloud/function/stream/mixed/PojoStreamingExplicitEndpointTests.java +++ b/spring-cloud-function-stream/src/test/java/org/springframework/cloud/function/stream/mixed/PojoStreamingExplicitProcessorEnabledTests.java @@ -43,17 +43,15 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Marius Bogoevici */ @RunWith(SpringRunner.class) -@SpringBootTest(classes = PojoStreamingExplicitEndpointTests.StreamingFunctionApplication.class, properties = { - "spring.cloud.function.stream.default-route=uppercase", - "logging.level.org.springframework.integration=DEBUG", "debug=TRUE" }) -public class PojoStreamingExplicitEndpointTests { +@SpringBootTest(classes = PojoStreamingExplicitProcessorEnabledTests.StreamingFunctionApplication.class, properties = "spring.cloud.function.stream.processor.name=uppercase") +public class PojoStreamingExplicitProcessorEnabledTests { @Autowired Processor processor; @Autowired MessageCollector messageCollector; - + @Autowired StreamingFunctionApplication app; @@ -68,14 +66,15 @@ public class PojoStreamingExplicitEndpointTests { @Test public void testRoutingBeatsDefaultEndpoint() throws Exception { - processor.input() - .send(MessageBuilder.withPayload("{\"name\":\"hello\"}").setHeader(StreamConfigurationProperties.ROUTE_KEY, "sink").build()); + processor.input().send(MessageBuilder.withPayload("{\"name\":\"hello\"}") + .setHeader(StreamConfigurationProperties.ROUTE_KEY, "sink").build()); assertThat(app.foos).hasSize(1); + assertThat(app.foos.get(0).getName()).isEqualTo("hello"); } @SpringBootApplication public static class StreamingFunctionApplication { - + private List foos = new ArrayList<>(); @Bean diff --git a/spring-cloud-function-stream/src/test/java/org/springframework/cloud/function/stream/mixed/PojoStreamingExplicitSinkEnabledTests.java b/spring-cloud-function-stream/src/test/java/org/springframework/cloud/function/stream/mixed/PojoStreamingExplicitSinkEnabledTests.java new file mode 100644 index 000000000..b43b89e33 --- /dev/null +++ b/spring-cloud-function-stream/src/test/java/org/springframework/cloud/function/stream/mixed/PojoStreamingExplicitSinkEnabledTests.java @@ -0,0 +1,122 @@ +/* + * Copyright 2017 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 + * + * http://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.stream.mixed; + +import java.util.ArrayList; +import java.util.List; +import java.util.function.Consumer; +import java.util.function.Function; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.stream.messaging.Sink; +import org.springframework.context.annotation.Bean; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Marius Bogoevici + */ +@RunWith(SpringRunner.class) +@SpringBootTest(classes = PojoStreamingExplicitSinkEnabledTests.StreamingFunctionApplication.class, properties = { + "spring.cloud.function.stream.source.enabled=false", + "spring.cloud.function.stream.sink.name=uppercase,sink" }) +public class PojoStreamingExplicitSinkEnabledTests { + + @Autowired + Sink sink; + + @Autowired + List collector; + + @Before + public void init() { + collector.clear(); + } + + @Test + public void routing() throws Exception { + sink.input().send(MessageBuilder.withPayload("{\"name\":\"hello\"}").build()); + assertThat(collector).hasSize(1); + assertThat(collector.get(0).getName()).isEqualTo("HELLO"); + } + + @SpringBootApplication + public static class StreamingFunctionApplication { + + @Bean + public Function uppercase() { + return f -> new Bar(f.getName().toUpperCase()); + } + + @Bean + public List collector() { + return new ArrayList<>(); + } + + @Bean + public Consumer sink(final List list) { + return s -> list.add(s); + } + + } + + protected static class Foo { + private String name; + + Foo() { + } + + public Foo(String name) { + this.name = name; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + } + + protected static class Bar { + private String name; + + Bar() { + } + + public Bar(String name) { + this.name = name; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + } +} diff --git a/spring-cloud-function-stream/src/test/java/org/springframework/cloud/function/stream/mixed/PojoStreamingExplicitSourceEnabledTests.java b/spring-cloud-function-stream/src/test/java/org/springframework/cloud/function/stream/mixed/PojoStreamingExplicitSourceEnabledTests.java new file mode 100644 index 000000000..1ee596820 --- /dev/null +++ b/spring-cloud-function-stream/src/test/java/org/springframework/cloud/function/stream/mixed/PojoStreamingExplicitSourceEnabledTests.java @@ -0,0 +1,120 @@ +/* + * Copyright 2017 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 + * + * http://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.stream.mixed; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.function.Function; +import java.util.function.Supplier; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.stream.messaging.Source; +import org.springframework.cloud.stream.test.binder.MessageCollector; +import org.springframework.context.annotation.Bean; +import org.springframework.messaging.Message; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.assertj.core.api.Assertions.assertThat; + +import reactor.core.publisher.Flux; + +/** + * @author Dave Syer + */ +@RunWith(SpringRunner.class) +@SpringBootTest(classes = PojoStreamingExplicitSourceEnabledTests.StreamingFunctionApplication.class, properties = { + "spring.cloud.function.stream.sink.enabled=false", + "spring.cloud.function.stream.source.name=words,uppercase" }) +public class PojoStreamingExplicitSourceEnabledTests { + + @Autowired + Source source; + + @Autowired + MessageCollector messageCollector; + + @Test + public void routing() throws Exception { + Message message = messageCollector.forChannel(source.output()).poll(1000, + TimeUnit.MILLISECONDS); + assertThat(((Bar) message.getPayload()).getName()).isEqualTo("FOO"); + } + + @SpringBootApplication + public static class StreamingFunctionApplication { + + @Bean + public Function uppercase() { + return f -> new Bar(f.getName().toUpperCase()); + } + + @Bean + public List collector() { + return new ArrayList<>(); + } + + @Bean + public Supplier> words(final List list) { + return () -> Flux.just(new Foo("foo"), new Foo("bar")); + } + + } + + protected static class Foo { + private String name; + + Foo() { + } + + public Foo(String name) { + this.name = name; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + } + + protected static class Bar { + private String name; + + Bar() { + } + + public Bar(String name) { + this.name = name; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + } +} diff --git a/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/DefaultRouteTests.java b/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/DefaultRouteTests.java new file mode 100644 index 000000000..6b5093cbe --- /dev/null +++ b/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/DefaultRouteTests.java @@ -0,0 +1,78 @@ +/* + * Copyright 2012-2015 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 + * + * http://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.web; + +import java.net.URI; +import java.util.function.Function; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.context.embedded.LocalServerPort; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.context.SpringBootTest.WebEnvironment; +import org.springframework.boot.test.web.client.TestRestTemplate; +import org.springframework.context.annotation.Bean; +import org.springframework.http.HttpStatus; +import org.springframework.http.RequestEntity; +import org.springframework.http.ResponseEntity; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.assertj.core.api.Assertions.assertThat; + +import reactor.core.publisher.Flux; + +/** + * @author Dave Syer + * + */ +@RunWith(SpringRunner.class) +@SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT, properties = "") +public class DefaultRouteTests { + + @LocalServerPort + private int port; + @Autowired + private TestRestTemplate rest; + + @Test + public void explicit() throws Exception { + ResponseEntity result = rest.exchange( + RequestEntity.post(new URI("/uppercase")).body("foo"), String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(result.getBody()).isEqualTo("FOO"); + } + + @Test + public void implicit() throws Exception { + ResponseEntity result = rest.exchange( + RequestEntity.post(new URI("/")).body("foo"), String.class); + assertThat(result.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(result.getBody()).isEqualTo("FOO"); + } + + @EnableAutoConfiguration + @org.springframework.boot.test.context.TestConfiguration + protected static class TestConfiguration { + @Bean + public Function, Flux> uppercase() { + return flux -> flux.map(value -> value.toUpperCase()); + } + } +}