Add (javax.)Validation Support for Stream Listener

This change allows StreamListener method arguments annotated with @Valid to be
validated against a registered Validator.

Resolves #1382

-=david=-

spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAnnotatedMethodArgumentsTests.java

@@ -36,8 +36,11 @@ 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.support.MethodArgumentNotValidException;
 import org.springframework.util.MimeType;
 
+import javax.validation.Valid;
+
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.fail;
 import static org.springframework.cloud.stream.binding.StreamListenerErrorMessages.INVALID_DECLARATIVE_METHOD_PARAMETERS;
@@ -86,6 +89,36 @@ public class StreamListenerAnnotatedMethodArgumentsTests {
 		}
 	}
 
+	@Test
+	public void testValidAnnotationAtMethodParameterWithPojoThatPassesValidation() {
+		ConfigurableApplicationContext context = SpringApplication.run(TestPojoWithValidAnnotationThatPassesValidation.class,
+				"--server.port=0");
+
+		TestPojoWithValidAnnotationThatPassesValidation testPojoWithValidAnnotationThatPassesValidation = context.getBean(TestPojoWithValidAnnotationThatPassesValidation.class);
+		Sink sink = context.getBean(Sink.class);
+		String id = UUID.randomUUID().toString();
+		sink.input().send(MessageBuilder.withPayload("{\"foo\":\"" + id + "\"}")
+				.setHeader("contentType", MimeType.valueOf("application/json")).build());
+		assertThat(testPojoWithValidAnnotationThatPassesValidation.receivedArguments.get(0)).hasFieldOrPropertyWithValue("foo", id);
+		context.close();
+	}
+
+	@Test
+	public void testValidAnnotationAtMethodParameterWithPojoThatFailsValidation() {
+		ConfigurableApplicationContext context = SpringApplication.run(TestPojoWithValidAnnotationThatPassesValidation.class,
+				"--server.port=0");
+
+		Sink sink = context.getBean(Sink.class);
+		try {
+			sink.input().send(MessageBuilder.withPayload("{\"foo\":\"\"}")
+					.setHeader("contentType", MimeType.valueOf("application/json")).build());
+			fail("Exception expected: MethodArgumentNotValidException!");
+		} catch(MethodArgumentNotValidException e) {
+			assertThat(e.getMessage()).contains("default message [foo]]; default message [must not be blank]]");
+		}
+		context.close();
+	}
+
 	@EnableBinding(Processor.class)
 	@EnableAutoConfiguration
 	public static class TestPojoWithAnnotatedArguments {
@@ -118,4 +151,17 @@ public class StreamListenerAnnotatedMethodArgumentsTests {
 			this.receivedArguments.add(contentType);
 		}
 	}
+
+	@EnableBinding(Processor.class)
+	@EnableAutoConfiguration
+	public static class TestPojoWithValidAnnotationThatPassesValidation {
+
+		List<Object> receivedArguments = new ArrayList<>();
+
+		@StreamListener(Processor.INPUT)
+		public void receive(@Valid StreamListenerTestUtils.PojoWithValidation pojoWithValidation) {
+			this.receivedArguments.add(pojoWithValidation);
+		}
+	}
+
 }

spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerHandlerMethodTests.java

@@ -46,6 +46,8 @@ import org.springframework.messaging.handler.annotation.Payload;
 import org.springframework.messaging.handler.annotation.SendTo;
 import org.springframework.util.Assert;
 
+import javax.validation.Valid;
+
 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;
@@ -578,4 +580,5 @@ public class StreamListenerHandlerMethodTests {
 			});
 		}
 	}
+
 }

spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerTestUtils.java

@@ -21,6 +21,8 @@ import org.springframework.cloud.stream.annotation.Output;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.SubscribableChannel;
 
+import javax.validation.constraints.NotBlank;
+
 /**
  * @author Ilayaperumal Gopinathan
  */
@@ -85,4 +87,16 @@ public class StreamListenerTestUtils {
 			return sb.toString();
 		}
 	}
+
+	public static class PojoWithValidation {
+
+		@NotBlank
+		private String foo;
+
+		public String getFoo() { return this.foo; }
+
+		public void setFoo(String foo) { this.foo = foo; }
+
+	}
+
 }

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java

@@ -57,12 +57,14 @@ import org.springframework.messaging.handler.annotation.support.DefaultMessageHa
 import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.StringUtils;
+import org.springframework.validation.Validator;
 
 /**
  * @author Marius Bogoevici
  * @author Ilayaperumal Gopinathan
  * @author Oleg Zhurakousky
  * @author Soby Chacko
+ * @author David Harrigan
  *
  * @deprecated since it really represents 'auto-configuration' it will be renamed/restructured in the next release.
  */
@@ -157,10 +159,12 @@ public class BinderFactoryConfiguration {
 
 	@Bean
 	public static MessageHandlerMethodFactory messageHandlerMethodFactory(CompositeMessageConverterFactory compositeMessageConverterFactory,
-																		@Qualifier(IntegrationContextUtils.ARGUMENT_RESOLVERS_BEAN_NAME) HandlerMethodArgumentResolversHolder ahmar) {
+																		@Qualifier(IntegrationContextUtils.ARGUMENT_RESOLVERS_BEAN_NAME) HandlerMethodArgumentResolversHolder ahmar,
+																		Validator validator) {
 		DefaultMessageHandlerMethodFactory messageHandlerMethodFactory = new DefaultMessageHandlerMethodFactory();
 		messageHandlerMethodFactory.setMessageConverter(compositeMessageConverterFactory.getMessageConverterForAllRegistered());
 		messageHandlerMethodFactory.setCustomArgumentResolvers(ahmar.getResolvers());
+		messageHandlerMethodFactory.setValidator(validator);
 		return messageHandlerMethodFactory;
 	}