Dynamic destination refactoring

source-samples/dynamic-destination-source/pom.xml

@@ -9,71 +9,57 @@
 	<description>Spring Cloud Stream Sample JDBC Source App</description>
 
 	<parent>
-		<groupId>io.spring.cloud.stream.sample</groupId>
-		<artifactId>spring-cloud-stream-samples-parent</artifactId>
-		<version>0.0.1-SNAPSHOT</version>
-        <relativePath>../..</relativePath>
+		<groupId>org.springframework.boot</groupId>
+		<artifactId>spring-boot-starter-parent</artifactId>
+		<version>2.2.0.RELEASE</version>
+		<relativePath/> <!-- lookup parent from repository -->
 	</parent>
 
+	<properties>
+		<spring-cloud.version>Hoxton.BUILD-SNAPSHOT</spring-cloud.version>
+	</properties>
+
+	<dependencyManagement>
+		<dependencies>
+			<dependency>
+				<groupId>org.springframework.cloud</groupId>
+				<artifactId>spring-cloud-dependencies</artifactId>
+				<version>${spring-cloud.version}</version>
+				<type>pom</type>
+				<scope>import</scope>
+			</dependency>
+		</dependencies>
+	</dependencyManagement>
+
 	<dependencies>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-binder-kafka</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-test</artifactId>
 			<scope>test</scope>
 		</dependency>
 		<dependency>
-			<groupId>org.springframework.cloud</groupId>
-			<artifactId>spring-cloud-stream-test-support</artifactId>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-actuator</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-web</artifactId>
+		</dependency>
 	</dependencies>
 
-	<profiles>
-		<profile>
-			<id>kafka-binder</id>
-			<activation>
-				<activeByDefault>true</activeByDefault>
-			</activation>
-			<dependencies>
-				<dependency>
-					<groupId>org.springframework.cloud</groupId>
-					<artifactId>spring-cloud-stream-binder-kafka</artifactId>
-				</dependency>
-			</dependencies>
-			<build>
-				<plugins>
-					<plugin>
-						<groupId>org.springframework.boot</groupId>
-						<artifactId>spring-boot-maven-plugin</artifactId>
-						<configuration>
-							<classifier>kafka</classifier>
-						</configuration>
-					</plugin>
-				</plugins>
-			</build>
-		</profile>
-		<profile>
-			<id>rabbit-binder</id>
-			<dependencies>
-				<dependency>
-					<groupId>org.springframework.cloud</groupId>
-					<artifactId>spring-cloud-stream-binder-rabbit</artifactId>
-				</dependency>
-			</dependencies>
-			<build>
-				<plugins>
-					<plugin>
-						<groupId>org.springframework.boot</groupId>
-						<artifactId>spring-boot-maven-plugin</artifactId>
-						<configuration>
-							<classifier>rabbit</classifier>
-						</configuration>
-					</plugin>
-				</plugins>
-			</build>
-		</profile>
-	</profiles>
-
 	<build>
 		<plugins>
 			<plugin>
@@ -83,4 +69,55 @@
 		</plugins>
 	</build>
 
+	<repositories>
+		<repository>
+			<id>spring-snapshots</id>
+			<name>Spring Snapshots</name>
+			<url>https://repo.spring.io/libs-snapshot-local</url>
+			<snapshots>
+				<enabled>true</enabled>
+			</snapshots>
+			<releases>
+				<enabled>false</enabled>
+			</releases>
+		</repository>
+		<repository>
+			<id>spring-milestones</id>
+			<name>Spring Milestones</name>
+			<url>https://repo.spring.io/libs-milestone-local</url>
+			<snapshots>
+				<enabled>false</enabled>
+			</snapshots>
+		</repository>
+	</repositories>
+	<pluginRepositories>
+		<pluginRepository>
+			<id>spring-snapshots</id>
+			<name>Spring Snapshots</name>
+			<url>https://repo.spring.io/libs-snapshot-local</url>
+			<snapshots>
+				<enabled>true</enabled>
+			</snapshots>
+			<releases>
+				<enabled>false</enabled>
+			</releases>
+		</pluginRepository>
+		<pluginRepository>
+			<id>spring-milestones</id>
+			<name>Spring Milestones</name>
+			<url>https://repo.spring.io/libs-milestone-local</url>
+			<snapshots>
+				<enabled>false</enabled>
+			</snapshots>
+		</pluginRepository>
+		<pluginRepository>
+			<id>spring-releases</id>
+			<name>Spring Releases</name>
+			<url>https://repo.spring.io/libs-release-local</url>
+			<snapshots>
+				<enabled>false</enabled>
+			</snapshots>
+		</pluginRepository>
+	</pluginRepositories>
+
 </project>

source-samples/dynamic-destination-source/src/main/java/demo/DynamicDestinationSourceApplication.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright 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
+ *
+ *      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 demo;
+
+import java.util.Map;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestHeader;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.ResponseStatus;
+import org.springframework.web.bind.annotation.RestController;
+import reactor.core.publisher.EmitterProcessor;
+import reactor.core.publisher.Flux;
+
+import static org.springframework.web.bind.annotation.RequestMethod.POST;
+
+@SpringBootApplication
+@RestController
+public class DynamicDestinationSourceApplication {
+
+	@Autowired
+	private ObjectMapper jsonMapper;
+
+	private final EmitterProcessor<Message<?>> processor = EmitterProcessor.create();
+
+	public static void main(String[] args) {
+		SpringApplication.run(DynamicDestinationSourceApplication.class, args);
+	}
+
+	@SuppressWarnings("unchecked")
+	@RequestMapping(path = "/", method = POST, consumes = "*/*")
+	@ResponseStatus(HttpStatus.ACCEPTED)
+	public void handleRequest(@RequestBody String body, @RequestHeader(HttpHeaders.CONTENT_TYPE) Object contentType) throws Exception {
+		Map<String, String> payload = jsonMapper.readValue(body, Map.class);
+		String destinationName = payload.get("id");
+		Message<?> message = MessageBuilder.withPayload(payload)
+				.setHeader("spring.cloud.stream.sendto.destination", destinationName).build();
+		processor.onNext(message);
+	}
+
+	@Bean
+	public Supplier<Flux<Message<?>>> supplier() {
+		return () ->{
+			System.out.println("hello");
+			return processor;
+		};
+	}
+
+	//Following sink is used as test consumer. It logs the data received through the consumer.
+	static class TestSink {
+
+		private final Log logger = LogFactory.getLog(getClass());
+
+		@Bean
+		public Consumer<String> receive1() {
+			return data -> logger.info("Data received from customer-1..." + data);
+		}
+
+		@Bean
+		public Consumer<String> receive2() {
+			return data -> logger.info("Data received from customer-2..." + data);
+		}
+	}
+}

source-samples/dynamic-destination-source/src/main/java/demo/SourceApplication.java

@@ -1,29 +0,0 @@
-/*
- * Copyright 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
- *
- *      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 demo;
-
-import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.SpringBootApplication;
-
-@SpringBootApplication
-public class SourceApplication {
-
-	public static void main(String[] args) {
-		SpringApplication.run(SourceApplication.class, args);
-	}
-
-}

source-samples/dynamic-destination-source/src/main/java/demo/SourceWithDynamicDestination.java

@@ -1,119 +0,0 @@
-/*
- * Copyright 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
- *
- *      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 demo;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.cloud.stream.annotation.EnableBinding;
-import org.springframework.cloud.stream.annotation.Input;
-import org.springframework.cloud.stream.annotation.StreamListener;
-import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
-import org.springframework.context.annotation.Bean;
-import org.springframework.expression.spel.standard.SpelExpressionParser;
-import org.springframework.http.HttpHeaders;
-import org.springframework.http.HttpStatus;
-import org.springframework.integration.annotation.ServiceActivator;
-import org.springframework.integration.channel.DirectChannel;
-import org.springframework.integration.router.ExpressionEvaluatingRouter;
-import org.springframework.messaging.MessageChannel;
-import org.springframework.messaging.MessageHeaders;
-import org.springframework.messaging.SubscribableChannel;
-import org.springframework.messaging.support.MessageBuilder;
-import org.springframework.stereotype.Controller;
-import org.springframework.web.bind.annotation.RequestBody;
-import org.springframework.web.bind.annotation.RequestHeader;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.ResponseStatus;
-
-import java.util.Collections;
-
-import static org.springframework.web.bind.annotation.RequestMethod.POST;
-
-/**
- * @author Ilayaperumal Gopinathan
- * @author Soby Chacko
- */
-@EnableBinding
-@Controller
-public class SourceWithDynamicDestination {
-
-	@Autowired
-	private BinderAwareChannelResolver resolver;
-
-	@Autowired
-	@Qualifier("sourceChannel")
-	private MessageChannel localChannel;
-
-	@RequestMapping(path = "/", method = POST, consumes = "*/*")
-	@ResponseStatus(HttpStatus.ACCEPTED)
-	public void handleRequest(@RequestBody String body, @RequestHeader(HttpHeaders.CONTENT_TYPE) Object contentType) {
-		sendMessage(body, contentType);
-	}
-
-	private void sendMessage(Object body, Object contentType) {
-		localChannel.send(MessageBuilder.createMessage(body,
-				new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, contentType))));
-	}
-
-	@Bean(name = "sourceChannel")
-	public MessageChannel localChannel() {
-		return new DirectChannel();
-	}
-
-	@Bean
-	@ServiceActivator(inputChannel = "sourceChannel")
-	public ExpressionEvaluatingRouter router() {
-		ExpressionEvaluatingRouter router = new ExpressionEvaluatingRouter(new SpelExpressionParser().parseExpression("payload.id"));
-		router.setDefaultOutputChannelName("default-output");
-		router.setChannelResolver(resolver);
-		return router;
-	}
-
-	//Following sink is used as test consumer. It logs the data received through the consumer.
-	@EnableBinding(Sink.class)
-	static class TestSink {
-
-		private final Log logger = LogFactory.getLog(getClass());
-
-		@StreamListener(Sink.INPUT1)
-		public void receive(String data) {
-			logger.info("Data received from customer-1..." + data);
-		}
-
-		@StreamListener(Sink.INPUT2)
-		public void receiveX(String data) {
-			logger.info("Data received from customer-2..." + data);
-		}
-	}
-
-	interface Sink {
-
-		String INPUT1 = "input1";
-		String INPUT2 = "input2";
-
-
-		@Input(INPUT1)
-		SubscribableChannel input1();
-
-
-		@Input(INPUT2)
-		SubscribableChannel input2();
-
-	}
-}

source-samples/dynamic-destination-source/src/main/resources/application.yml

@@ -1,3 +1,6 @@
 # Input bindings used for testing
-spring.cloud.stream.bindings.input1.destination: customerId-1
-spring.cloud.stream.bindings.input2.destination: customerId-2
+spring.cloud.stream:
+  function.definition: supplier;receive1;receive2;
+  bindings:
+    receive1-in-0.destination: customerId-1
+    receive2-in-0.destination: customerId-2

source-samples/dynamic-destination-source/src/test/java/demo/ModuleApplicationTests.java

@@ -25,7 +25,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.test.context.web.WebAppConfiguration;
 
 @RunWith(SpringJUnit4ClassRunner.class)
-@SpringBootTest(classes = SourceApplication.class)
+@SpringBootTest(classes = DynamicDestinationSourceApplication.class)
 @WebAppConfiguration
 @DirtiesContext
 public class ModuleApplicationTests {