diff --git a/pom.xml b/pom.xml index 11da0fe..cf7007d 100644 --- a/pom.xml +++ b/pom.xml @@ -26,6 +26,7 @@ multibinder-differentsystems rxjava-processor multi-io + stream-listener diff --git a/stream-listener/README.md b/stream-listener/README.md new file mode 100644 index 0000000..8849016 --- /dev/null +++ b/stream-listener/README.md @@ -0,0 +1,42 @@ +Spring Cloud Stream Stream Listener Sample +============================= + +In this *Spring Cloud Stream* sample, the application shows how to use StreamListener support to enable message mapping and automatic type conversion. + +## Requirements + +To run this sample, you will need to have installed: + +* Java 8 or Above + +This example requires Redis to be running on localhost. + +## Code Tour + +This sample is a Spring Boot application that bundles multiple application together to showcase how to use StreamListener to enable +message mapping and automatic type conversion. + +* TypeConversionApplication - the Spring Boot Main Application +* Converters - the class that holds the required custom message converter that converts POJO of type `Foo` to `Bar` +* SampleSource - the app that generates a message of type `Foo` that has the value `hi` +* SampleTransformer - the app that has the message handler method annotated with @StreamListener to map the process input channel to a type `Bar`. + This will make sure to apply `FooToBarConverter` automatically without having a need to specify `content-type` for the channel explicitly. + The annotation @SendTo on the message handler method will make sure to send the output to the provided output channel. +* SampleSink - the app that receives the converted message from the transformer output. + +Please note that the applications (SampleSource, SampleTransformer and SampleSink) are bundled inside the single application for the demo +purpose only. In practice, these applications run on their own. If at all they need to be bundled together, the best practice is to use +`AggregateApplicationBuilder`. Refer the sample `double` for more info on aggregate application. + +## Building with Maven + +Build the sample by executing: + + >$ mvn clean package + +## Running the Sample + +To start the source module execute the following: + + >$ java -jar target/spring-cloud-stream-sample-stream-listener-1.0.0.BUILD-SNAPSHOT-exec.jar + diff --git a/stream-listener/pom.xml b/stream-listener/pom.xml new file mode 100644 index 0000000..0b0cd49 --- /dev/null +++ b/stream-listener/pom.xml @@ -0,0 +1,59 @@ + + + 4.0.0 + + spring-cloud-stream-sample-stream-listener + jar + + spring-cloud-stream-sample-stream-listener + Demo project for stream listener + + + org.springframework.cloud + spring-cloud-stream-samples + 1.0.0.BUILD-SNAPSHOT + + + + demo.TypeConversionApplication + + + + + org.springframework.cloud + spring-cloud-stream + + + org.springframework.cloud + spring-cloud-stream-binder-redis + + + org.springframework.boot + spring-boot-starter-redis + + + 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/stream-listener/src/main/java/demo/Converters.java b/stream-listener/src/main/java/demo/Converters.java new file mode 100644 index 0000000..0f09f67 --- /dev/null +++ b/stream-listener/src/main/java/demo/Converters.java @@ -0,0 +1,101 @@ +/* + * 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 demo; + +import org.springframework.cloud.stream.converter.AbstractFromMessageConverter; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.Message; +import org.springframework.messaging.converter.MessageConversionException; +import org.springframework.util.MimeType; + +/** + * @author Ilayaperumal Gopinathan + */ +@Configuration +public class Converters { + + //Register custom converter + @Bean + public AbstractFromMessageConverter fooConverter() { + return new FooToBarConverter(); + } + + public static class Foo { + + private String value = "foo"; + + public String getValue() { + return this.value; + } + + public void setValue(String value) { + this.value = value; + } + } + + public static class Bar { + + private String value = "init"; + + public Bar(String value) { + this.value = value; + } + + public String getValue() { + return this.value; + } + + public void setValue(String value) { + this.value = value; + } + + } + + public static class FooToBarConverter extends AbstractFromMessageConverter { + + public FooToBarConverter() { + super(MimeType.valueOf("test/bar")); + } + + @Override + protected Class[] supportedTargetTypes() { + return new Class[] {Bar.class}; + } + + @Override + protected Class[] supportedPayloadTypes() { + return new Class[] {Foo.class}; + } + + @Override + public Object convertFromInternal(Message message, Class targetClass, Object conversionHint) { + Object result = null; + try { + if (message.getPayload() instanceof Foo) { + Foo fooPayload = (Foo) message.getPayload(); + result = new Bar(fooPayload.getValue()); + } + } + catch (Exception e) { + logger.error(e.getMessage(), e); + throw new MessageConversionException(e.getMessage()); + } + return result; + } + } +} diff --git a/stream-listener/src/main/java/demo/SampleSink.java b/stream-listener/src/main/java/demo/SampleSink.java new file mode 100644 index 0000000..e4c652e --- /dev/null +++ b/stream-listener/src/main/java/demo/SampleSink.java @@ -0,0 +1,45 @@ +/* + * 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.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.Input; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.messaging.SubscribableChannel; + +/** + * @author Ilayaperumal Gopinathan + */ +@EnableBinding(SampleSink.Sink.class) +public class SampleSink { + + // Sink application definition + @StreamListener(Sink.SAMPLE) + public void receive(Converters.Bar barMessage) { + System.out.println("******************"); + System.out.println("At the Sink"); + System.out.println("******************"); + System.out.println("Received transformed message " + barMessage.getValue() + " of type " + barMessage.getClass()); + } + + public interface Sink { + String SAMPLE = "sample-sink"; + + @Input(SAMPLE) + SubscribableChannel sampleSink(); + } +} diff --git a/stream-listener/src/main/java/demo/SampleSource.java b/stream-listener/src/main/java/demo/SampleSource.java new file mode 100644 index 0000000..cf5276b --- /dev/null +++ b/stream-listener/src/main/java/demo/SampleSource.java @@ -0,0 +1,57 @@ +/* + * 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 demo; + +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.Output; +import org.springframework.context.annotation.Bean; +import org.springframework.integration.annotation.InboundChannelAdapter; +import org.springframework.integration.annotation.Poller; +import org.springframework.integration.core.MessageSource; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.GenericMessage; + +/** + * @author Ilayaperumal Gopinathan + */ +@EnableBinding(SampleSource.Source.class) +public class SampleSource { + + @Bean + @InboundChannelAdapter(value = Source.SAMPLE, poller = @Poller(fixedDelay = "1000", maxMessagesPerPoll = "1")) + public MessageSource timerMessageSource() { + return new MessageSource() { + public Message receive() { + System.out.println("******************"); + System.out.println("At the Source"); + System.out.println("******************"); + Converters.Foo foo = new Converters.Foo(); + foo.setValue("hi"); + System.out.println("Sending value: " + foo.getValue() + " of type " + foo.getClass()); + return new GenericMessage(foo); + } + }; + } + + public interface Source { + String SAMPLE = "sample-source"; + + @Output(SAMPLE) + MessageChannel sampleSource(); + } +} diff --git a/stream-listener/src/main/java/demo/SampleTransformer.java b/stream-listener/src/main/java/demo/SampleTransformer.java new file mode 100644 index 0000000..faab174 --- /dev/null +++ b/stream-listener/src/main/java/demo/SampleTransformer.java @@ -0,0 +1,45 @@ +/* + * 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.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.messaging.Processor; +import org.springframework.messaging.handler.annotation.SendTo; + +/** + * @author Ilayaperumal Gopinathan + */ +@EnableBinding(Processor.class) +public class SampleTransformer { + + private static final String TRANSFORMATION_VALUE = "HI"; + + // Transformer application definition + + @StreamListener(Processor.INPUT) + @SendTo(Processor.OUTPUT) + public Converters.Bar receive(Converters.Bar barMessage) { + System.out.println("******************"); + System.out.println("At the transformer"); + System.out.println("******************"); + System.out.println("Received value "+ barMessage.getValue() + " of type " + barMessage.getClass()); + System.out.println("Transforming the value to " + TRANSFORMATION_VALUE + " and with the type " + barMessage.getClass()); + barMessage.setValue(TRANSFORMATION_VALUE); + return barMessage; + } +} diff --git a/stream-listener/src/main/java/demo/TypeConversionApplication.java b/stream-listener/src/main/java/demo/TypeConversionApplication.java new file mode 100644 index 0000000..d3cf00f --- /dev/null +++ b/stream-listener/src/main/java/demo/TypeConversionApplication.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 TypeConversionApplication { + + public static void main(String[] args) { + SpringApplication.run(TypeConversionApplication.class, args); + } + +} diff --git a/stream-listener/src/main/resources/application.yml b/stream-listener/src/main/resources/application.yml new file mode 100644 index 0000000..9f6e482 --- /dev/null +++ b/stream-listener/src/main/resources/application.yml @@ -0,0 +1,14 @@ +server: + port: 8082 +spring: + cloud: + stream: + bindings: + sample-source: + destination: testtock + input: + destination: testtock + output: + destination: xformed + sample-sink: + destination: xformed diff --git a/stream-listener/src/test/java/demo/ModuleApplicationTests.java b/stream-listener/src/test/java/demo/ModuleApplicationTests.java new file mode 100644 index 0000000..50b3bcf --- /dev/null +++ b/stream-listener/src/test/java/demo/ModuleApplicationTests.java @@ -0,0 +1,36 @@ +/* + * 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.test.annotation.DirtiesContext; +import org.springframework.test.context.web.WebAppConfiguration; +import org.springframework.boot.test.SpringApplicationConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +@RunWith(SpringJUnit4ClassRunner.class) +@SpringApplicationConfiguration(classes = TypeConversionApplication.class) +@WebAppConfiguration +@DirtiesContext +public class ModuleApplicationTests { + + @Test + public void contextLoads() { + } + +}