diff --git a/source-samples/dynamic-destination-source/pom.xml b/source-samples/dynamic-destination-source/pom.xml index dcef41c..03eeabe 100644 --- a/source-samples/dynamic-destination-source/pom.xml +++ b/source-samples/dynamic-destination-source/pom.xml @@ -9,71 +9,57 @@ Spring Cloud Stream Sample JDBC Source App - io.spring.cloud.stream.sample - spring-cloud-stream-samples-parent - 0.0.1-SNAPSHOT - ../.. + org.springframework.boot + spring-boot-starter-parent + 2.2.0.RELEASE + + + Hoxton.BUILD-SNAPSHOT + + + + + + org.springframework.cloud + spring-cloud-dependencies + ${spring-cloud.version} + pom + import + + + + + + org.springframework.cloud + spring-cloud-stream-binder-kafka + org.springframework.boot spring-boot-starter-test test - org.springframework.cloud - spring-cloud-stream-test-support + org.springframework.kafka + spring-kafka-test test + + org.springframework.boot + spring-boot-starter-actuator + + + org.springframework.boot + spring-boot-starter + + + org.springframework.boot + spring-boot-starter-web + - - - kafka-binder - - true - - - - org.springframework.cloud - spring-cloud-stream-binder-kafka - - - - - - org.springframework.boot - spring-boot-maven-plugin - - kafka - - - - - - - rabbit-binder - - - org.springframework.cloud - spring-cloud-stream-binder-rabbit - - - - - - org.springframework.boot - spring-boot-maven-plugin - - rabbit - - - - - - - @@ -83,4 +69,55 @@ + + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + true + + + false + + + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + false + + + + + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + true + + + false + + + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + false + + + + spring-releases + Spring Releases + https://repo.spring.io/libs-release-local + + false + + + + diff --git a/source-samples/dynamic-destination-source/src/main/java/demo/DynamicDestinationSourceApplication.java b/source-samples/dynamic-destination-source/src/main/java/demo/DynamicDestinationSourceApplication.java new file mode 100644 index 0000000..dd90782 --- /dev/null +++ b/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> 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 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>> 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 receive1() { + return data -> logger.info("Data received from customer-1..." + data); + } + + @Bean + public Consumer receive2() { + return data -> logger.info("Data received from customer-2..." + data); + } + } +} diff --git a/source-samples/dynamic-destination-source/src/main/java/demo/SourceApplication.java b/source-samples/dynamic-destination-source/src/main/java/demo/SourceApplication.java deleted file mode 100644 index cabd754..0000000 --- a/source-samples/dynamic-destination-source/src/main/java/demo/SourceApplication.java +++ /dev/null @@ -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); - } - -} diff --git a/source-samples/dynamic-destination-source/src/main/java/demo/SourceWithDynamicDestination.java b/source-samples/dynamic-destination-source/src/main/java/demo/SourceWithDynamicDestination.java deleted file mode 100644 index bd93ada..0000000 --- a/source-samples/dynamic-destination-source/src/main/java/demo/SourceWithDynamicDestination.java +++ /dev/null @@ -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(); - - } -} diff --git a/source-samples/dynamic-destination-source/src/main/resources/application.yml b/source-samples/dynamic-destination-source/src/main/resources/application.yml index c4a461a..9c65ce4 100644 --- a/source-samples/dynamic-destination-source/src/main/resources/application.yml +++ b/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 diff --git a/source-samples/dynamic-destination-source/src/test/java/demo/ModuleApplicationTests.java b/source-samples/dynamic-destination-source/src/test/java/demo/ModuleApplicationTests.java index b5f5827..54c59a4 100644 --- a/source-samples/dynamic-destination-source/src/test/java/demo/ModuleApplicationTests.java +++ b/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 {