diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAsMetaAnnotationTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAsMetaAnnotationTests.java new file mode 100644 index 000000000..712f18cc8 --- /dev/null +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAsMetaAnnotationTests.java @@ -0,0 +1,121 @@ +package org.springframework.cloud.stream.config; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.lang.annotation.Documented; +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; + +import org.junit.Test; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.messaging.Sink; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.core.annotation.AliasFor; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.handler.annotation.MessageMapping; +import org.springframework.messaging.handler.annotation.Payload; + + +public class StreamListenerAsMetaAnnotationTests { + + @Test + public void testCustomAnnotation() { + ConfigurableApplicationContext context = SpringApplication.run(TestPojoWithCustomAnnotatedArguments.class, + "--server.port=0"); + + TestPojoWithCustomAnnotatedArguments testPojoWithAnnotatedArguments = context + .getBean(TestPojoWithCustomAnnotatedArguments.class); + Sink sink = context.getBean(Sink.class); + String id = UUID.randomUUID().toString(); + sink.input().send(MessageBuilder.withPayload("{\"foo\":\"barbar" + id + "\"}") + .setHeader("contentType", "application/json").setHeader("testHeader", "testValue") + .setHeader("type", "foo").build()); + assertThat(testPojoWithAnnotatedArguments.receivedFoo).hasSize(1); + assertThat(testPojoWithAnnotatedArguments.receivedFoo.get(0)).hasFieldOrPropertyWithValue("foo", + "barbar" + id); + context.close(); + } + + @Test + public void testAnnotation() { + ConfigurableApplicationContext context = SpringApplication.run(TestPojoWithAnnotatedArguments.class, + "--server.port=0"); + + TestPojoWithAnnotatedArguments testPojoWithAnnotatedArguments = context + .getBean(TestPojoWithAnnotatedArguments.class); + Sink sink = context.getBean(Sink.class); + String id = UUID.randomUUID().toString(); + sink.input().send(MessageBuilder.withPayload("{\"foo\":\"barbar" + id + "\"}") + .setHeader("contentType", "application/json").setHeader("testHeader", "testValue") + .setHeader("type", "foo").build()); + assertThat(testPojoWithAnnotatedArguments.receivedFoo).hasSize(1); + assertThat(testPojoWithAnnotatedArguments.receivedFoo.get(0)).hasFieldOrPropertyWithValue("foo", + "barbar" + id); + context.close(); + } + + @EnableBinding(Sink.class) + @EnableAutoConfiguration + public static class TestPojoWithCustomAnnotatedArguments { + + List receivedFoo = new ArrayList<>(); + + List receivedBar = new ArrayList<>(); + + @EventHandler(value = Sink.INPUT, condition = "headers['type']=='foo'") + public void receive(@Payload StreamListenerTestUtils.FooPojo fooPojo) { + this.receivedFoo.add(fooPojo); + } + } + + @EnableBinding(Sink.class) + @EnableAutoConfiguration + public static class TestPojoWithAnnotatedArguments { + + List receivedFoo = new ArrayList<>(); + + List receivedBar = new ArrayList<>(); + + @StreamListener(value = Sink.INPUT, condition = "headers['type']=='foo'") + public void receive(@Payload StreamListenerTestUtils.FooPojo fooPojo) { + this.receivedFoo.add(fooPojo); + } + } +} + +@Target({ ElementType.METHOD, ElementType.ANNOTATION_TYPE }) +@Retention(RetentionPolicy.RUNTIME) +@MessageMapping +@Documented +@StreamListener +@interface EventHandler { + /** + * The name of the binding target (e.g. channel) that the method subscribes to. + * @return the name of the binding target. + */ + @AliasFor(annotation=StreamListener.class, attribute="target") + String value() default ""; + + /** + * The name of the binding target (e.g. channel) that the method subscribes to. + * @return the name of the binding target. + */ + @AliasFor(annotation=StreamListener.class, attribute="target") + String target() default ""; + + /** + * A condition that must be met by all items that are dispatched to this method. + * @return a SpEL expression that must evaluate to a {@code boolean} value. + */ + @AliasFor(annotation=StreamListener.class, attribute="condition") + String condition() default ""; + +} \ No newline at end of file