Prefer explicit route to default

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/ContextFunctionCatalogAutoConfiguration.java

@@ -250,7 +250,7 @@ public class ContextFunctionCatalogAutoConfiguration {
 			if (stages.length == 0 && source.size() == 1) {
 				stages = new String[] { source.keySet().iterator().next() };
 			}
-			Object function = lookup(stages[0], source);
+			Object function = stages.length>0 ? lookup(stages[0], source) : null;
 			if (function == null) {
 				return null;
 			}

spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/BeanFactoryFunctionCatalogTests.java

@@ -54,6 +54,13 @@ public class BeanFactoryFunctionCatalogTests {
 		assertThat(foos.apply(Flux.just(2)).blockFirst()).isEqualTo("4");
 	}
 
+	@Test
+	public void lookupNonExistentConsumerWithEmptyName() {
+		processor.register(new FunctionRegistration<>(new Foos()).names("foos"));
+		Consumer<Flux<String>> foos = processor.lookupConsumer("");
+		assertThat(foos).isNull();
+	}
+
 	@Test
 	public void composeFunction() {
 		processor.register(new FunctionRegistration<>(new Foos()).names("foos"));

spring-cloud-function-samples/function-sample/src/main/resources/application.properties

@@ -1 +1,2 @@
+spring.cloud.function.stream.endpoint: uppercase
 spring.cloud.function.scan.packages: com.example.functions

spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/StreamListeningFunctionInvoker.java

@@ -131,16 +131,14 @@ public class StreamListeningFunctionInvoker implements SmartInitializingSingleto
 	}
 
 	private FluxMessageProcessor select(Message<?> input) {
-		String name = defaultEndpoint;
+		String name = null;
-		if (name != null) {
+		if (input.getHeaders().containsKey(StreamConfigurationProperties.ROUTE_KEY)) {
-			name = stash(name);
+			String key = (String) input.getHeaders()
+					.get(StreamConfigurationProperties.ROUTE_KEY);
+			name = stash(key);
 		}
-		if (name == null) {
+		if (name==null && defaultEndpoint != null) {
-			if (input.getHeaders().containsKey(StreamConfigurationProperties.ROUTE_KEY)) {
+			name = stash(defaultEndpoint);
-				String key = (String) input.getHeaders()
-						.get(StreamConfigurationProperties.ROUTE_KEY);
-				name = stash(key);
-			}
 		}
 		if (name == null) {
 			Set<String> names = new LinkedHashSet<>(functionCatalog.getFunctionNames());

spring-cloud-function-stream/src/test/java/org/springframework/cloud/function/stream/mixed/PojoStreamingExplicitEndpointTests.java

@@ -16,7 +16,10 @@
 
 package org.springframework.cloud.function.stream.mixed;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
@@ -26,6 +29,7 @@ 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.function.stream.StreamConfigurationProperties;
 import org.springframework.cloud.stream.messaging.Processor;
 import org.springframework.cloud.stream.test.binder.MessageCollector;
 import org.springframework.context.annotation.Bean;
@@ -50,8 +54,11 @@ public class PojoStreamingExplicitEndpointTests {
 	@Autowired
 	MessageCollector messageCollector;
 	
+	@Autowired
+	StreamingFunctionApplication app;
+
 	@Test
-	public void test() throws Exception {
+	public void testDefaultEndpoint() throws Exception {
 		processor.input()
 				.send(MessageBuilder.withPayload("{\"name\":\"hello\"}").build());
 		Message<?> result = messageCollector.forChannel(processor.output()).poll(1000,
@@ -59,9 +66,18 @@ public class PojoStreamingExplicitEndpointTests {
 		assertThat(result.getPayload()).isInstanceOf(Foo.class);
 	}
 
+	@Test
+	public void testRoutingBeatsDefaultEndpoint() throws Exception {
+		processor.input()
+				.send(MessageBuilder.withPayload("{\"name\":\"hello\"}").setHeader(StreamConfigurationProperties.ROUTE_KEY, "sink").build());
+		assertThat(app.foos).hasSize(1);
+	}
+
 	@SpringBootApplication
 	public static class StreamingFunctionApplication {
 		
+		private List<Foo> foos = new ArrayList<>();
+
 		@Bean
 		public Function<Foo, Foo> uppercase() {
 			return f -> new Foo(f.getName().toUpperCase());
@@ -72,6 +88,11 @@ public class PojoStreamingExplicitEndpointTests {
 			return () -> new Foo("world");
 		}
 
+		@Bean
+		public Consumer<Foo> sink() {
+			return foo -> foos.add(foo);
+		}
+
 	}
 
 	protected static class Foo {