diff --git a/.gitignore b/.gitignore index eade3d4..61b922a 100644 --- a/.gitignore +++ b/.gitignore @@ -13,6 +13,7 @@ _site/ .settings .springBeans .DS_Store +*/.idea *.sw* *.iml *.ipr diff --git a/dynamic-source/README.md b/dynamic-source/README.md new file mode 100644 index 0000000..211bc06 --- /dev/null +++ b/dynamic-source/README.md @@ -0,0 +1,43 @@ +Spring Cloud Stream Source Sample +============================= + +In this *Spring Cloud Stream* sample, a source application publishes messages to dynamically created destinations. + +## Requirements + +To run this sample, you will need to have installed: + +* Java 8 or Above + +This example requires RabbitMQ to be running on localhost. + +## Code Tour + +The class `SourceWithDynamicDestination` is a REST controller that registers the 'POST' request mapping for '/'. +When a payload is sent to 'http://localhost:8080/' by a POST request (port 8080 is the default), this application +then uses `BinderAwareChannelResolver` to resolve the destination dynamically at runtime. Currently, this resolver uses +`payload` as the SpEL expression to resolve the destination name. Hence, if a payload `testing` is sent to the app, then +this source application sends the message `testing` into the Rabbit exchange `testing`. This exchange or topic (in case +of Kafka if Kafka binder is used) is created dynamically and bound to send the payload. + + +Upon starting the application on the default port 8080, if the following data are sent: + +curl -H "Content-Type: application/json" -X POST -d '{"id":"customerId-1","bill-pay":"100"}' http://localhost:8080 + +curl -H "Content-Type: application/json" -X POST -d '{"id":"customerId-2","bill-pay":"150"}' http://localhost:8080 + +The destinations 'customerId-1' and 'customerId-2' are created at the broker (for example: exchange in case of Rabbit or topic in case of Kafka with the names 'customerId-1' and 'customerId-2') and the data are published to the appropriate destinations dynamically. + +## Building with Maven + +Build the sample by executing: + + source>$ mvn clean package + +## Running the Sample + +To start the source module execute the following: + + source>$ java -jar target/spring-cloud-stream-sample-dynamic-source-1.1.0.BUILD-SNAPSHOT-exec.jar + diff --git a/dynamic-source/pom.xml b/dynamic-source/pom.xml new file mode 100644 index 0000000..fd3e30d --- /dev/null +++ b/dynamic-source/pom.xml @@ -0,0 +1,62 @@ + + + 4.0.0 + + spring-cloud-stream-sample-dynamic-source + jar + spring-cloud-stream-sample-dynamic-source + Demo project for source module + + + org.springframework.cloud + spring-cloud-stream-samples + 1.1.0.BUILD-SNAPSHOT + + + + demo.SourceApplication + + + + + org.springframework.cloud + spring-cloud-stream + + + org.springframework.cloud + spring-cloud-stream-codec + + + org.springframework.integration + spring-integration-java-dsl + 1.2.1.RELEASE + + + org.springframework.cloud + spring-cloud-stream-binder-rabbit + + + org.springframework.boot + spring-boot-configuration-processor + true + + + org.springframework.boot + spring-boot-starter-test + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + exec + + + + + + diff --git a/dynamic-source/src/main/java/demo/SourceApplication.java b/dynamic-source/src/main/java/demo/SourceApplication.java new file mode 100644 index 0000000..2770520 --- /dev/null +++ b/dynamic-source/src/main/java/demo/SourceApplication.java @@ -0,0 +1,29 @@ +/* + * 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 + * + * 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 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/dynamic-source/src/main/java/demo/SourceWithDynamicDestination.java b/dynamic-source/src/main/java/demo/SourceWithDynamicDestination.java new file mode 100644 index 0000000..c29691a --- /dev/null +++ b/dynamic-source/src/main/java/demo/SourceWithDynamicDestination.java @@ -0,0 +1,87 @@ +/* + * 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 + * + * 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 demo; + +import java.util.Collections; + +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.binding.BinderAwareChannelResolver; +import org.springframework.context.annotation.Bean; +import org.springframework.expression.Expression; +import org.springframework.expression.common.LiteralExpression; +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.dsl.support.Function; +import org.springframework.integration.dsl.support.FunctionExpression; +import org.springframework.integration.router.AbstractMappingMessageRouter; +import org.springframework.integration.router.ExpressionEvaluatingRouter; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHeaders; +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 static org.springframework.web.bind.annotation.RequestMethod.POST; + +/** + * @author Ilayaperumal Gopinathan + */ +@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; + } +} diff --git a/dynamic-source/src/test/java/demo/ModuleApplicationTests.java b/dynamic-source/src/test/java/demo/ModuleApplicationTests.java new file mode 100644 index 0000000..35a953a --- /dev/null +++ b/dynamic-source/src/test/java/demo/ModuleApplicationTests.java @@ -0,0 +1,37 @@ +/* + * 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 + * + * 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 demo; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.web.WebAppConfiguration; + +@RunWith(SpringJUnit4ClassRunner.class) +@SpringBootTest(classes = SourceApplication.class) +@WebAppConfiguration +@DirtiesContext +public class ModuleApplicationTests { + + @Test + public void contextLoads() { + } + +}