Align StreamListener method/parameter mappings

- Use the same mapping of method/parameters for both declarative and message handler Stream Listeners
       - Declarative mode of the StreamListener method is determined if at least one of the the method parameters is annotated with @Input or @Output with either bound elements (e.g. channels) or conversion targets from bound elements via a registered StreamListenerParameterAdapter
       - If the method is non-declarative then it is considered to be in message handler mode
       - Declarative mode now accepts @Output annotation at method level as well along with @SendTo
       - Declarative mode also accepts @Input annotation value at the method level via @StreamListener valuue.
       - Message handler mode accepts @Input annotation value at the method parameter level and @Output annotation at the method annotation level
       - Both @Output and @SendTo annotations are supported to specify the outbound target value while @SendTo is allowed only as a method level annotation
       - Add assertions on allowable use of method and parameter annotations

    - Add parameterized tests to cover all possible cases
        - Message handler tests
        - Reactor and RxJava tests

Resolves #664

Add more error handling and simplify the usage patterns

Address review comments

 - Support @Input/@Output only for declarative StreamListener methods
 - Support multiple @Output only when there is no return type in the StreamListener method
 - Add/Update tests

WIP

Refactor PR based on the review comments
This commit is contained in:
Ilayaperumal Gopinathan
2016-10-07 22:56:13 +05:30
committed by Marius Bogoevici
parent 8595d094c9
commit 453a9b0c10
34 changed files with 3388 additions and 1057 deletions

View File

@@ -0,0 +1,117 @@
/*
* 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 org.springframework.cloud.stream.config;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import org.junit.Test;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
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.messaging.Processor;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.fail;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_MESSAGE_HANDLER_METHOD_PARAMS;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
public class StreamListenerAnnotatedMethodArgumentsTests {
@Test
@SuppressWarnings("unchecked")
public void testAnnotatedArguments() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestPojoWithAnnotatedArguments1.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").build());
assertThat(testPojoWithAnnotatedArguments.receivedArguments).hasSize(3);
assertThat(testPojoWithAnnotatedArguments.receivedArguments.get(0)).isInstanceOf(StreamListenerTestInterfaces.FooPojo.class);
assertThat(testPojoWithAnnotatedArguments.receivedArguments.get(0)).hasFieldOrPropertyWithValue("foo",
"barbar" + id);
assertThat(testPojoWithAnnotatedArguments.receivedArguments.get(1)).isInstanceOf(Map.class);
assertThat((Map<String, String>) testPojoWithAnnotatedArguments.receivedArguments.get(1))
.containsEntry(MessageHeaders.CONTENT_TYPE, "application/json");
assertThat((Map<String, String>) testPojoWithAnnotatedArguments.receivedArguments.get(1))
.containsEntry("testHeader", "testValue");
assertThat(testPojoWithAnnotatedArguments.receivedArguments.get(2)).isEqualTo("application/json");
context.close();
}
@Test
public void testInputAnnotationAtMethodParameter() throws Exception {
try {
SpringApplication.run(TestPojoWithAnnotatedArguments2.class, "--server.port=0");
fail("Exception expected: "+ INVALID_MESSAGE_HANDLER_METHOD_PARAMS);
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(INVALID_MESSAGE_HANDLER_METHOD_PARAMS);
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestPojoWithAnnotatedArguments1 extends TestPojoWithAnnotatedArguments {
@StreamListener(Processor.INPUT)
public void receive(@Payload StreamListenerTestInterfaces.FooPojo fooPojo,
@Headers Map<String, Object> headers,
@Header(MessageHeaders.CONTENT_TYPE) String contentType) {
this.receivedArguments.add(fooPojo);
this.receivedArguments.add(headers);
this.receivedArguments.add(contentType);
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestPojoWithAnnotatedArguments2 extends TestPojoWithAnnotatedArguments {
@StreamListener
public void receive(@Input(Processor.INPUT) @Payload StreamListenerTestInterfaces.FooPojo fooPojo,
@Headers Map<String, Object> headers,
@Header(MessageHeaders.CONTENT_TYPE) String contentType) {
this.receivedArguments.add(fooPojo);
this.receivedArguments.add(headers);
this.receivedArguments.add(contentType);
}
}
public static class TestPojoWithAnnotatedArguments {
List<Object> receivedArguments = new ArrayList<>();
}
}

View File

@@ -0,0 +1,74 @@
/*
* 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 org.springframework.cloud.stream.config;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
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.integration.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
public class StreamListenerContentTypeConversionTests {
@Test
public void testContentTypeConversion() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestSink1.class,
"--server.port=0");
@SuppressWarnings("unchecked")
TestSink1 testSink = context.getBean(TestSink1.class);
Sink sink = context.getBean(Sink.class);
String id = UUID.randomUUID().toString();
sink.input().send(
MessageBuilder.withPayload("{\"foo\":\"barbar" + id + "\"}")
.setHeader("contentType", "application/json").build());
assertThat(testSink.latch.await(10, TimeUnit.SECONDS));
assertThat(testSink.receivedArguments).hasSize(1);
assertThat(testSink.receivedArguments.get(0)).hasFieldOrPropertyWithValue("foo",
"barbar" + id);
context.close();
}
@EnableBinding(Sink.class)
@EnableAutoConfiguration
public static class TestSink1 {
List<StreamListenerTestInterfaces.FooPojo> receivedArguments = new ArrayList<>();
CountDownLatch latch = new CountDownLatch(1);
@StreamListener(Sink.INPUT)
public void receive(StreamListenerTestInterfaces.FooPojo fooPojo) {
this.receivedArguments.add(fooPojo);
this.latch.countDown();
}
}
}

View File

@@ -0,0 +1,63 @@
/*
* 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 org.springframework.cloud.stream.config;
import org.junit.Test;
import org.springframework.beans.factory.BeanCreationException;
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.Processor;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.fail;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
public class StreamListenerDuplicateMappingTests {
@Test
@SuppressWarnings("unchecked")
public void testDuplicateMapping() throws Exception {
try {
ConfigurableApplicationContext context = SpringApplication.run(TestDuplicateMapping1.class,
"--server.port=0");
fail("Exception expected on duplicate mapping");
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).startsWith("Duplicate @StreamListener mapping");
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestDuplicateMapping1 {
@StreamListener(Processor.INPUT)
public void receive(Message<String> fooMessage) {
}
@StreamListener(Processor.INPUT)
public void receive2(Message<String> fooMessage) {
}
}
}

View File

@@ -0,0 +1,140 @@
/*
* 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 org.springframework.cloud.stream.config;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
@RunWith(Parameterized.class)
public class StreamListenerHandlerBeanTests {
private Class<?> configClass;
public StreamListenerHandlerBeanTests(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection InputConfigs() {
return Arrays.asList(new Class[] { TestHandlerBean1.class, TestHandlerBean2.class });
}
@Test
@SuppressWarnings("unchecked")
public void testHandlerBean() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(this.configClass,
"--spring.cloud.stream.bindings.output.contentType=application/json",
"--server.port=0");
MessageCollector collector = context.getBean(MessageCollector.class);
Processor processor = context.getBean(Processor.class);
String id = UUID.randomUUID().toString();
processor.input().send(
MessageBuilder.withPayload("{\"foo\":\"barbar" + id + "\"}")
.setHeader("contentType", "application/json").build());
HandlerBean handlerBean = context.getBean(HandlerBean.class);
assertThat(handlerBean.receivedPojos).hasSize(1);
assertThat(handlerBean.receivedPojos.get(0)).hasFieldOrPropertyWithValue("foo",
"barbar" + id);
Message<String> message = (Message<String>) collector.forChannel(
processor.output()).poll(1, TimeUnit.SECONDS);
assertThat(message).isNotNull();
assertThat(message.getPayload()).isEqualTo("{\"bar\":\"barbar" + id + "\"}");
assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE, MimeType.class)
.includes(MimeTypeUtils.APPLICATION_JSON));
context.close();
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestHandlerBean1 {
@Bean
public HandlerBean1 handlerBean() {
return new HandlerBean1();
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestHandlerBean2 {
@Bean
public HandlerBean2 handlerBean() {
return new HandlerBean2();
}
}
public static class HandlerBean1 extends HandlerBean {
@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public StreamListenerTestInterfaces.BarPojo receive(StreamListenerTestInterfaces.FooPojo fooMessage) {
this.receivedPojos.add(fooMessage);
StreamListenerTestInterfaces.BarPojo barPojo = new StreamListenerTestInterfaces.BarPojo();
barPojo.setBar(fooMessage.getFoo());
return barPojo;
}
}
public static class HandlerBean2 extends HandlerBean {
@StreamListener(Processor.INPUT)
@Output(Processor.OUTPUT)
public StreamListenerTestInterfaces.BarPojo receive(StreamListenerTestInterfaces.FooPojo fooMessage) {
this.receivedPojos.add(fooMessage);
StreamListenerTestInterfaces.BarPojo barPojo = new StreamListenerTestInterfaces.BarPojo();
barPojo.setBar(fooMessage.getFoo());
return barPojo;
}
}
public static class HandlerBean {
List<StreamListenerTestInterfaces.FooPojo> receivedPojos = new ArrayList<>();
}
}

View File

@@ -0,0 +1,407 @@
/*
* 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 org.springframework.cloud.stream.config;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.util.Assert;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.fail;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.AMBIGUOUS_MESSAGE_HANDLER_METHOD_ARGUMENTS;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INPUT_AT_STREAM_LISTENER;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_INBOUND_NAME;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_MESSAGE_HANDLER_METHOD_PARAMS;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_OUTBOUND_NAME;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_OUTPUT_VALUES;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.NO_INPUT_DESTINATION;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.RETURN_TYPE_MULTIPLE_OUTBOUND_SPECIFIED;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.RETURN_TYPE_NO_OUTBOUND_SPECIFIED;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.TARGET_BEAN_NOT_EXISTS;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
public class StreamListenerHandlerMethodTests {
@Test
public void testMethodWIthInputOnStreamListener() throws Exception {
try {
SpringApplication.run(TestMethodWIthInputOnStreamListener.class, "--server.port=0");
fail("Exception expected: "+ INPUT_AT_STREAM_LISTENER);
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(INPUT_AT_STREAM_LISTENER);
}
}
@Test
public void testReturnTypeWithMultipleOutput() throws Exception {
try {
SpringApplication.run(TestReturnTypeWithMultipleOutput.class, "--server.port=0");
fail("Exception expected: "+ RETURN_TYPE_MULTIPLE_OUTBOUND_SPECIFIED);
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(RETURN_TYPE_MULTIPLE_OUTBOUND_SPECIFIED);
}
}
@Test
public void testReturnTypeWithNoOutput() throws Exception {
try {
SpringApplication.run(TestReturnTypeWithNoOutput.class, "--server.port=0");
fail("Exception expected: " + RETURN_TYPE_NO_OUTBOUND_SPECIFIED);
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(RETURN_TYPE_NO_OUTBOUND_SPECIFIED);
}
}
@Test
public void testMethodInputAnnotationWithNoValue() throws Exception {
try {
SpringApplication.run(TestMethodInputAnnotationWithNoValue.class, "--server.port=0");
fail("Exception expected: "+ INVALID_INBOUND_NAME);
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(INVALID_INBOUND_NAME);
}
}
@Test
public void testMethodOutputAnnotationWithNoValue() throws Exception {
try {
SpringApplication.run(TestMethodOutputAnnotationWithNoValue.class, "--server.port=0");
fail("Exception expected: "+ INVALID_OUTBOUND_NAME);
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(INVALID_OUTBOUND_NAME);
}
}
@Test
public void testMethodInvalidInboundName() throws Exception {
try {
SpringApplication.run(TestMethodInvalidInboundName.class, "--server.port=0");
fail("Exception expected on using invalid inbound name");
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(TARGET_BEAN_NOT_EXISTS + ": invalid");
}
}
@Test
public void testMethodInvalidOutboundName() throws Exception {
try {
SpringApplication.run(TestMethodInvalidOutboundName.class, "--server.port=0");
fail("Exception expected on using invalid outbound name");
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(TARGET_BEAN_NOT_EXISTS + ": invalid");
}
}
@Test
public void testAmbiguousMethodArguments1() throws Exception {
try {
SpringApplication.run(TestAmbiguousMethodArguments1.class, "--server.port=0");
fail("Exception expected: "+ AMBIGUOUS_MESSAGE_HANDLER_METHOD_ARGUMENTS);
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(AMBIGUOUS_MESSAGE_HANDLER_METHOD_ARGUMENTS);
}
}
@Test
public void testAmbiguousMethodArguments2() throws Exception {
try {
SpringApplication.run(TestAmbiguousMethodArguments2.class, "--server.port=0");
fail("Exception expected:" + AMBIGUOUS_MESSAGE_HANDLER_METHOD_ARGUMENTS);
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(AMBIGUOUS_MESSAGE_HANDLER_METHOD_ARGUMENTS);
}
}
@Test
public void testMethodWithInputAsMethodAndParameter() throws Exception {
try {
SpringApplication.run(TestMethodWithInputAsMethodAndParameter.class, "--server.port=0");
fail("Exception expected: " + INVALID_MESSAGE_HANDLER_METHOD_PARAMS);
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(INVALID_MESSAGE_HANDLER_METHOD_PARAMS);
}
}
@Test
public void testMethodWithOutputAsMethodAndParameter() throws Exception {
try {
SpringApplication.run(TestMethodWithOutputAsMethodAndParameter.class, "--server.port=0");
fail("Exception expected:" + INVALID_OUTPUT_VALUES);
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).startsWith(INVALID_OUTPUT_VALUES);
}
}
@Test
public void testMethodWithoutInput() throws Exception {
try {
SpringApplication.run(TestMethodWithoutInput.class, "--server.port=0");
fail("Exception expected when inbound target is not set");
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(NO_INPUT_DESTINATION);
}
}
@Test
public void testMethodWithMultipleInputParameters() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestMethodWithMultipleInputParameters.class, "--server.port=0");
Processor processor = context.getBean(Processor.class);
StreamListenerTestInterfaces.FooInboundChannel1 inboundChannel2 = context.getBean(StreamListenerTestInterfaces.FooInboundChannel1.class);
String id = UUID.randomUUID().toString();
final CountDownLatch latch = new CountDownLatch(2);
((SubscribableChannel) processor.output()).subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
Assert.isTrue(message.getPayload().equals("footesting") || message.getPayload().equals("BARTESTING"));
latch.countDown();
}
});
processor.input().send(MessageBuilder.withPayload("{\"foo\":\"fooTESTing" + "\"}")
.setHeader("contentType", "application/json").build());
inboundChannel2.input().send(MessageBuilder.withPayload("{\"bar\":\"bartestING" + "\"}")
.setHeader("contentType", "application/json").build());
assertThat(latch.await(1, TimeUnit.SECONDS));
context.close();
}
@Test
public void testMethodWithMultipleOutputParameters() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestMethodWithMultipleOutputParameters.class, "--server.port=0");
Processor processor = context.getBean(Processor.class);
String id = UUID.randomUUID().toString();
StreamListenerTestInterfaces.FooOutboundChannel1 source2 = context.getBean(StreamListenerTestInterfaces.FooOutboundChannel1.class);
final CountDownLatch latch = new CountDownLatch(2);
((SubscribableChannel) processor.output()).subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
Assert.isTrue(message.getPayload().equals("testing"));
Assert.isTrue(message.getHeaders().get("output").equals("output2"));
latch.countDown();
}
});
((SubscribableChannel) source2.output()).subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
Assert.isTrue(message.getPayload().equals("TESTING"));
Assert.isTrue(message.getHeaders().get("output").equals("output1"));
latch.countDown();
}
});
processor.input().send(MessageBuilder.withPayload("testING").setHeader("output", "output1").build());
processor.input().send(MessageBuilder.withPayload("TESTing").setHeader("output", "output2").build());
assertThat(latch.await(1, TimeUnit.SECONDS));
context.close();
}
@EnableBinding({Processor.class, StreamListenerTestInterfaces.FooOutboundChannel1.class})
@EnableAutoConfiguration
public static class TestMethodWithMultipleOutputParameters {
@StreamListener
public void receive(@Input(Processor.INPUT) SubscribableChannel input, @Output(Processor.OUTPUT) final MessageChannel output1,
@Output(StreamListenerTestInterfaces.FooOutboundChannel1.OUTPUT) final MessageChannel output2) {
input.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
if (message.getHeaders().get("output").equals("output1")) {
output1.send(org.springframework.messaging.support.MessageBuilder.withPayload(message.getPayload().toString().toUpperCase()).build());
}
else if (message.getHeaders().get("output").equals("output2")) {
output2.send(org.springframework.messaging.support.MessageBuilder.withPayload(message.getPayload().toString().toLowerCase()).build());
}
}
});
}
}
@EnableBinding({Sink.class})
@EnableAutoConfiguration
public static class TestMethodWithoutInput {
@StreamListener
public void receive(StreamListenerTestInterfaces.FooPojo fooPojo) {
}
}
@EnableBinding({Sink.class})
@EnableAutoConfiguration
public static class TestMethodWIthInputOnStreamListener {
@StreamListener
@Input(Sink.INPUT)
public void receive(StreamListenerTestInterfaces.FooPojo fooPojo) {
}
}
@EnableBinding({Sink.class})
@EnableAutoConfiguration
public static class TestAmbiguousMethodArguments1 {
@StreamListener(Processor.INPUT)
public void receive(@Payload StreamListenerTestInterfaces.FooPojo fooPojo, String value) {
}
}
@EnableBinding({Sink.class})
@EnableAutoConfiguration
public static class TestAmbiguousMethodArguments2 {
@StreamListener(Processor.INPUT)
public void receive(@Payload StreamListenerTestInterfaces.FooPojo fooPojo, @Payload StreamListenerTestInterfaces.BarPojo barPojo) {
}
}
@EnableBinding({Processor.class, StreamListenerTestInterfaces.FooOutboundChannel1.class})
@EnableAutoConfiguration
public static class TestReturnTypeWithMultipleOutput {
@StreamListener
public String receive(@Input(Processor.INPUT) SubscribableChannel input1, @Output(Processor.OUTPUT) MessageChannel output1,
@Output(StreamListenerTestInterfaces.FooOutboundChannel1.OUTPUT) MessageChannel output2) {
return "foo";
}
}
@EnableBinding({Processor.class, StreamListenerTestInterfaces.FooOutboundChannel1.class})
@EnableAutoConfiguration
public static class TestReturnTypeWithNoOutput {
@StreamListener
public String receive(@Input(Processor.INPUT) SubscribableChannel input1) {
return "foo";
}
}
@EnableBinding({Processor.class})
@EnableAutoConfiguration
public static class TestMethodInputAnnotationWithNoValue {
@StreamListener
public void receive(@Input SubscribableChannel input) {
}
}
@EnableBinding({Processor.class})
@EnableAutoConfiguration
public static class TestMethodOutputAnnotationWithNoValue {
@StreamListener
public void receive(@Input(Processor.OUTPUT) SubscribableChannel input, @Output MessageChannel output) {
}
}
@EnableBinding({Sink.class})
@EnableAutoConfiguration
public static class TestMethodInvalidInboundName {
@StreamListener
public void receive(@Input("invalid") SubscribableChannel input) {
}
}
@EnableBinding({Processor.class})
@EnableAutoConfiguration
public static class TestMethodInvalidOutboundName {
@StreamListener
public void receive(@Input(Processor.INPUT) SubscribableChannel input, @Output("invalid") MessageChannel output) {
}
}
@EnableBinding({Sink.class})
@EnableAutoConfiguration
public static class TestMethodWithInputAsMethodAndParameter {
@StreamListener
public void receive(@Input(Sink.INPUT) StreamListenerTestInterfaces.FooPojo fooPojo) {
}
}
@EnableBinding({Processor.class, StreamListenerTestInterfaces.FooOutboundChannel1.class})
@EnableAutoConfiguration
public static class TestMethodWithOutputAsMethodAndParameter {
@StreamListener
@Output(StreamListenerTestInterfaces.FooOutboundChannel1.OUTPUT)
public void receive(@Input(Processor.INPUT) SubscribableChannel input, @Output(Processor.OUTPUT) final MessageChannel output1) {
input.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
output1.send(org.springframework.messaging.support.MessageBuilder.withPayload(message.getPayload().toString().toUpperCase()).build());
}
});
}
}
@EnableBinding({Processor.class, StreamListenerTestInterfaces.FooInboundChannel1.class})
@EnableAutoConfiguration
public static class TestMethodWithMultipleInputParameters {
@StreamListener
public void receive(@Input(Processor.INPUT) SubscribableChannel input1, @Input(StreamListenerTestInterfaces.FooInboundChannel1.INPUT) SubscribableChannel input2,
final @Output(Processor.OUTPUT) MessageChannel output) {
input1.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
output.send(org.springframework.messaging.support.MessageBuilder.withPayload(message.getPayload().toString().toUpperCase()).build());
}
});
input2.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
output.send(org.springframework.messaging.support.MessageBuilder.withPayload(message.getPayload().toString().toUpperCase()).build());
}
});
}
}
}

View File

@@ -0,0 +1,114 @@
/*
* 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 org.springframework.cloud.stream.config;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.SendTo;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
@RunWith(Parameterized.class)
public class StreamListenerMessageArgumentTests {
private Class<?> configClass;
public StreamListenerMessageArgumentTests(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection InputConfigs() {
return Arrays.asList(new Class[] {TestPojoWithMessageArgument1.class, TestPojoWithMessageArgument2.class});
}
@Test
@SuppressWarnings("unchecked")
public void testMessageArgument() throws Exception {
ConfigurableApplicationContext context = SpringApplication
.run(this.configClass, "--server.port=0");
MessageCollector collector = context.getBean(MessageCollector.class);
Processor processor = context.getBean(Processor.class);
String id = UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload("barbar" + id)
.setHeader("contentType", "text/plain").build());
TestPojoWithMessageArgument testPojoWithMessageArgument = context
.getBean(TestPojoWithMessageArgument.class);
assertThat(testPojoWithMessageArgument.receivedMessages).hasSize(1);
assertThat(testPojoWithMessageArgument.receivedMessages.get(0).getPayload()).isEqualTo("barbar" + id);
Message<StreamListenerTestInterfaces.BarPojo> message = (Message<StreamListenerTestInterfaces.BarPojo>) collector
.forChannel(processor.output()).poll(1, TimeUnit.SECONDS);
assertThat(message).isNotNull();
assertThat(message.getPayload().getBar()).isEqualTo("barbar" + id);
context.close();
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestPojoWithMessageArgument1 extends TestPojoWithMessageArgument {
@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public StreamListenerTestInterfaces.BarPojo receive(Message<String> fooMessage) {
this.receivedMessages.add(fooMessage);
StreamListenerTestInterfaces.BarPojo barPojo = new StreamListenerTestInterfaces.BarPojo();
barPojo.setBar(fooMessage.getPayload());
return barPojo;
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestPojoWithMessageArgument2 extends TestPojoWithMessageArgument {
@StreamListener(Processor.INPUT)
@Output(Processor.OUTPUT)
public StreamListenerTestInterfaces.BarPojo receive(Message<String> fooMessage) {
this.receivedMessages.add(fooMessage);
StreamListenerTestInterfaces.BarPojo barPojo = new StreamListenerTestInterfaces.BarPojo();
barPojo.setBar(fooMessage.getPayload());
return barPojo;
}
}
public static class TestPojoWithMessageArgument {
List<Message<String>> receivedMessages = new ArrayList<>();
}
}

View File

@@ -0,0 +1,164 @@
/*
* 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 org.springframework.cloud.stream.config;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Suite;
import org.junit.runners.model.InitializationError;
import org.junit.runners.model.RunnerBuilder;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
@RunWith(StreamListenerMethodReturnWithConversionTests.class)
@Suite.SuiteClasses({StreamListenerMethodReturnWithConversionTests.TestReturnConversion.class,
StreamListenerMethodReturnWithConversionTests.TestReturnNoConversion.class})
public class StreamListenerMethodReturnWithConversionTests extends Suite {
public StreamListenerMethodReturnWithConversionTests(Class<?> klass, RunnerBuilder builder) throws InitializationError {
super(klass, builder);
}
@RunWith(Parameterized.class)
public static class TestReturnConversion {
private Class<?> configClass;
public TestReturnConversion(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection InputConfigs() {
return Arrays.asList(new Class[] {TestPojoWithMimeType1.class, TestPojoWithMimeType2.class});
}
@Test
@SuppressWarnings("unchecked")
public void testReturnConversion() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(this.configClass,
"--spring.cloud.stream.bindings.output.contentType=application/json", "--server.port=0");
MessageCollector collector = context.getBean(MessageCollector.class);
Processor processor = context.getBean(Processor.class);
String id = UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload("{\"foo\":\"barbar" + id + "\"}")
.setHeader("contentType", "application/json").build());
TestPojoWithMimeType testPojoWithMimeType = context.getBean(TestPojoWithMimeType.class);
assertThat(testPojoWithMimeType.receivedPojos).hasSize(1);
assertThat(testPojoWithMimeType.receivedPojos.get(0)).hasFieldOrPropertyWithValue("foo", "barbar" + id);
Message<String> message = (Message<String>) collector.forChannel(processor.output()).poll(1, TimeUnit.SECONDS);
assertThat(message).isNotNull();
assertThat(message.getPayload()).isEqualTo("{\"bar\":\"barbar" + id + "\"}");
assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE, MimeType.class).includes(MimeTypeUtils.APPLICATION_JSON));
context.close();
}
}
@RunWith(Parameterized.class)
public static class TestReturnNoConversion {
private Class<?> configClass;
public TestReturnNoConversion(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection InputConfigs() {
return Arrays.asList(new Class[] {TestPojoWithMimeType1.class, TestPojoWithMimeType2.class});
}
@Test
@SuppressWarnings("unchecked")
public void testReturnNoConversion() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(this.configClass, "--server.port=0");
MessageCollector collector = context.getBean(MessageCollector.class);
Processor processor = context.getBean(Processor.class);
String id = UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload("{\"foo\":\"barbar" + id + "\"}")
.setHeader("contentType", "application/json").build());
TestPojoWithMimeType testPojoWithMimeType = context.getBean(TestPojoWithMimeType.class);
assertThat(testPojoWithMimeType.receivedPojos).hasSize(1);
assertThat(testPojoWithMimeType.receivedPojos.get(0)).hasFieldOrPropertyWithValue("foo", "barbar" + id);
Message<StreamListenerTestInterfaces.BarPojo> message = (Message<StreamListenerTestInterfaces.BarPojo>) collector.forChannel(processor.output()).poll(1,
TimeUnit.SECONDS);
assertThat(message).isNotNull();
assertThat(message.getPayload().getBar()).isEqualTo("barbar" + id);
assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE, MimeType.class) == null);
context.close();
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestPojoWithMimeType1 extends TestPojoWithMimeType {
@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public StreamListenerTestInterfaces.BarPojo receive(StreamListenerTestInterfaces.FooPojo fooPojo) {
this.receivedPojos.add(fooPojo);
StreamListenerTestInterfaces.BarPojo barPojo = new StreamListenerTestInterfaces.BarPojo();
barPojo.setBar(fooPojo.getFoo());
return barPojo;
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestPojoWithMimeType2 extends TestPojoWithMimeType {
@StreamListener(Processor.INPUT)
@Output(Processor.OUTPUT)
public StreamListenerTestInterfaces.BarPojo receive(StreamListenerTestInterfaces.FooPojo fooPojo) {
this.receivedPojos.add(fooPojo);
StreamListenerTestInterfaces.BarPojo barPojo = new StreamListenerTestInterfaces.BarPojo();
barPojo.setBar(fooPojo.getFoo());
return barPojo;
}
}
public static class TestPojoWithMimeType {
List<StreamListenerTestInterfaces.FooPojo> receivedPojos = new ArrayList<>();
}
}

View File

@@ -0,0 +1,115 @@
/*
* 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 org.springframework.cloud.stream.config;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.SendTo;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
@RunWith(Parameterized.class)
public class StreamListenerMethodWithReturnMessageTests {
private Class<?> configClass;
public StreamListenerMethodWithReturnMessageTests(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection InputConfigs() {
return Arrays.asList(new Class[]{TestPojoWithMessageReturn1.class, TestPojoWithMessageReturn2.class});
}
@Test
@SuppressWarnings("unchecked")
public void testReturnMessage() throws Exception {
ConfigurableApplicationContext context = SpringApplication
.run(this.configClass, "--server.port=0");
MessageCollector collector = context.getBean(MessageCollector.class);
Processor processor = context.getBean(Processor.class);
String id = UUID.randomUUID().toString();
processor.input()
.send(MessageBuilder.withPayload("{\"foo\":\"barbar" + id + "\"}")
.setHeader("contentType", "application/json").build());
TestPojoWithMessageReturn testPojoWithMessageReturn = context
.getBean(TestPojoWithMessageReturn.class);
assertThat(testPojoWithMessageReturn.receivedPojos).hasSize(1);
assertThat(testPojoWithMessageReturn.receivedPojos.get(0)).hasFieldOrPropertyWithValue("foo", "barbar" + id);
Message<StreamListenerTestInterfaces.BarPojo> message = (Message<StreamListenerTestInterfaces.BarPojo>) collector
.forChannel(processor.output()).poll(1, TimeUnit.SECONDS);
assertThat(message).isNotNull();
assertThat(message.getPayload().getBar()).isEqualTo("barbar" + id);
context.close();
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestPojoWithMessageReturn1 extends TestPojoWithMessageReturn {
@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public Message<?> receive(StreamListenerTestInterfaces.FooPojo fooPojo) {
this.receivedPojos.add(fooPojo);
StreamListenerTestInterfaces.BarPojo barPojo = new StreamListenerTestInterfaces.BarPojo();
barPojo.setBar(fooPojo.getFoo());
return MessageBuilder.withPayload(barPojo).setHeader("foo", "bar").build();
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestPojoWithMessageReturn2 extends TestPojoWithMessageReturn {
@StreamListener(Processor.INPUT)
@Output(Processor.OUTPUT)
public Message<?> receive(StreamListenerTestInterfaces.FooPojo fooPojo) {
this.receivedPojos.add(fooPojo);
StreamListenerTestInterfaces.BarPojo bazPojo = new StreamListenerTestInterfaces.BarPojo();
bazPojo.setBar(fooPojo.getFoo());
return MessageBuilder.withPayload(bazPojo).setHeader("foo", "bar").build();
}
}
public static class TestPojoWithMessageReturn {
List<StreamListenerTestInterfaces.FooPojo> receivedPojos = new ArrayList<>();
}
}

View File

@@ -0,0 +1,111 @@
/*
* 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 org.springframework.cloud.stream.config;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.SendTo;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
@RunWith(Parameterized.class)
public class StreamListenerMethodWithReturnValueTests {
private Class<?> configClass;
public StreamListenerMethodWithReturnValueTests(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection InputConfigs() {
return Arrays.asList(new Class[]{TestStringProcessor1.class, TestStringProcessor2.class});
}
@Test
@SuppressWarnings("unchecked")
public void testReturn() throws Exception {
ConfigurableApplicationContext context = SpringApplication
.run(this.configClass, "--server.port=0");
MessageCollector collector = context.getBean(MessageCollector.class);
Processor processor = context.getBean(Processor.class);
String id = UUID.randomUUID().toString();
processor.input()
.send(MessageBuilder.withPayload("{\"foo\":\"barbar" + id + "\"}")
.setHeader("contentType", "application/json").build());
Message<String> message = (Message<String>) collector
.forChannel(processor.output()).poll(1, TimeUnit.SECONDS);
TestStringProcessor testStringProcessor = context
.getBean(TestStringProcessor.class);
assertThat(testStringProcessor.receivedPojos).hasSize(1);
assertThat(testStringProcessor.receivedPojos.get(0)).hasFieldOrPropertyWithValue("foo", "barbar" + id);
assertThat(message).isNotNull();
assertThat(message.getPayload()).isEqualTo("barbar" + id);
context.close();
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestStringProcessor1 extends TestStringProcessor {
@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public String receive(StreamListenerTestInterfaces.FooPojo fooPojo) {
this.receivedPojos.add(fooPojo);
return fooPojo.getFoo();
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestStringProcessor2 extends TestStringProcessor {
@StreamListener(Processor.INPUT)
@Output(Processor.OUTPUT)
public String receive(StreamListenerTestInterfaces.FooPojo fooPojo) {
this.receivedPojos.add(fooPojo);
return fooPojo.getFoo();
}
}
public static class TestStringProcessor {
List<StreamListenerTestInterfaces.FooPojo> receivedPojos = new ArrayList<>();
}
}

View File

@@ -0,0 +1,71 @@
/*
* 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 org.springframework.cloud.stream.config;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
/**
* @author Ilayaperumal Gopinathan
*/
public class StreamListenerTestInterfaces {
public static class FooPojo {
private String foo;
public String getFoo() {
return this.foo;
}
public void setFoo(String foo) {
this.foo = foo;
}
}
public static class BarPojo {
private String bar;
public String getBar() {
return this.bar;
}
public void setBar(String bar) {
this.bar = bar;
}
}
public interface FooInboundChannel1 {
public String INPUT = "foo1-input";
@Input(FooInboundChannel1.INPUT)
SubscribableChannel input();
}
public interface FooOutboundChannel1 {
String OUTPUT = "foo1-output";
@Output(FooOutboundChannel1.OUTPUT)
MessageChannel output();
}
}

View File

@@ -0,0 +1,157 @@
/*
* 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 org.springframework.cloud.stream.config;
import java.util.concurrent.TimeUnit;
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.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_DECLARATIVE_METHOD_PARAMETERS;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.TARGET_BEAN_NOT_EXISTS;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
public class StreamListenerWithAnnotatedInputOutputArgsTests {
@Test
public void testInputOutputArgs() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestInputOutputArgs.class, "--server.port=0");
sendMessageAndValidate(context);
}
@Test
public void testInputOutputArgsWithMoreParameters() {
try {
SpringApplication.run(TestInputOutputArgsWithMoreParameters.class, "--server.port=0");
fail("Expected exception: "+ INVALID_DECLARATIVE_METHOD_PARAMETERS);
}
catch (Exception e) {
assertThat(e.getMessage()).contains(INVALID_DECLARATIVE_METHOD_PARAMETERS);
}
}
@Test
public void testInputOutputArgsWithInvalidBindableTarget() {
try {
SpringApplication.run(TestInputOutputArgsWithInvalidBindableTarget.class, "--server.port=0");
fail("Exception expected on using invalid bindable target as method parameter");
}
catch (Exception e) {
assertThat(e.getMessage()).contains(TARGET_BEAN_NOT_EXISTS + ": invalid");
}
}
@Test
public void testInputOutputArgsWithParameterOrderChanged() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestInputOutputArgsWithParameterOrderChanged.class, "--server.port=0");
sendMessageAndValidate(context);
}
private void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
@SuppressWarnings("unchecked")
Processor processor = context.getBean(Processor.class);
processor.input().send(MessageBuilder.withPayload("hello").setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<?> result = messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo("HELLO");
context.close();
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestInputOutputArgs {
@StreamListener
public void receive(@Input(Processor.INPUT) SubscribableChannel input, @Output(Processor.OUTPUT) final MessageChannel output) {
input.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
output.send(MessageBuilder.withPayload(message.getPayload().toString().toUpperCase()).build());
}
});
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestInputOutputArgsWithMoreParameters {
@StreamListener
public void receive(@Input(Processor.INPUT) SubscribableChannel input, @Output(Processor.OUTPUT) final MessageChannel output,
String someArg) {
input.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
output.send(MessageBuilder.withPayload(message.getPayload().toString().toUpperCase()).build());
}
});
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestInputOutputArgsWithInvalidBindableTarget {
@StreamListener
public void receive(@Input("invalid") SubscribableChannel input, @Output(Processor.OUTPUT) final MessageChannel output) {
input.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
output.send(MessageBuilder.withPayload(message.getPayload().toString().toUpperCase()).build());
}
});
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestInputOutputArgsWithParameterOrderChanged {
@StreamListener
public void receive(@Output(Processor.OUTPUT) final MessageChannel output, @Input("input") SubscribableChannel input) {
input.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
output.send(MessageBuilder.withPayload(message.getPayload().toString().toUpperCase()).build());
}
});
}
}
}

View File

@@ -1,388 +0,0 @@
/*
* 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 org.springframework.cloud.stream.config;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.beans.factory.BeanCreationException;
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.Processor;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.fail;
/**
* @author Marius Bogoevici
*/
public class StreamListenerWithHandlerTests {
@Test
public void testContentTypeConversion() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestSink.class, "--server.port=0");
@SuppressWarnings("unchecked")
TestSink testSink = context.getBean(TestSink.class);
Sink sink = context.getBean(Sink.class);
String id = UUID.randomUUID().toString();
sink.input().send(MessageBuilder.withPayload("{\"bar\":\"barbar" + id + "\"}")
.setHeader("contentType", "application/json").build());
assertThat(testSink.latch.await(10, TimeUnit.SECONDS));
assertThat(testSink.receivedArguments).hasSize(1);
assertThat(testSink.receivedArguments.get(0)).hasFieldOrPropertyWithValue("bar", "barbar" + id);
context.close();
}
@Test
@SuppressWarnings("unchecked")
public void testAnnotatedArguments() throws Exception {
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("{\"bar\":\"barbar" + id + "\"}")
.setHeader("contentType", "application/json").setHeader("testHeader", "testValue").build());
assertThat(testPojoWithAnnotatedArguments.receivedArguments).hasSize(3);
assertThat(testPojoWithAnnotatedArguments.receivedArguments.get(0)).isInstanceOf(FooPojo.class);
assertThat(testPojoWithAnnotatedArguments.receivedArguments.get(0)).hasFieldOrPropertyWithValue("bar",
"barbar" + id);
assertThat(testPojoWithAnnotatedArguments.receivedArguments.get(1)).isInstanceOf(Map.class);
assertThat((Map<String, String>) testPojoWithAnnotatedArguments.receivedArguments.get(1))
.containsEntry(MessageHeaders.CONTENT_TYPE, "application/json");
assertThat((Map<String, String>) testPojoWithAnnotatedArguments.receivedArguments.get(1))
.containsEntry("testHeader", "testValue");
assertThat(testPojoWithAnnotatedArguments.receivedArguments.get(2)).isEqualTo("application/json");
context.close();
}
@Test
@SuppressWarnings("unchecked")
public void testReturn() throws Exception {
ConfigurableApplicationContext context = SpringApplication
.run(TestStringProcessor.class, "--server.port=0");
MessageCollector collector = context.getBean(MessageCollector.class);
Processor processor = context.getBean(Processor.class);
String id = UUID.randomUUID().toString();
processor.input()
.send(MessageBuilder.withPayload("{\"bar\":\"barbar" + id + "\"}")
.setHeader("contentType", "application/json").build());
Message<String> message = (Message<String>) collector
.forChannel(processor.output()).poll(1, TimeUnit.SECONDS);
TestStringProcessor testStringProcessor = context
.getBean(TestStringProcessor.class);
assertThat(testStringProcessor.receivedPojos).hasSize(1);
assertThat(testStringProcessor.receivedPojos.get(0)).hasFieldOrPropertyWithValue("bar", "barbar" + id);
assertThat(message).isNotNull();
assertThat(message.getPayload()).isEqualTo("barbar" + id);
context.close();
}
@Test
@SuppressWarnings("unchecked")
public void testReturnConversion() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestPojoWithMimeType.class,
"--spring.cloud.stream.bindings.output.contentType=application/json", "--server.port=0");
MessageCollector collector = context.getBean(MessageCollector.class);
Processor processor = context.getBean(Processor.class);
String id = UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload("{\"bar\":\"barbar" + id + "\"}")
.setHeader("contentType", "application/json").build());
TestPojoWithMimeType testPojoWithMimeType = context.getBean(TestPojoWithMimeType.class);
assertThat(testPojoWithMimeType.receivedPojos).hasSize(1);
assertThat(testPojoWithMimeType.receivedPojos.get(0)).hasFieldOrPropertyWithValue("bar", "barbar" + id);
Message<String> message = (Message<String>) collector.forChannel(processor.output()).poll(1, TimeUnit.SECONDS);
assertThat(message).isNotNull();
assertThat(message.getPayload()).isEqualTo("{\"qux\":\"barbar" + id + "\"}");
assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE, MimeType.class).includes(MimeTypeUtils.APPLICATION_JSON));
context.close();
}
@Test
@SuppressWarnings("unchecked")
public void testReturnNoConversion() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestPojoWithMimeType.class, "--server.port=0");
MessageCollector collector = context.getBean(MessageCollector.class);
Processor processor = context.getBean(Processor.class);
String id = UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload("{\"bar\":\"barbar" + id + "\"}")
.setHeader("contentType", "application/json").build());
TestPojoWithMimeType testPojoWithMimeType = context.getBean(TestPojoWithMimeType.class);
assertThat(testPojoWithMimeType.receivedPojos).hasSize(1);
assertThat(testPojoWithMimeType.receivedPojos.get(0)).hasFieldOrPropertyWithValue("bar", "barbar" + id);
Message<BazPojo> message = (Message<BazPojo>) collector.forChannel(processor.output()).poll(1,
TimeUnit.SECONDS);
assertThat(message).isNotNull();
assertThat(message.getPayload().getQux()).isEqualTo("barbar" + id);
context.close();
}
@Test
@SuppressWarnings("unchecked")
public void testReturnMessage() throws Exception {
ConfigurableApplicationContext context = SpringApplication
.run(TestPojoWithMessageReturn.class, "--server.port=0");
MessageCollector collector = context.getBean(MessageCollector.class);
Processor processor = context.getBean(Processor.class);
String id = UUID.randomUUID().toString();
processor.input()
.send(MessageBuilder.withPayload("{\"bar\":\"barbar" + id + "\"}")
.setHeader("contentType", "application/json").build());
TestPojoWithMessageReturn testPojoWithMessageReturn = context
.getBean(TestPojoWithMessageReturn.class);
assertThat(testPojoWithMessageReturn.receivedPojos).hasSize(1);
assertThat(testPojoWithMessageReturn.receivedPojos.get(0)).hasFieldOrPropertyWithValue("bar", "barbar" + id);
Message<BazPojo> message = (Message<BazPojo>) collector
.forChannel(processor.output()).poll(1, TimeUnit.SECONDS);
assertThat(message).isNotNull();
assertThat(message.getPayload().getQux()).isEqualTo("barbar" + id);
context.close();
}
@Test
@SuppressWarnings("unchecked")
public void testMessageArgument() throws Exception {
ConfigurableApplicationContext context = SpringApplication
.run(TestPojoWithMessageArgument.class, "--server.port=0");
MessageCollector collector = context.getBean(MessageCollector.class);
Processor processor = context.getBean(Processor.class);
String id = UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload("barbar" + id)
.setHeader("contentType", "text/plain").build());
TestPojoWithMessageArgument testPojoWithMessageArgument = context
.getBean(TestPojoWithMessageArgument.class);
assertThat(testPojoWithMessageArgument.receivedMessages).hasSize(1);
assertThat(testPojoWithMessageArgument.receivedMessages.get(0).getPayload()).isEqualTo("barbar" + id);
Message<BazPojo> message = (Message<BazPojo>) collector
.forChannel(processor.output()).poll(1, TimeUnit.SECONDS);
assertThat(message).isNotNull();
assertThat(message.getPayload().getQux()).isEqualTo("barbar" + id);
context.close();
}
@Test
@SuppressWarnings("unchecked")
public void testDuplicateMapping() throws Exception {
try {
ConfigurableApplicationContext context = SpringApplication.run(TestDuplicateMapping.class,
"--server.port=0");
fail("Exception expected on duplicate mapping");
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).startsWith("Duplicate @StreamListener mapping");
}
}
@Test
@SuppressWarnings("unchecked")
public void testHandlerBean() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestHandlerBean.class,
"--spring.cloud.stream.bindings.output.contentType=application/json", "--server.port=0");
MessageCollector collector = context.getBean(MessageCollector.class);
Processor processor = context.getBean(Processor.class);
String id = UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload("{\"bar\":\"barbar" + id + "\"}")
.setHeader("contentType", "application/json").build());
HandlerBean handlerBean = context.getBean(HandlerBean.class);
assertThat(handlerBean.receivedPojos).hasSize(1);
assertThat(handlerBean.receivedPojos.get(0)).hasFieldOrPropertyWithValue("bar", "barbar" + id);
Message<String> message = (Message<String>) collector.forChannel(processor.output()).poll(1, TimeUnit.SECONDS);
assertThat(message).isNotNull();
assertThat(message.getPayload()).isEqualTo("{\"qux\":\"barbar" + id + "\"}");
assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE, MimeType.class).includes(MimeTypeUtils.APPLICATION_JSON));
context.close();
}
@EnableBinding(Sink.class)
@EnableAutoConfiguration
public static class TestSink {
List<FooPojo> receivedArguments = new ArrayList<>();
CountDownLatch latch = new CountDownLatch(1);
@StreamListener(Sink.INPUT)
public void receive(FooPojo fooPojo) {
this.receivedArguments.add(fooPojo);
this.latch.countDown();
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestStringProcessor {
List<FooPojo> receivedPojos = new ArrayList<>();
@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public String receive(FooPojo fooPojo) {
this.receivedPojos.add(fooPojo);
return fooPojo.getBar();
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestPojoWithMimeType {
List<FooPojo> receivedPojos = new ArrayList<>();
@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public BazPojo receive(FooPojo fooPojo) {
this.receivedPojos.add(fooPojo);
BazPojo bazPojo = new BazPojo();
bazPojo.setQux(fooPojo.getBar());
return bazPojo;
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestPojoWithAnnotatedArguments {
List<Object> receivedArguments = new ArrayList<>();
@StreamListener(Processor.INPUT)
public void receive(@Payload FooPojo fooPojo,
@Headers Map<String, Object> headers,
@Header(MessageHeaders.CONTENT_TYPE) String contentType) {
this.receivedArguments.add(fooPojo);
this.receivedArguments.add(headers);
this.receivedArguments.add(contentType);
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestPojoWithMessageReturn {
List<FooPojo> receivedPojos = new ArrayList<>();
@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public Message<?> receive(FooPojo fooPojo) {
this.receivedPojos.add(fooPojo);
BazPojo bazPojo = new BazPojo();
bazPojo.setQux(fooPojo.getBar());
return MessageBuilder.withPayload(bazPojo).setHeader("foo", "bar").build();
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestPojoWithMessageArgument {
List<Message<String>> receivedMessages = new ArrayList<>();
@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public BazPojo receive(Message<String> fooMessage) {
this.receivedMessages.add(fooMessage);
BazPojo bazPojo = new BazPojo();
bazPojo.setQux(fooMessage.getPayload());
return bazPojo;
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestDuplicateMapping {
@StreamListener(Processor.INPUT)
public void receive(Message<String> fooMessage) {
}
@StreamListener(Processor.INPUT)
public void receive2(Message<String> fooMessage) {
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestHandlerBean {
@Bean
public HandlerBean handlerBean() {
return new HandlerBean();
}
}
public static class HandlerBean {
List<FooPojo> receivedPojos = new ArrayList<>();
@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public BazPojo receive(FooPojo fooMessage) {
this.receivedPojos.add(fooMessage);
BazPojo bazPojo = new BazPojo();
bazPojo.setQux(fooMessage.getBar());
return bazPojo;
}
}
public static class FooPojo {
private String bar;
public String getBar() {
return this.bar;
}
public void setBar(String bar) {
this.bar = bar;
}
}
public static class BazPojo {
private String qux;
public String getQux() {
return this.qux;
}
public void setQux(String qux) {
this.qux = qux;
}
}
}

View File

@@ -20,7 +20,6 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import reactor.core.publisher.MonoProcessor;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter;
import org.springframework.core.MethodParameter;
import org.springframework.core.ResolvableType;
@@ -42,7 +41,6 @@ public class MessageChannelToFluxSenderParameterAdapter
public boolean supports(Class<?> boundElementType, MethodParameter methodParameter) {
ResolvableType type = ResolvableType.forMethodParameter(methodParameter);
return MessageChannel.class.isAssignableFrom(boundElementType)
&& methodParameter.getParameterAnnotation(Output.class) != null
&& FluxSender.class.isAssignableFrom(type.getRawClass());
}

View File

@@ -18,7 +18,6 @@ package org.springframework.cloud.stream.reactive;
import reactor.core.publisher.Flux;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter;
import org.springframework.core.MethodParameter;
import org.springframework.core.ResolvableType;
@@ -48,7 +47,6 @@ public class MessageChannelToInputFluxParameterAdapter
@Override
public boolean supports(Class<?> boundElementType, MethodParameter methodParameter) {
return SubscribableChannel.class.isAssignableFrom(boundElementType)
&& methodParameter.getParameterAnnotation(Input.class) != null
&& Flux.class.isAssignableFrom(methodParameter.getParameterType());
}

View File

@@ -19,7 +19,6 @@ package org.springframework.cloud.stream.reactive;
import reactor.adapter.RxJava1Adapter;
import rx.Observable;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter;
import org.springframework.core.MethodParameter;
import org.springframework.messaging.MessageChannel;
@@ -44,7 +43,6 @@ public class MessageChannelToInputObservableParameterAdapter
public boolean supports(Class<?> boundElementType, MethodParameter methodParameter) {
return SubscribableChannel.class.isAssignableFrom(boundElementType)
&& methodParameter.getParameterAnnotation(Input.class) != null
&& Observable.class.isAssignableFrom(methodParameter.getParameterType());
}

View File

@@ -20,7 +20,6 @@ import reactor.adapter.RxJava1Adapter;
import rx.Observable;
import rx.Single;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter;
import org.springframework.core.MethodParameter;
import org.springframework.core.ResolvableType;
@@ -47,7 +46,6 @@ public class MessageChannelToObservableSenderParameterAdapter implements
public boolean supports(Class<?> boundElementType, MethodParameter methodParameter) {
ResolvableType type = ResolvableType.forMethodParameter(methodParameter);
return MessageChannel.class.isAssignableFrom(boundElementType)
&& methodParameter.getParameterAnnotation(Output.class) != null
&& ObservableSender.class.isAssignableFrom(type.getRawClass());
}

View File

@@ -40,6 +40,7 @@ import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
*/
@SuppressWarnings("unchecked")
public class MessageChannelToInputFluxParameterAdapterTests {
@Test

View File

@@ -0,0 +1,103 @@
/*
* 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 org.springframework.cloud.stream.reactive;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import reactor.core.publisher.Flux;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM;
/**
* @author Ilayaperumal Gopinathan
*/
@SuppressWarnings("unchecked")
public class StreamListenerGenericFluxInputOutputArgsWithMessageTests {
@Test
public void testGenericFluxInputOutputArgsWithMessage() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestGenericStringFluxInputOutputArgsWithMessageImpl1.class, "--server.port=0");
sendMessageAndValidate(context);
context.close();
}
@Test
public void testInvalidInputValueWithOutputMethodParameters() {
try {
SpringApplication.run(TestGenericStringFluxInputOutputArgsWithMessageImpl2.class, "--server.port=0");
fail("Expected exception: " + INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM);
}
catch (Exception e) {
assertThat(e.getMessage()).contains(INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM);
}
}
@SuppressWarnings("unchecked")
private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
Processor processor = context.getBean(Processor.class);
String sentPayload = "hello " + UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload(sentPayload).setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<?> result = messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
}
public static class TestGenericStringFluxInputOutputArgsWithMessageImpl1 extends TestGenericFluxInputOutputArgsWithMessage1<String> {
}
public static class TestGenericStringFluxInputOutputArgsWithMessageImpl2 extends TestGenericFluxInputOutputArgsWithMessage2<String> {
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestGenericFluxInputOutputArgsWithMessage1<A> {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<A> input,
@Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> MessageBuilder.withPayload((A) m.toString().toUpperCase()).build()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestGenericFluxInputOutputArgsWithMessage2<A> {
@StreamListener(Processor.INPUT)
public void receive(Flux<A> input,
@Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> MessageBuilder.withPayload((A) m.toString().toUpperCase()).build()));
}
}
}

View File

@@ -14,11 +14,18 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.config;
package org.springframework.cloud.stream.reactive;
import java.util.Arrays;
import java.util.Collection;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import reactor.core.publisher.Flux;
import rx.Observable;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
@@ -30,50 +37,63 @@ import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
public class StreamListenerWithAnnotatedArgsTests {
@RunWith(Parameterized.class)
public class StreamListenerReactiveInputOutputArgsTests {
private Class<?> configClass;
public StreamListenerReactiveInputOutputArgsTests(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection InputConfigs() {
return Arrays.asList(new Class[]{ReactorTestInputOutputArgs.class, RxJava1TestInputOutputArgs.class});
}
@Test
public void testInputOutputArgs() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestInputOutputArgs.class, "--server.port=0");
ConfigurableApplicationContext context = SpringApplication.run(this.configClass, "--server.port=0");
sendMessageAndValidate(context);
}
private void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
@SuppressWarnings("unchecked")
Processor processor = context.getBean(Processor.class);
processor.input().send(MessageBuilder.withPayload("hello").setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<?> result = messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo("HELLO");
context.close();
}
private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
@SuppressWarnings("unchecked")
Processor processor = context.getBean(Processor.class);
String sentPayload = "hello " + UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload(sentPayload).setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<?> result = messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestInputOutputArgs {
public static class ReactorTestInputOutputArgs {
@StreamListener
public void receive(@Input(Processor.INPUT) SubscribableChannel input, @Output(Processor.OUTPUT) final MessageChannel output) {
input.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
output.send(MessageBuilder.withPayload(message.getPayload().toString().toUpperCase()).build());
}
});
public void receive(@Input(Processor.INPUT) Flux<String> input, @Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> m.toUpperCase()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestInputOutputArgs {
@StreamListener
public void receive(@Input(Processor.INPUT) Observable<String> input, @Output(Processor.OUTPUT) ObservableSender output) {
output.send(input.map(m -> m.toUpperCase()));
}
}
}

View File

@@ -0,0 +1,102 @@
/*
* 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 org.springframework.cloud.stream.reactive;
import java.util.Arrays;
import java.util.Collection;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import reactor.core.publisher.Flux;
import rx.Observable;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
@RunWith(Parameterized.class)
public class StreamListenerReactiveInputOutputArgsWithMessageTests {
private Class<?> configClass;
public StreamListenerReactiveInputOutputArgsWithMessageTests(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection InputConfigs() {
return Arrays.asList(new Class[]{ReactorTestInputOutputArgsWithMessage.class, RxJava1TestInputOutputArgsWithMessage.class});
}
@Test
public void testInputOutputArgs() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(this.configClass, "--server.port=0");
sendMessageAndValidate(context);
context.close();
}
private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
@SuppressWarnings("unchecked")
Processor processor = context.getBean(Processor.class);
String sentPayload = "hello " + UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload(sentPayload).setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<?> result = messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestInputOutputArgsWithMessage {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<Message<?>> input,
@Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> MessageBuilder
.withPayload(m.getPayload().toString().toUpperCase()).build()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestInputOutputArgsWithMessage {
@StreamListener
public void receive(@Input(Processor.INPUT) Observable<Message<String>> input,
@Output(Processor.OUTPUT) ObservableSender output) {
output.send(input.map(m -> MessageBuilder.withPayload(m.getPayload().toUpperCase()).build()));
}
}
}

View File

@@ -0,0 +1,126 @@
/*
* 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 org.springframework.cloud.stream.reactive;
import java.util.Arrays;
import java.util.Collection;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import reactor.core.publisher.Flux;
import rx.Observable;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
@RunWith(Parameterized.class)
public class StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests {
private Class<?> configClass;
public StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection InputConfigs() {
return Arrays.asList(new Class[]{TestInputOutputArgsWithFluxSenderAndFailure.class, TestInputOutputArgsWithObservableSenderAndFailure.class});
}
@Test
public void testInputOutputArgsWithFluxSenderAndFailure() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(this.configClass, "--server.port=0");
sendMessageAndValidate(context);
sendFailingMessage(context);
sendMessageAndValidate(context);
context.close();
}
private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
@SuppressWarnings("unchecked")
Processor processor = context.getBean(Processor.class);
String sentPayload = "hello " + UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload(sentPayload).setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<?> result = messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
}
private static void sendFailingMessage(ConfigurableApplicationContext context) throws InterruptedException {
@SuppressWarnings("unchecked")
Processor processor = context.getBean(Processor.class);
processor.input().send(MessageBuilder.withPayload("fail").setHeader("contentType", "text/plain").build());
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestInputOutputArgsWithFluxSenderAndFailure {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<Message<String>> input, @Output(Processor.OUTPUT) FluxSender output) {
output.send(input
.map(m -> m.getPayload().toString())
.map(m -> {
if (!m.equals("fail")) {
return m.toUpperCase();
}
else {
throw new RuntimeException();
}
})
.map(o -> MessageBuilder.withPayload(o).build()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestInputOutputArgsWithObservableSenderAndFailure {
@StreamListener
public void receive(@Input(Processor.INPUT) Observable<Message<String>> input, @Output(Processor.OUTPUT) ObservableSender output) {
output.send(input
.map(m -> m.getPayload().toString())
.map(m -> {
if (!m.equals("fail")) {
return m.toUpperCase();
}
else {
throw new RuntimeException();
}
})
.map(o -> MessageBuilder.withPayload(o).build()));
}
}
}

View File

@@ -0,0 +1,107 @@
/*
* 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 org.springframework.cloud.stream.reactive;
import java.util.Arrays;
import java.util.Collection;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import reactor.core.publisher.Flux;
import rx.Observable;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
@RunWith(Parameterized.class)
public class StreamListenerReactiveInputOutputArgsWithSenderTests {
private Class<?> configClass;
public StreamListenerReactiveInputOutputArgsWithSenderTests(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection InputConfigs() {
return Arrays.asList(new Class[]{ReactorTestInputOutputArgsWithFluxSender.class, RxJava1TestInputOutputArgsWithObservableSender.class});
}
@Test
public void testInputOutputArgsWithFluxSender() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(this.configClass,
"--server.port=0");
// send multiple message
sendMessageAndValidate(context);
sendMessageAndValidate(context);
sendMessageAndValidate(context);
context.close();
}
private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
@SuppressWarnings("unchecked")
Processor processor = context.getBean(Processor.class);
String sentPayload = "hello " + UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload(sentPayload).setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<?> result = messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestInputOutputArgsWithFluxSender {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<Message<String>> input, @Output(Processor.OUTPUT) FluxSender output) {
output.send(input
.map(m -> m.getPayload().toString().toUpperCase())
.map(o -> MessageBuilder.withPayload(o).build()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestInputOutputArgsWithObservableSender {
@StreamListener
public void receive(@Input(Processor.INPUT) Observable<Message<?>> input, @Output(Processor.OUTPUT)
ObservableSender output) {
output.send(input
.map(m -> m.getPayload().toString().toUpperCase())
.map(o -> MessageBuilder.withPayload(o).build()));
}
}
}

View File

@@ -0,0 +1,102 @@
/*
* 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 org.springframework.cloud.stream.reactive;
import org.junit.Test;
import reactor.core.publisher.Flux;
import rx.Observable;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.RETURN_TYPE_NO_OUTBOUND_SPECIFIED;
/**
* @author Ilayaperumal Gopinathan
*/
public class StreamListenerReactiveMethodTests {
@Test
public void testReactiveInvalidInputValueWithOutputMethodParameters() {
try {
SpringApplication.run(ReactorTestInputOutputArgs.class, "--server.port=0");
fail("IllegalArgumentException should have been thrown");
}
catch (Exception e) {
assertThat(e.getMessage()).contains(INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM);
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestInputOutputArgs {
@StreamListener(Processor.INPUT)
public void receive(Flux<String> input, @Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> m.toUpperCase()));
}
}
@Test
public void testRxJava1InvalidInputValueWithOutputMethodParameters() {
try {
SpringApplication.run(RxJava1TestInputOutputArgs.class, "--server.port=0");
fail("IllegalArgumentException should have been thrown");
}
catch (Exception e) {
assertThat(e.getMessage()).contains(INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM);
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestInputOutputArgs {
@StreamListener(Processor.INPUT)
public void receive(Observable<String> input, @Output(Processor.OUTPUT) ObservableSender output) {
output.send(input.map(m -> m.toUpperCase()));
}
}
@Test
public void testMethodReturnTypeWithNoOutboundSpecified() {
try {
SpringApplication.run(ReactorTestReturn5.class, "--server.port=0");
fail("Exception expected: " + RETURN_TYPE_NO_OUTBOUND_SPECIFIED);
}
catch (Exception e) {
assertThat(e.getMessage()).contains(RETURN_TYPE_NO_OUTBOUND_SPECIFIED);
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturn5 {
@StreamListener
public Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
return input.map(m -> m.toUpperCase());
}
}
}

View File

@@ -0,0 +1,176 @@
/*
* 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 org.springframework.cloud.stream.reactive;
import java.util.Arrays;
import java.util.Collection;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import reactor.core.publisher.Flux;
import rx.Observable;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
@RunWith(Parameterized.class)
public class StreamListenerReactiveMethodWithReturnTypeTests {
private Class<?> configClass;
public StreamListenerReactiveMethodWithReturnTypeTests(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection InputConfigs() {
return Arrays.asList(new Class[]{ReactorTestReturn1.class, ReactorTestReturn2.class, ReactorTestReturn3.class, ReactorTestReturn4.class,
RxJava1TestReturn1.class, RxJava1TestReturn2.class, RxJava1TestReturn3.class, RxJava1TestReturn4.class});
}
@Test
public void testReturn() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(this.configClass, "--server.port=0");
sendMessageAndValidate(context);
sendMessageAndValidate(context);
sendMessageAndValidate(context);
context.close();
}
private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
@SuppressWarnings("unchecked")
Processor processor = context.getBean(Processor.class);
String sentPayload = "hello " + UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload(sentPayload).setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<?> result = messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturn1 {
@StreamListener
public
@Output(Processor.OUTPUT)
Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
return input.map(m -> m.toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturn2 {
@StreamListener(Processor.INPUT)
@Output(Processor.OUTPUT)
public Flux<String> receive(Flux<String> input) {
return input.map(m -> m.toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturn3 {
@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public Flux<String> receive(Flux<String> input) {
return input.map(m -> m.toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturn4 {
@StreamListener
@SendTo(Processor.OUTPUT)
public Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
return input.map(m -> m.toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturn1 {
@StreamListener
public
@Output(Processor.OUTPUT)
Observable<String> receive(@Input(Processor.INPUT) Observable<String> input) {
return input.map(m -> m.toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturn2 {
@StreamListener(Processor.INPUT)
public
@Output(Processor.OUTPUT)
Observable<String> receive(Observable<String> input) {
return input.map(m -> m.toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturn3 {
@StreamListener(Processor.INPUT)
public
@SendTo(Processor.OUTPUT)
Observable<String> receive(Observable<String> input) {
return input.map(m -> m.toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturn4 {
@StreamListener
public
@SendTo(Processor.OUTPUT)
Observable<String> receive(@Input(Processor.INPUT) Observable<String> input) {
return input.map(m -> m.toUpperCase());
}
}
}

View File

@@ -0,0 +1,244 @@
/*
* 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 org.springframework.cloud.stream.reactive;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.Arrays;
import java.util.Collection;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import reactor.core.publisher.Flux;
import rx.Observable;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.support.MessageBuilder;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
@RunWith(Parameterized.class)
public class StreamListenerReactiveReturnWithFailureTests {
private Class<?> configClass;
public StreamListenerReactiveReturnWithFailureTests(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection InputConfigs() {
return Arrays.asList(new Class[] {ReactorTestReturnWithFailure1.class, ReactorTestReturnWithFailure2.class,
ReactorTestReturnWithFailure3.class, ReactorTestReturnWithFailure4.class, RxJava1TestReturnWithFailure1.class,
RxJava1TestReturnWithFailure2.class, RxJava1TestReturnWithFailure3.class, RxJava1TestReturnWithFailure4.class});
}
@Test
public void testReturnWithFailure() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(this.configClass, "--server.port=0");
sendMessageAndValidate(context);
sendFailingMessage(context);
sendMessageAndValidate(context);
sendFailingMessage(context);
sendMessageAndValidate(context);
context.close();
}
private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
@SuppressWarnings("unchecked")
Processor processor = context.getBean(Processor.class);
String sentPayload = "hello " + UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload(sentPayload).setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<?> result = messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
}
private static void sendFailingMessage(ConfigurableApplicationContext context) throws InterruptedException {
@SuppressWarnings("unchecked")
Processor processor = context.getBean(Processor.class);
processor.input().send(MessageBuilder.withPayload("fail").setHeader("contentType", "text/plain").build());
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithFailure1 {
@StreamListener
public
@Output(Processor.OUTPUT)
Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
return input.map(m -> {
if (!m.equals("fail")) {
return m.toUpperCase();
}
else {
throw new RuntimeException();
}
});
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithFailure2 {
@StreamListener(Processor.INPUT)
public
@Output(Processor.OUTPUT)
Flux<String> receive(Flux<String> input) {
return input.map(m -> {
if (!m.equals("fail")) {
return m.toUpperCase();
}
else {
throw new RuntimeException();
}
});
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithFailure3 {
@StreamListener(Processor.INPUT)
public
@SendTo(Processor.OUTPUT)
Flux<String> receive(Flux<String> input) {
return input.map(m -> {
if (!m.equals("fail")) {
return m.toUpperCase();
}
else {
throw new RuntimeException();
}
});
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithFailure4 {
@StreamListener
public
@SendTo(Processor.OUTPUT)
Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
return input.map(m -> {
if (!m.equals("fail")) {
return m.toUpperCase();
}
else {
throw new RuntimeException();
}
});
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturnWithFailure1 {
@StreamListener
public
@Output(Processor.OUTPUT)
Observable<String> receive(@Input(Processor.INPUT) Observable<String> input) {
return input.map(m -> {
if (!m.equals("fail")) {
return m.toUpperCase();
}
else {
throw new RuntimeException();
}
});
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturnWithFailure2 {
@StreamListener
public
@SendTo(Processor.OUTPUT)
Observable<String> receive(@Input(Processor.INPUT) Observable<String> input) {
return input.map(m -> {
if (!m.equals("fail")) {
return m.toUpperCase();
}
else {
throw new RuntimeException();
}
});
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturnWithFailure3 {
@StreamListener(Processor.INPUT)
public
@SendTo(Processor.OUTPUT)
Observable<String> receive(Observable<String> input) {
return input.map(m -> {
if (!m.equals("fail")) {
return m.toUpperCase();
}
else {
throw new RuntimeException();
}
});
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturnWithFailure4 {
@StreamListener(Processor.INPUT)
public
@Output(Processor.OUTPUT)
Observable<String> receive(Observable<String> input) {
return input.map(m -> {
if (!m.equals("fail")) {
return m.toUpperCase();
}
else {
throw new RuntimeException();
}
});
}
}
}

View File

@@ -0,0 +1,180 @@
/*
* 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 org.springframework.cloud.stream.reactive;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.Arrays;
import java.util.Collection;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import reactor.core.publisher.Flux;
import rx.Observable;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.support.MessageBuilder;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
@RunWith(Parameterized.class)
public class StreamListenerReactiveReturnWithMessageTests {
private Class<?> configClass;
public StreamListenerReactiveReturnWithMessageTests(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection InputConfigs() {
return Arrays.asList(new Class[] {ReactorTestReturnWithMessage1.class, ReactorTestReturnWithMessage2.class,
ReactorTestReturnWithMessage3.class, ReactorTestReturnWithMessage4.class, RxJava1TestReturnWithMessage1.class,
RxJava1TestReturnWithMessage2.class, RxJava1TestReturnWithMessage3.class, RxJava1TestReturnWithMessage4.class});
}
@Test
public void testReturnWithMessage() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(this.configClass, "--server.port=0");
sendMessageAndValidate(context);
context.close();
}
private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
@SuppressWarnings("unchecked")
Processor processor = context.getBean(Processor.class);
String sentPayload = "hello " + UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload(sentPayload).setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<?> result = messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithMessage1 {
@StreamListener
public
@Output(Processor.OUTPUT)
Flux<String> receive(@Input(Processor.INPUT) Flux<Message<String>> input) {
return input.map(m -> m.getPayload().toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithMessage2 {
@StreamListener(Processor.INPUT)
public
@Output(Processor.OUTPUT)
Flux<String> receive(Flux<Message<String>> input) {
return input.map(m -> m.getPayload().toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithMessage3 {
@StreamListener(Processor.INPUT)
public
@SendTo(Processor.OUTPUT)
Flux<String> receive(Flux<Message<String>> input) {
return input.map(m -> m.getPayload().toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithMessage4 {
@StreamListener
public
@SendTo(Processor.OUTPUT)
Flux<String> receive(@Input(Processor.INPUT) Flux<Message<String>> input) {
return input.map(m -> m.getPayload().toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturnWithMessage1 {
@StreamListener
public
@Output(Processor.OUTPUT)
Observable<String> receive(@Input(Processor.INPUT) Observable<Message<String>> input) {
return input.map(m -> m.getPayload().toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturnWithMessage2 {
@StreamListener
public
@SendTo(Processor.OUTPUT)
Observable<String> receive(@Input(Processor.INPUT) Observable<Message<String>> input) {
return input.map(m -> m.getPayload().toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturnWithMessage3 {
@StreamListener(Processor.INPUT)
public
@Output(Processor.OUTPUT)
Observable<String> receive(Observable<Message<String>> input) {
return input.map(m -> m.getPayload().toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturnWithMessage4 {
@StreamListener(Processor.INPUT)
public
@SendTo(Processor.OUTPUT)
Observable<String> receive(Observable<Message<String>> input) {
return input.map(m -> m.getPayload().toUpperCase());
}
}
}

View File

@@ -0,0 +1,204 @@
/*
* 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 org.springframework.cloud.stream.reactive;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import reactor.core.publisher.Flux;
import rx.Observable;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.support.MessageBuilder;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
@RunWith(Parameterized.class)
public class StreamListenerReactiveReturnWithPojoTests {
private Class<?> configClass;
public StreamListenerReactiveReturnWithPojoTests(Class<?> configClass) {
this.configClass = configClass;
}
@Parameterized.Parameters
public static Collection InputConfigs() {
return Arrays.asList(new Class[] {ReactorTestReturnWithPojo1.class, ReactorTestReturnWithPojo2.class,
ReactorTestReturnWithPojo3.class, ReactorTestReturnWithPojo4.class, RxJava1TestReturnWithPojo1.class,
RxJava1TestReturnWithPojo2.class, RxJava1TestReturnWithPojo3.class, RxJava1TestReturnWithPojo4.class});
}
@Test
public void testReturnWithPojo() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(this.configClass, "--server.port=0");
@SuppressWarnings("unchecked")
Processor processor = context.getBean(Processor.class);
processor.input().send(MessageBuilder.withPayload("{\"message\":\"helloPojo\"}")
.setHeader("contentType", "application/json").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<?> result = messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isInstanceOf(BarPojo.class);
assertThat(((BarPojo) result.getPayload()).getBarMessage()).isEqualTo("helloPojo");
context.close();
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithPojo1 {
@StreamListener
public
@Output(Processor.OUTPUT)
Flux<BarPojo> receive(@Input(Processor.INPUT) Flux<FooPojo> input) {
return input.map(m -> new BarPojo(m.getMessage()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithPojo2 {
@StreamListener(Processor.INPUT)
public
@Output(Processor.OUTPUT)
Flux<BarPojo> receive(Flux<FooPojo> input) {
return input.map(m -> new BarPojo(m.getMessage()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithPojo3 {
@StreamListener(Processor.INPUT)
public
@SendTo(Processor.OUTPUT)
Flux<BarPojo> receive(Flux<FooPojo> input) {
return input.map(m -> new BarPojo(m.getMessage()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturnWithPojo4 {
@StreamListener
public
@SendTo(Processor.OUTPUT)
Flux<BarPojo> receive(@Input(Processor.INPUT) Flux<FooPojo> input) {
return input.map(m -> new BarPojo(m.getMessage()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturnWithPojo1 {
@StreamListener
public
@Output(Processor.OUTPUT)
Observable<BarPojo> receive(@Input(Processor.INPUT) Observable<FooPojo> input) {
return input.map(m -> new BarPojo(m.getMessage()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturnWithPojo2 {
@StreamListener
public
@SendTo(Processor.OUTPUT)
Observable<BarPojo> receive(@Input(Processor.INPUT) Observable<FooPojo> input) {
return input.map(m -> new BarPojo(m.getMessage()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturnWithPojo3 {
@StreamListener(Processor.INPUT)
public
@Output(Processor.OUTPUT)
Observable<BarPojo> receive(Observable<FooPojo> input) {
return input.map(m -> new BarPojo(m.getMessage()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturnWithPojo4 {
@StreamListener(Processor.INPUT)
public
@SendTo(Processor.OUTPUT)
Observable<BarPojo> receive(Observable<FooPojo> input) {
return input.map(m -> new BarPojo(m.getMessage()));
}
}
public static class FooPojo {
private String message;
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
}
public static class BarPojo {
private String barMessage;
public BarPojo(String barMessage) {
this.barMessage = barMessage;
}
public String getBarMessage() {
return barMessage;
}
public void setBarMessage(String barMessage) {
this.barMessage = barMessage;
}
}
}

View File

@@ -1,317 +0,0 @@
/*
* 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 org.springframework.cloud.stream.reactive;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import reactor.core.publisher.Flux;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
public class StreamListenerReactorTests {
@Test
public void testInputOutputArgs() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestInputOutputArgs.class, "--server.port=0");
sendMessageAndValidate(context);
context.close();
}
private void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
@SuppressWarnings("unchecked")
Processor processor = context.getBean(Processor.class);
String sentPayload = "hello " + UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload(sentPayload).setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<?> result = messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
}
private void sendFailingMessage(ConfigurableApplicationContext context) throws InterruptedException {
@SuppressWarnings("unchecked")
Processor processor = context.getBean(Processor.class);
processor.input().send(MessageBuilder.withPayload("fail").setHeader("contentType", "text/plain").build());
}
@Test
public void testInputOutputArgsWithMessage() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestInputOutputArgsWithMessage.class,
"--server.port=0");
sendMessageAndValidate(context);
context.close();
}
@Test
public void testWildCardFluxInputOutputArgsWithMessage() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestWildCardFluxInputOutputArgsWithMessage.class,
"--server.port=0");
sendMessageAndValidate(context);
context.close();
}
@Test
public void testGenericFluxInputOutputArgsWithMessage() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestGenericStringFluxInputOutputArgsWithMessage.class,
"--server.port=0");
sendMessageAndValidate(context);
context.close();
}
@Test
public void testInputOutputArgsWithFluxSender() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestInputOutputArgsWithFluxSender.class,
"--server.port=0");
// send multiple message
sendMessageAndValidate(context);
sendMessageAndValidate(context);
sendMessageAndValidate(context);
context.close();
}
@Test
public void testInputOutputArgsWithFluxSenderAndFailure() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestInputOutputArgsWithFluxSenderAndFailure.class, "--server.port=0");
sendMessageAndValidate(context);
sendFailingMessage(context);
sendMessageAndValidate(context);
context.close();
}
@Test
public void testReturn() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestReturn.class, "--server.port=0");
sendMessageAndValidate(context);
sendMessageAndValidate(context);
sendMessageAndValidate(context);
context.close();
}
@Test
public void testReturnWithFailure() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestReturnWithFailure.class, "--server.port=0");
sendMessageAndValidate(context);
sendFailingMessage(context);
sendMessageAndValidate(context);
sendFailingMessage(context);
sendMessageAndValidate(context);
context.close();
}
@Test
public void testReturnWithMessage() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestReturnWithMessage.class, "--server.port=0");
sendMessageAndValidate(context);
context.close();
}
@Test
public void testReturnWithPojo() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestReturnWithPojo.class, "--server.port=0");
@SuppressWarnings("unchecked")
Processor processor = context.getBean(Processor.class);
processor.input().send(MessageBuilder.withPayload("{\"message\":\"helloPojo\"}")
.setHeader("contentType", "application/json").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<?> result = messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isInstanceOf(BarPojo.class);
assertThat(((BarPojo) result.getPayload()).getBarMessage()).isEqualTo("helloPojo");
context.close();
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestInputOutputArgs {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<String> input, @Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> m.toUpperCase()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestInputOutputArgsWithMessage {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<Message<?>> input,
@Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> MessageBuilder
.withPayload(m.getPayload().toString().toUpperCase()).build()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestWildCardFluxInputOutputArgsWithMessage {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<?> input,
@Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> MessageBuilder.withPayload(m.toString().toUpperCase()).build()));
}
}
public static class TestGenericStringFluxInputOutputArgsWithMessage extends TestGenericFluxInputOutputArgsWithMessage<String> {
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestGenericFluxInputOutputArgsWithMessage<A> {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<A> input,
@Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> MessageBuilder.withPayload((A)m.toString().toUpperCase()).build()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestInputOutputArgsWithFluxSender {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<Message<String>> input, @Output(Processor.OUTPUT) FluxSender output) {
output.send(input
.map(m -> m.getPayload().toString().toUpperCase())
.map(o -> MessageBuilder.withPayload(o).build()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestInputOutputArgsWithFluxSenderAndFailure {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<Message<?>> input, @Output(Processor.OUTPUT) FluxSender output) {
output.send(input
.map(m -> m.getPayload().toString())
.map(m -> {
if (!m.equals("fail")) {
return m.toUpperCase();
}
else {
throw new RuntimeException();
}
})
.map(o -> MessageBuilder.withPayload(o).build()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestReturn {
@StreamListener
public
@Output(Processor.OUTPUT)
Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
return input.map(m -> m.toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestReturnWithFailure {
@StreamListener
public
@Output(Processor.OUTPUT)
Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
return input.map(m -> {
if (!m.equals("fail")) {
return m.toUpperCase();
}
else {
throw new RuntimeException();
}
});
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestReturnWithMessage {
@StreamListener
public
@Output(Processor.OUTPUT)
Flux<String> receive(@Input(Processor.INPUT) Flux<Message<String>> input) {
return input.map(m -> m.getPayload().toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestReturnWithPojo {
@StreamListener
public
@Output(Processor.OUTPUT)
Flux<BarPojo> receive(@Input(Processor.INPUT) Flux<FooPojo> input) {
return input.map(m -> new BarPojo(m.getMessage()));
}
}
public static class FooPojo {
private String message;
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
}
public static class BarPojo {
private String barMessage;
public BarPojo(String barMessage) {
this.barMessage = barMessage;
}
public String getBarMessage() {
return barMessage;
}
public void setBarMessage(String barMessage) {
this.barMessage = barMessage;
}
}
}

View File

@@ -1,245 +0,0 @@
/*
* 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 org.springframework.cloud.stream.reactive;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import rx.Observable;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
*/
public class StreamListenerRxJava1Tests {
@Test
public void testInputOutputArgs() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestInputOutputArgs.class, "--server.port=0");
sendMessageAndValidate(context);
context.close();
}
private void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
@SuppressWarnings("unchecked")
Processor processor = context.getBean(Processor.class);
String sentPayload = "hello " + UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload(sentPayload).setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<?> result = messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
}
private void sendFailingMessage(ConfigurableApplicationContext context) throws InterruptedException {
@SuppressWarnings("unchecked")
Processor processor = context.getBean(Processor.class);
processor.input().send(MessageBuilder.withPayload("fail").setHeader("contentType", "text/plain").build());
}
@Test
public void testInputOutputArgsWithMessage() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestInputOutputArgsWithMessage.class,
"--server.port=0");
sendMessageAndValidate(context);
context.close();
}
@Test
public void testInputOutputArgsWithObservableSender() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestInputOutputArgsWithObservableSender.class,
"--server.port=0");
// send multiple message
sendMessageAndValidate(context);
sendMessageAndValidate(context);
sendMessageAndValidate(context);
context.close();
}
@Test
public void testReturn() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestReturn.class, "--server.port=0");
sendMessageAndValidate(context);
sendMessageAndValidate(context);
sendMessageAndValidate(context);
context.close();
}
@Test
public void testReturnWithFailure() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestReturnWithFailure.class, "--server.port=0");
sendMessageAndValidate(context);
sendFailingMessage(context);
sendMessageAndValidate(context);
context.close();
}
@Test
public void testReturnWithMessage() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestReturnWithMessage.class, "--server.port=0");
sendMessageAndValidate(context);
context.close();
}
@Test
public void testReturnWithPojo() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestReturnWithPojo.class, "--server.port=0");
@SuppressWarnings("unchecked")
Processor processor = context.getBean(Processor.class);
processor.input().send(MessageBuilder.withPayload("{\"message\":\"helloPojo\"}")
.setHeader("contentType", "application/json").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<?> result = messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isInstanceOf(BarPojo.class);
assertThat(((BarPojo) result.getPayload()).getBarMessage()).isEqualTo("helloPojo");
context.close();
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestInputOutputArgs {
@StreamListener
public void receive(@Input(Processor.INPUT) Observable<String> input, @Output(Processor.OUTPUT) ObservableSender output) {
output.send(input.map(m -> m.toUpperCase()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestInputOutputArgsWithMessage {
@StreamListener
public void receive(@Input(Processor.INPUT) Observable<Message<String>> input,
@Output(Processor.OUTPUT) ObservableSender output) {
output.send(input.map(m -> MessageBuilder.withPayload(m.getPayload().toUpperCase()).build()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestInputOutputArgsWithObservableSender {
@StreamListener
public void receive(@Input(Processor.INPUT) Observable<Message<?>> input, @Output(Processor.OUTPUT)
ObservableSender output) {
output.send(input
.map(m -> m.getPayload().toString().toUpperCase())
.map(o -> MessageBuilder.withPayload(o).build()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestReturn {
@StreamListener
public
@Output(Processor.OUTPUT)
Observable<String> receive(@Input(Processor.INPUT) Observable<String> input) {
return input.map(m -> m.toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestReturnWithFailure {
@StreamListener
public
@Output(Processor.OUTPUT)
Observable<String> receive(@Input(Processor.INPUT) Observable<String> input) {
return input.map(m -> {
if (!m.equals("fail")) {
return m.toUpperCase();
}
else {
throw new RuntimeException();
}
});
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestReturnWithMessage {
@StreamListener
public
@Output(Processor.OUTPUT)
Observable<String> receive(@Input(Processor.INPUT) Observable<Message<String>> input) {
return input.map(m -> m.getPayload().toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestReturnWithPojo {
@StreamListener
public
@Output(Processor.OUTPUT)
Observable<BarPojo> receive(@Input(Processor.INPUT) Observable<FooPojo> input) {
return input.map(m -> new BarPojo(m.getMessage()));
}
}
public static class FooPojo {
private String message;
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
}
public static class BarPojo {
private String barMessage;
public BarPojo(String barMessage) {
this.barMessage = barMessage;
}
public String getBarMessage() {
return barMessage;
}
public void setBarMessage(String barMessage) {
this.barMessage = barMessage;
}
}
}

View File

@@ -0,0 +1,119 @@
/*
* 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 org.springframework.cloud.stream.reactive;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import reactor.core.publisher.Flux;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_DECLARATIVE_METHOD_PARAMETERS;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM;
/**
* @author Ilayaperumal Gopinathan
*/
public class StreamListenerWildCardFluxInputOutputArgsWithMessageTests {
@Test
public void testWildCardFluxInputOutputArgsWithMessage() throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(TestWildCardFluxInputOutputArgsWithMessage1.class, "--server.port=0");
sendMessageAndValidate(context);
context.close();
}
@Test
public void testInputAsStreamListenerAndOutputAsParameterUsage() {
try {
SpringApplication.run(TestWildCardFluxInputOutputArgsWithMessage2.class, "--server.port=0");
fail("Expected exception: " + INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM);
}
catch (Exception e) {
assertThat(e.getMessage()).contains(INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM);
}
}
@Test
public void testIncorrectUsage1() throws Exception {
try {
SpringApplication.run(TestWildCardFluxInputOutputArgsWithMessage3.class, "--server.port=0");
fail("Expected exception: " + INVALID_DECLARATIVE_METHOD_PARAMETERS);
}
catch (Exception e) {
assertThat(e.getMessage()).contains(INVALID_DECLARATIVE_METHOD_PARAMETERS);
}
}
private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException {
@SuppressWarnings("unchecked")
Processor processor = context.getBean(Processor.class);
String sentPayload = "hello " + UUID.randomUUID().toString();
processor.input().send(MessageBuilder.withPayload(sentPayload).setHeader("contentType", "text/plain").build());
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Message<?> result = messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS);
assertThat(result).isNotNull();
assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase());
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestWildCardFluxInputOutputArgsWithMessage1 {
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<?> input,
@Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> MessageBuilder.withPayload(m.toString().toUpperCase()).build()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestWildCardFluxInputOutputArgsWithMessage2 {
@StreamListener(Processor.INPUT)
public void receive(Flux<?> input, @Output(Processor.OUTPUT) FluxSender output) {
output.send(input.map(m -> MessageBuilder.withPayload(m.toString().toUpperCase()).build()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class TestWildCardFluxInputOutputArgsWithMessage3 {
@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public void receive(Flux<?> input, FluxSender output) {
output.send(input.map(m -> MessageBuilder.withPayload(m.toString().toUpperCase()).build()));
}
}
}

View File

@@ -33,12 +33,17 @@ import org.springframework.messaging.handler.annotation.MessageMapping;
* the method is invoked and how their return results are processed. This annotation
* can be applied for two separate classes of methods.
*
* <h3>Declarative mode</h3>
*
* A method is considered as declarative if its method parameters are annotated with {@link Input} and/or {@link Output}
* which have either bound elements (e.g. channels) or conversion targets from bound elements via a registered
* {@link StreamListenerParameterAdapter}. In this case, the method is invoked once when the application starts.
*
* <h3>Individual message handler mode</h3>
*
* Methods where the annotation has a value, are treated as message handlers, and are invoked for each
* Non declarative method is treated as message handler based, and is invoked for each
* incoming message received from that target. In this case, the
* method can have a flexible signature, as described by {@link MessageMapping}.
* The value must be the name of an {@link Input} bound target.
*
* If the method returns a {@link org.springframework.messaging.Message}, the result will be automatically sent
* to a channel, as follows:
@@ -53,14 +58,10 @@ import org.springframework.messaging.handler.annotation.MessageMapping;
* <li>The value set on the {@link org.springframework.messaging.handler.annotation.SendTo} annotation, if present</li>
* </ul>
*
* <h3>Declarative mode</h3>
* In both the modes, the StreamListener annotation value must be the name of an {@link Input} bound target.
*
* If the annotation has an empty value (the default), the method is a declarative
* pipeline definition and will be invoked once, when the application starts.
* All parameters must be annotated with either {@link Input} or {@link Output} and can
* be either bound elements (e.g. channels) or conversion targets from bound elements
* via a registered {@link StreamListenerParameterAdapter}.
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
* @see {@link MessageMapping}
* @see {@link EnableBinding}
* @see {@link org.springframework.messaging.handler.annotation.SendTo}

View File

@@ -16,7 +16,6 @@
package org.springframework.cloud.stream.binding;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
@@ -27,6 +26,7 @@ import org.springframework.aop.framework.Advised;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.cloud.stream.annotation.Input;
@@ -43,18 +43,27 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.core.DestinationResolver;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;
import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INPUT_AT_STREAM_LISTENER;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_DECLARATIVE_METHOD_PARAMS;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_INBOUND_NAME;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_OUTBOUND_NAME;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.RETURN_TYPE_MULTIPLE_OUTBOUND_SPECIFIED;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.RETURN_TYPE_NO_OUTBOUND_SPECIFIED;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.TARGET_BEAN_NOT_EXISTS;
import static org.springframework.cloud.stream.binding.StreamListenerMethodUtils.getInboundElementNameFromMethod;
import static org.springframework.cloud.stream.binding.StreamListenerMethodUtils.getOutboundElementNameFromMethod;
/**
* {@link BeanPostProcessor} that handles {@link StreamListener} annotations found on bean methods.
*
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
public class StreamListenerAnnotationBeanPostProcessor
implements BeanPostProcessor, ApplicationContextAware, SmartInitializingSingleton {
@@ -65,6 +74,8 @@ public class StreamListenerAnnotationBeanPostProcessor
private final Map<String, InvocableHandlerMethod> mappedBindings = new HashMap<>();
private final Map<String, Object> boundElements = new HashMap<>();
private ConfigurableApplicationContext applicationContext;
private final List<StreamListenerParameterAdapter<?, Object>> streamListenerParameterAdapters = new ArrayList<>();
@@ -109,40 +120,27 @@ public class StreamListenerAnnotationBeanPostProcessor
public void doWith(final Method method) throws IllegalArgumentException, IllegalAccessException {
StreamListener streamListener = AnnotationUtils.findAnnotation(method, StreamListener.class);
if (streamListener != null) {
if (!method.getReturnType().equals(Void.TYPE)) {
Assert.isTrue(method.getAnnotation(Input.class) == null,
"A @StreamListener may never be annotated with @Input." +
"If it should listen to a specific input, use the value of @StreamListener " +
"instead.");
}
Assert.isTrue(method.getAnnotation(Input.class) == null, INPUT_AT_STREAM_LISTENER);
String methodAnnotatedInboundName = getInboundElementNameFromMethod(streamListener);
String methodAnnotatedOutboundName = getOutboundElementNameFromMethod(method);
int inputAnnotationCount = StreamListenerMethodUtils.inputAnnotationCount(method);
int outputAnnotationCount = StreamListenerMethodUtils.outputAnnotationCount(method);
boolean isDeclarative = isDeclarativeStreamListenerMethod(method, methodAnnotatedInboundName, methodAnnotatedOutboundName);
StreamListenerMethodUtils.assertStreamListenerMethod(method, inputAnnotationCount, outputAnnotationCount, methodAnnotatedInboundName, methodAnnotatedOutboundName, isDeclarative);
Class<?>[] parameterTypes = method.getParameterTypes();
if (StringUtils.hasText(streamListener.value())) {
for (int i = 0; i < parameterTypes.length; i++) {
MethodParameter methodParameter = MethodParameter.forMethodOrConstructor(method, i);
Assert.isTrue(methodParameter.getParameterAnnotation(Input.class) == null &&
methodParameter.getParameterAnnotation(Output.class) == null,
"A message handling @StreamListener method cannot have parameters annotated " +
"with @Input or @Output");
if (!method.getReturnType().equals(Void.TYPE)) {
if (!StringUtils.hasText(methodAnnotatedOutboundName)) {
if (outputAnnotationCount == 0) {
throw new IllegalArgumentException(RETURN_TYPE_NO_OUTBOUND_SPECIFIED);
}
Assert.isTrue((outputAnnotationCount == 1), RETURN_TYPE_MULTIPLE_OUTBOUND_SPECIFIED);
}
if (!method.getReturnType().equals(Void.TYPE)) {
Assert.isTrue(method.getAnnotation(Output.class) == null,
"A message handling @StreamListener method cannot be annotated with @Output");
}
registerHandlerMethodOnListenedChannel(method, streamListener, bean);
}
if (isDeclarative && (!StringUtils.hasText(methodAnnotatedInboundName) || parameterTypes.length == 1)) {
invokeSetupMethodOnListenedChannel(method, bean, methodAnnotatedInboundName, methodAnnotatedOutboundName);
}
else {
for (int i = 0; i < parameterTypes.length; i++) {
MethodParameter methodParameter = MethodParameter.forMethodOrConstructor(method, i);
Assert.isTrue(methodParameter.getParameterAnnotation(Input.class) != null ^
methodParameter.getParameterAnnotation(Output.class) != null,
"A declarative @StreamListener method must have its parameters annotated" +
"with @Input or @Output, but not with both.");
}
if (!method.getReturnType().equals(Void.TYPE)) {
Assert.isTrue(method.getAnnotation(Output.class) != null,
"A declarative @StreamListener method must be annotated with @Output");
}
invokeSetupMethodOnListenedChannel(method, bean);
registerHandlerMethodOnListenedChannel(method, streamListener, bean);
}
}
}
@@ -150,34 +148,94 @@ public class StreamListenerAnnotationBeanPostProcessor
return bean;
}
private boolean isDeclarativeStreamListenerMethod(Method method, String methodAnnotatedInboundName, String methodAnnotatedOutboundName) {
int methodArgumentsLength = method.getParameterTypes().length;
for (int parameterIndex = 0; parameterIndex < methodArgumentsLength; parameterIndex++) {
MethodParameter methodParameter = MethodParameter.forMethodOrConstructor(method, parameterIndex);
if (methodParameter.hasParameterAnnotation(Input.class)) {
String inboundName = (String) AnnotationUtils.getValue(methodParameter.getParameterAnnotation(Input.class));
Assert.isTrue(StringUtils.hasText(inboundName), INVALID_INBOUND_NAME);
return isDeclarativeMethodParameter(getBindableBean(inboundName), methodParameter);
}
if (methodParameter.hasParameterAnnotation(Output.class)) {
String outboundName = (String) AnnotationUtils.getValue(methodParameter.getParameterAnnotation(Output.class));
Assert.isTrue(StringUtils.hasText(outboundName), INVALID_OUTBOUND_NAME);
return isDeclarativeMethodParameter(getBindableBean(outboundName), methodParameter);
}
if (StringUtils.hasText(methodAnnotatedOutboundName)) {
return isDeclarativeMethodParameter(getBindableBean(methodAnnotatedOutboundName), methodParameter);
}
if (StringUtils.hasText(methodAnnotatedInboundName)) {
return isDeclarativeMethodParameter(getBindableBean(methodAnnotatedInboundName), methodParameter);
}
}
return false;
}
private boolean isDeclarativeMethodParameter(Object targetBean, MethodParameter methodParameter) {
if (targetBean != null) {
if (methodParameter.getParameterType().isAssignableFrom(targetBean.getClass())) {
return true;
}
for (StreamListenerParameterAdapter<?, Object> streamListenerParameterAdapter : this.streamListenerParameterAdapters) {
if (streamListenerParameterAdapter.supports(targetBean.getClass(), methodParameter)) {
return true;
}
}
}
return false;
}
private Object getBindableBean(String boundElementName) {
try {
if (!this.boundElements.containsKey(boundElementName)) {
this.boundElements.put(boundElementName, this.applicationContext.getBean(boundElementName));
}
return this.boundElements.get(boundElementName);
}
catch (NoSuchBeanDefinitionException e) {
throw new IllegalStateException(TARGET_BEAN_NOT_EXISTS + ": " + boundElementName, e);
}
}
@SuppressWarnings({"rawtypes", "unchecked"})
private void invokeSetupMethodOnListenedChannel(Method method, Object bean) {
private void invokeSetupMethodOnListenedChannel(Method method, Object bean, String inboundName, String outboundName) {
Object[] arguments = new Object[method.getParameterTypes().length];
for (int parameterIndex = 0; parameterIndex < arguments.length; parameterIndex++) {
MethodParameter methodParameter = MethodParameter.forMethodOrConstructor(method, parameterIndex);
Class<?> parameterType = methodParameter.getParameterType();
Annotation targetReferenceAnnotation = methodParameter.hasParameterAnnotation(Input.class) ?
methodParameter.getParameterAnnotation(Input.class) : methodParameter.getParameterAnnotation(
Output.class);
Object targetReferenceAnnotationValue = AnnotationUtils.getValue(targetReferenceAnnotation);
Assert.isInstanceOf(String.class, targetReferenceAnnotationValue, "Annotation value must be a String");
Assert.hasText((String) targetReferenceAnnotationValue, "Annotation value must not be empty");
Object targetBean = this.applicationContext.getBean((String) targetReferenceAnnotationValue);
if (parameterType.isAssignableFrom(targetBean.getClass())) {
arguments[parameterIndex] = targetBean;
Object targetReferenceValue = null;
if (methodParameter.hasParameterAnnotation(Input.class)) {
targetReferenceValue = AnnotationUtils.getValue(methodParameter.getParameterAnnotation(Input.class));
}
else {
for (StreamListenerParameterAdapter<?, Object> streamListenerParameterAdapter :
this.streamListenerParameterAdapters) {
if (streamListenerParameterAdapter.supports(targetBean.getClass(), methodParameter)) {
arguments[parameterIndex] = streamListenerParameterAdapter.adapt(targetBean, methodParameter);
break;
else if (methodParameter.hasParameterAnnotation(Output.class)) {
targetReferenceValue = AnnotationUtils.getValue(methodParameter.getParameterAnnotation(Output.class));
}
else if (arguments.length == 1 && StringUtils.hasText(inboundName)) {
targetReferenceValue = inboundName;
}
if (targetReferenceValue != null) {
Assert.isInstanceOf(String.class, targetReferenceValue, "Annotation value must be a String");
Object targetBean = getBindableBean((String) targetReferenceValue);
if (parameterType.isAssignableFrom(targetBean.getClass())) {
arguments[parameterIndex] = targetBean;
}
else {
for (StreamListenerParameterAdapter<?, Object> streamListenerParameterAdapter :
this.streamListenerParameterAdapters) {
if (streamListenerParameterAdapter.supports(targetBean.getClass(), methodParameter)) {
arguments[parameterIndex] = streamListenerParameterAdapter.adapt(targetBean, methodParameter);
break;
}
}
}
Assert.notNull(arguments[parameterIndex],
"Cannot convert argument " + parameterIndex + " of " + method + "from " + targetBean.getClass()
+ " to " + parameterType);
}
else {
throw new IllegalStateException(INVALID_DECLARATIVE_METHOD_PARAMS);
}
Assert.notNull(arguments[parameterIndex],
"Cannot convert argument " + parameterIndex + " of " + method + "from " + targetBean.getClass()
.toString() + " to " + parameterType.toString());
}
try {
if (method.getReturnType().equals(Void.TYPE)) {
@@ -185,8 +243,15 @@ public class StreamListenerAnnotationBeanPostProcessor
}
else {
Object result = method.invoke(bean, arguments);
Output output = AnnotationUtils.getAnnotation(method, Output.class);
Object targetBean = this.applicationContext.getBean(output.value());
if (!StringUtils.hasText(outboundName)) {
for (int parameterIndex = 0; parameterIndex < method.getParameterTypes().length; parameterIndex++) {
MethodParameter methodParameter = MethodParameter.forMethodOrConstructor(method, parameterIndex);
if (methodParameter.hasParameterAnnotation(Output.class)) {
outboundName = methodParameter.getParameterAnnotation(Output.class).value();
}
}
}
Object targetBean = this.applicationContext.getBean(outboundName);
for (StreamListenerResultAdapter streamListenerResultAdapter : this
.streamListenerResultAdapters) {
if (streamListenerResultAdapter.supports(result.getClass(), targetBean.getClass())) {
@@ -217,7 +282,7 @@ public class StreamListenerAnnotationBeanPostProcessor
this.mappedBindings.put(streamListener.value(), invocableHandlerMethod);
SubscribableChannel channel = this.applicationContext.getBean(streamListener.value(),
SubscribableChannel.class);
final String defaultOutputChannel = extractDefaultOutput(method);
final String defaultOutputChannel = getOutboundElementNameFromMethod(method);
if (invocableHandlerMethod.isVoid()) {
Assert.isTrue(StringUtils.isEmpty(defaultOutputChannel),
"An output channel cannot be specified for a method that " +
@@ -228,6 +293,7 @@ public class StreamListenerAnnotationBeanPostProcessor
"An output channel must be specified for a method that " +
"can return a value");
}
StreamListenerMethodUtils.assertStreamListenerMessageHandlerMethod(method);
StreamListenerMessageHandler handler = new StreamListenerMessageHandler(invocableHandlerMethod);
handler.setApplicationContext(this.applicationContext);
handler.setChannelResolver(this.binderAwareChannelResolver);
@@ -243,17 +309,7 @@ public class StreamListenerAnnotationBeanPostProcessor
// Dump the mappings after the context has been created, ensuring that beans can be processed correctly
// again.
this.mappedBindings.clear();
}
private String extractDefaultOutput(Method method) {
SendTo sendTo = AnnotationUtils.findAnnotation(method, SendTo.class);
if (sendTo != null) {
Assert.isTrue(!ObjectUtils.isEmpty(sendTo.value()), "At least one output must be specified");
Assert.isTrue(sendTo.value().length == 1, "Multiple destinations cannot be specified");
Assert.hasText(sendTo.value()[0], "An empty destination cannot be specified");
return sendTo.value()[0];
}
return null;
this.boundElements.clear();
}
private Method checkProxy(Method methodArg, Object bean) {

View File

@@ -0,0 +1,64 @@
/*
* 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 org.springframework.cloud.stream.binding;
/**
* @author Ilayaperumal Gopinathan
*/
public interface StreamListenerErrorMessages {
public static final String INPUT_AT_STREAM_LISTENER = "A @StreamListener may never be annotated with @Input. If it should listen to a specific input, " +
"use the value of @StreamListener instead.";
public static final String RETURN_TYPE_NO_OUTBOUND_SPECIFIED = "StreamListener method with return type should have outbound target specified";
public static final String RETURN_TYPE_MULTIPLE_OUTBOUND_SPECIFIED = "StreamListener method with return type should have only one outbound target specified";
public static final String INVALID_INBOUND_NAME = "@Input annotation should always be associated with a valid inbound name";
public static final String INVALID_OUTBOUND_NAME = "@Output annotation should always be associated with a valid outbound name";
public static final String ATLEAST_ONE_OUTPUT = "At least one output must be specified";
public static final String SEND_TO_MULTIPLE_DESTINATIONS = "Multiple destinations cannot be specified";
public static final String SEND_TO_EMPTY_DESTINATION = "An empty destination cannot be specified";
public static final String INVALID_MESSAGE_HANDLER_METHOD_PARAMS = "@Input or @Output annotation is not supported as method parameter in StreamListener method with " +
"message handler mapping";
public static final String INVALID_INPUT_OUTPUT_METHOD_PARAMETERS = "@Input or @Output annotations are not permitted as " +
"method parameters when both inbound and outbound values are set as method annotated values";
public static final String NO_INPUT_DESTINATION = "No input destination is configured. Use either a @StreamListener attribute or @Input";
public static final String INVALID_DECLARATIVE_METHOD_PARAMETERS = "Declarative StreamListener method should only have inbound or outbound targets as method parameters";
public static final String AMBIGUOUS_MESSAGE_HANDLER_METHOD_ARGUMENTS = "Ambiguous method arguments for the StreamListener method";
public static final String INVALID_INPUT_VALUES = "Cannot set both StreamListener attribute and @Input annotation as method parameter";
public static final String INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM = "Cannot set StreamListener attribute when using" +
" @Output annotation as method parameter. Use @Input method parameter annotation to specify inbound value instead";
public static final String INVALID_OUTPUT_VALUES = "Cannot set both Output (@Output/@SendTo) method annotation value" +
" and @Output annotation as a method parameter";
public static final String INVALID_DECLARATIVE_METHOD_PARAMS = "Declarative StreamListener method should only have inbound or outbound targets as method parameters";
public static final String TARGET_BEAN_NOT_EXISTS = "Target bean doesn't exist for the bound element name";
}

View File

@@ -0,0 +1,151 @@
/*
* 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 org.springframework.cloud.stream.binding;
import java.lang.reflect.Method;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.core.MethodParameter;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.AMBIGUOUS_MESSAGE_HANDLER_METHOD_ARGUMENTS;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.ATLEAST_ONE_OUTPUT;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_DECLARATIVE_METHOD_PARAMETERS;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_INBOUND_NAME;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_INPUT_OUTPUT_METHOD_PARAMETERS;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_INPUT_VALUES;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_MESSAGE_HANDLER_METHOD_PARAMS;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_OUTBOUND_NAME;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_OUTPUT_VALUES;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.NO_INPUT_DESTINATION;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.SEND_TO_EMPTY_DESTINATION;
import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.SEND_TO_MULTIPLE_DESTINATIONS;
/**
* This class contains utility methods for handling {@link StreamListener} annotated bean methods.
*
* @author Ilayaperumal Gopinathan
*/
public class StreamListenerMethodUtils {
protected static int inputAnnotationCount(Method method) {
int inputAnnotationCount = 0;
for (int parameterIndex = 0; parameterIndex < method.getParameterTypes().length; parameterIndex++) {
MethodParameter methodParameter = MethodParameter.forMethodOrConstructor(method, parameterIndex);
if (methodParameter.hasParameterAnnotation(Input.class)) {
inputAnnotationCount++;
}
}
return inputAnnotationCount;
}
protected static int outputAnnotationCount(Method method) {
int outputAnnotationCount = 0;
for (int parameterIndex = 0; parameterIndex < method.getParameterTypes().length; parameterIndex++) {
MethodParameter methodParameter = MethodParameter.forMethodOrConstructor(method, parameterIndex);
if (methodParameter.hasParameterAnnotation(Output.class)) {
outputAnnotationCount++;
}
}
return outputAnnotationCount;
}
protected static void assertStreamListenerMethod(Method method, int inputAnnotationCount, int outputAnnotationCount,
String methodAnnotatedInboundName, String methodAnnotatedOutboundName, boolean isDeclarative) {
int methodArgumentsLength = method.getParameterTypes().length;
if (!isDeclarative) {
Assert.isTrue(inputAnnotationCount == 0 && outputAnnotationCount == 0, INVALID_MESSAGE_HANDLER_METHOD_PARAMS);
}
if (StringUtils.hasText(methodAnnotatedInboundName) && StringUtils.hasText(methodAnnotatedOutboundName)) {
Assert.isTrue(inputAnnotationCount == 0 && outputAnnotationCount == 0, INVALID_INPUT_OUTPUT_METHOD_PARAMETERS);
}
if (StringUtils.hasText(methodAnnotatedInboundName)) {
Assert.isTrue(inputAnnotationCount == 0, INVALID_INPUT_VALUES);
Assert.isTrue(outputAnnotationCount == 0, INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM);
}
else {
Assert.isTrue(inputAnnotationCount >= 1, NO_INPUT_DESTINATION);
}
if (StringUtils.hasText(methodAnnotatedOutboundName)) {
Assert.isTrue(outputAnnotationCount == 0, INVALID_OUTPUT_VALUES);
}
if (isDeclarative) {
for (int parameterIndex = 0; parameterIndex < methodArgumentsLength; parameterIndex++) {
MethodParameter methodParameter = MethodParameter.forMethodOrConstructor(method, parameterIndex);
if (methodParameter.hasParameterAnnotation(Input.class)) {
String inboundName = (String) AnnotationUtils.getValue(methodParameter.getParameterAnnotation(Input.class));
Assert.isTrue(StringUtils.hasText(inboundName), INVALID_INBOUND_NAME);
}
if (methodParameter.hasParameterAnnotation(Output.class)) {
String outboundName = (String) AnnotationUtils.getValue(methodParameter.getParameterAnnotation(Output.class));
Assert.isTrue(StringUtils.hasText(outboundName), INVALID_OUTBOUND_NAME);
}
}
if (methodArgumentsLength > 1){
Assert.isTrue(inputAnnotationCount + outputAnnotationCount == methodArgumentsLength, INVALID_DECLARATIVE_METHOD_PARAMETERS);
}
}
}
protected static void assertStreamListenerMessageHandlerMethod(Method method) {
int methodArgumentsLength = method.getParameterTypes().length;
if (methodArgumentsLength > 1) {
int numAnnotatedMethodParameters = 0;
int numPayloadAnnotations = 0;
for (int parameterIndex = 0; parameterIndex < methodArgumentsLength; parameterIndex++) {
MethodParameter methodParameter = MethodParameter.forMethodOrConstructor(method, parameterIndex);
if (methodParameter.hasParameterAnnotations()) {
numAnnotatedMethodParameters++;
}
if (methodParameter.hasParameterAnnotation(Payload.class)) {
numPayloadAnnotations++;
}
}
if (numPayloadAnnotations > 0) {
Assert.isTrue(methodArgumentsLength == numAnnotatedMethodParameters && numPayloadAnnotations <= 1,
AMBIGUOUS_MESSAGE_HANDLER_METHOD_ARGUMENTS);
}
}
}
protected static String getInboundElementNameFromMethod(StreamListener streamListener) {
return StringUtils.hasText(streamListener.value()) ? streamListener.value() : null;
}
protected static String getOutboundElementNameFromMethod(Method method) {
SendTo sendTo = AnnotationUtils.findAnnotation(method, SendTo.class);
if (sendTo != null) {
Assert.isTrue(!ObjectUtils.isEmpty(sendTo.value()), ATLEAST_ONE_OUTPUT);
Assert.isTrue(sendTo.value().length == 1, SEND_TO_MULTIPLE_DESTINATIONS);
Assert.hasText(sendTo.value()[0], SEND_TO_EMPTY_DESTINATION);
return sendTo.value()[0];
}
Output output = AnnotationUtils.findAnnotation(method, Output.class);
if (output != null) {
Assert.isTrue(StringUtils.hasText(output.value()), ATLEAST_ONE_OUTPUT);
return output.value();
}
return null;
}
}