Restructuring the repository

core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/TextPlainToJsonConversionTest.java.todo

@@ -0,0 +1,123 @@
+///*
+// * Copyright 2017-2018 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
+// *
+// *      https://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 com.fasterxml.jackson.databind.ObjectMapper;
+//import org.junit.Test;
+//import org.junit.runner.RunWith;
+//
+//import org.springframework.beans.factory.annotation.Autowired;
+//import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+//import org.springframework.boot.test.context.SpringBootTest;
+//import org.springframework.cloud.stream.annotation.EnableBinding;
+//import org.springframework.cloud.stream.annotation.StreamListener;
+//import org.springframework.cloud.stream.binder.BinderFactory;
+//import org.springframework.cloud.stream.messaging.Processor;
+//import org.springframework.cloud.stream.test.binder.TestSupportBinder;
+//import org.springframework.integration.support.MessageBuilder;
+//import org.springframework.messaging.Message;
+//import org.springframework.messaging.MessageChannel;
+//import org.springframework.messaging.MessageHeaders;
+//import org.springframework.messaging.MessagingException;
+//import org.springframework.messaging.handler.annotation.SendTo;
+//import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+//
+//import static org.assertj.core.api.Assertions.assertThat;
+//
+///**
+// * @author Marius Bogoevici
+// * @author Vinicius Carvalho
+// * @author Oleg Zhurakousky
+// * @since 1.2
+// */
+//@RunWith(SpringJUnit4ClassRunner.class)
+//// @checkstyle:off
+//@SpringBootTest(classes = TextPlainToJsonConversionTest.FooProcessor.class, webEnvironment = SpringBootTest.WebEnvironment.NONE)
+//// @checkstyle:on
+//public class TextPlainToJsonConversionTest {
+//
+//	@Autowired
+//	private Processor testProcessor;
+//
+//	@Autowired
+//	private BinderFactory binderFactory;
+//
+//	private ObjectMapper mapper = new ObjectMapper();
+//
+//	@SuppressWarnings("unchecked")
+//	@Test
+//	public void testNoContentTypeToJsonConversionOnInput() throws Exception {
+//		this.testProcessor.input()
+//				.send(MessageBuilder.withPayload("{\"name\":\"Bar\"}").build());
+//		Message<String> received = (Message<String>) ((TestSupportBinder) this.binderFactory
+//				.getBinder(null, MessageChannel.class)).messageCollector()
+//						.forChannel(this.testProcessor.output())
+//						.poll(1, TimeUnit.SECONDS);
+//		assertThat(received).isNotNull();
+//		Foo foo = this.mapper.readValue(received.getPayload(), Foo.class);
+//		assertThat(foo.getName()).isEqualTo("transformed-Bar");
+//	}
+//
+//	/**
+//	 * @since 2.0: Conversion from text/plain -> json is no longer supported. Strict
+//	 * contentType only.
+//	 */
+//	@Test(expected = MessagingException.class)
+//	public void testTextPlainToJsonConversionOnInput() {
+//		this.testProcessor.input().send(MessageBuilder.withPayload("{\"name\":\"Bar\"}")
+//				.setHeader(MessageHeaders.CONTENT_TYPE, "text/plain").build());
+//	}
+//
+//	@EnableBinding(Processor.class)
+//	@EnableAutoConfiguration
+//	public static class FooProcessor {
+//
+//		@StreamListener("input")
+//		@SendTo("output")
+//		public Foo consume(Foo foo) {
+//			Foo returnFoo = new Foo();
+//			returnFoo.setName("transformed-" + foo.getName());
+//			return returnFoo;
+//		}
+//
+//	}
+//
+//	public static class Foo {
+//
+//		private String name;
+//
+//		public Foo() {
+//		}
+//
+//		public String getName() {
+//			return this.name;
+//		}
+//
+//		public void setName(String name) {
+//			this.name = name;
+//		}
+//
+//		@Override
+//		public String toString() {
+//			return "Foo{name='" + this.name + "'}";
+//		}
+//
+//	}
+//
+//}

core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/contentType/ContentTypeTests.java.todo

@@ -0,0 +1,272 @@
+///*
+// * Copyright 2017-2019 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
+// *
+// *      https://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.contentType;
+//
+//import java.util.LinkedList;
+//import java.util.Map;
+//import java.util.concurrent.TimeUnit;
+//
+//import com.fasterxml.jackson.databind.ObjectMapper;
+//import org.junit.Test;
+//
+//import org.springframework.boot.SpringApplication;
+//import org.springframework.boot.autoconfigure.SpringBootApplication;
+//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.Source;
+//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.SubscribableChannel;
+//import org.springframework.messaging.handler.annotation.Headers;
+//import org.springframework.util.MimeType;
+//import org.springframework.util.MimeTypeUtils;
+//
+//import static org.assertj.core.api.Assertions.assertThat;
+//
+///**
+// * @author Vinicius Carvalho
+// * @author Oleg Zhurakousky
+// */
+//@SuppressWarnings("unchecked")
+//public class ContentTypeTests {
+//
+//	private ObjectMapper mapper = new ObjectMapper();
+//
+//	@Test
+//	public void testSendWithDefaultContentType() throws Exception {
+//		try (ConfigurableApplicationContext context = SpringApplication.run(
+//				SourceApplication.class, "--server.port=0",
+//				"--spring.jmx.enabled=false")) {
+//
+//			MessageCollector collector = context.getBean(MessageCollector.class);
+//			Source source = context.getBean(Source.class);
+//			User user = new User("Alice");
+//			source.output().send(MessageBuilder.withPayload(user).build());
+//			Message<String> message = (Message<String>) collector
+//					.forChannel(source.output()).poll(1, TimeUnit.SECONDS);
+//			User received = this.mapper.readValue(message.getPayload(), User.class);
+//			assertThat(
+//					message.getHeaders().get(MessageHeaders.CONTENT_TYPE, MimeType.class)
+//							.includes(MimeTypeUtils.APPLICATION_JSON));
+//			assertThat(user.getName()).isEqualTo(received.getName());
+//		}
+//	}
+//
+//	@Test
+//	public void testSendJsonAsString() throws Exception {
+//		try (ConfigurableApplicationContext context = SpringApplication.run(
+//				SourceApplication.class, "--server.port=0",
+//				"--spring.jmx.enabled=false")) {
+//			MessageCollector collector = context.getBean(MessageCollector.class);
+//			Source source = context.getBean(Source.class);
+//			User user = new User("Alice");
+//			String json = this.mapper.writeValueAsString(user);
+//			source.output().send(MessageBuilder.withPayload(user).build());
+//			Message<String> message = (Message<String>) collector
+//					.forChannel(source.output()).poll(1, TimeUnit.SECONDS);
+//			assertThat(
+//					message.getHeaders().get(MessageHeaders.CONTENT_TYPE, MimeType.class)
+//							.includes(MimeTypeUtils.APPLICATION_JSON));
+//			assertThat(json).isEqualTo(message.getPayload());
+//		}
+//	}
+//
+//	@Test
+//	public void testSendJsonString() throws Exception {
+//		try (ConfigurableApplicationContext context = SpringApplication.run(
+//				SourceApplication.class, "--server.port=0",
+//				"--spring.jmx.enabled=false")) {
+//			MessageCollector collector = context.getBean(MessageCollector.class);
+//			Source source = context.getBean(Source.class);
+//			source.output().send(MessageBuilder.withPayload("foo").build());
+//			Message<String> message = (Message<String>) collector
+//					.forChannel(source.output()).poll(1, TimeUnit.SECONDS);
+//			assertThat(
+//					message.getHeaders().get(MessageHeaders.CONTENT_TYPE, MimeType.class)
+//							.includes(MimeTypeUtils.APPLICATION_JSON));
+//			assertThat("foo").isEqualTo(message.getPayload());
+//		}
+//	}
+//
+//	@Test
+//	public void testSendBynaryData() throws Exception {
+//		try (ConfigurableApplicationContext context = SpringApplication.run(
+//				SourceApplication.class, "--server.port=0",
+//				"--spring.jmx.enabled=false")) {
+//
+//			MessageCollector collector = context.getBean(MessageCollector.class);
+//			Source source = context.getBean(Source.class);
+//			byte[] data = new byte[] { 0, 1, 2, 3 };
+//			source.output()
+//					.send(MessageBuilder.withPayload(data)
+//							.setHeader(MessageHeaders.CONTENT_TYPE,
+//									MimeTypeUtils.APPLICATION_OCTET_STREAM)
+//							.build());
+//			Message<byte[]> message = (Message<byte[]>) collector
+//					.forChannel(source.output()).poll(1, TimeUnit.SECONDS);
+//			assertThat(
+//					message.getHeaders().get(MessageHeaders.CONTENT_TYPE, MimeType.class)
+//							.includes(MimeTypeUtils.APPLICATION_OCTET_STREAM));
+//			assertThat(message.getPayload()).isEqualTo(data);
+//		}
+//	}
+//
+//	@Test
+//	public void testSendBinaryDataWithContentType() throws Exception {
+//		try (ConfigurableApplicationContext context = SpringApplication.run(
+//				SourceApplication.class, "--server.port=0", "--spring.jmx.enabled=false",
+//				"--spring.cloud.stream.bindings.output.contentType=image/jpeg")) {
+//			MessageCollector collector = context.getBean(MessageCollector.class);
+//			Source source = context.getBean(Source.class);
+//			byte[] data = new byte[] { 0, 1, 2, 3 };
+//			source.output().send(MessageBuilder.withPayload(data).build());
+//			Message<byte[]> message = (Message<byte[]>) collector
+//					.forChannel(source.output()).poll(1, TimeUnit.SECONDS);
+//			assertThat(message.getPayload()).isEqualTo(data);
+//		}
+//	}
+//
+//	@Test
+//	public void testSendBinaryDataWithContentTypeUsingHeaders() throws Exception {
+//		try (ConfigurableApplicationContext context = SpringApplication.run(
+//				SourceApplication.class, "--server.port=0",
+//				"--spring.jmx.enabled=false")) {
+//			MessageCollector collector = context.getBean(MessageCollector.class);
+//			Source source = context.getBean(Source.class);
+//			byte[] data = new byte[] { 0, 1, 2, 3 };
+//			source.output().send(MessageBuilder.withPayload(data)
+//					.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.IMAGE_JPEG)
+//					.build());
+//			Message<byte[]> message = (Message<byte[]>) collector
+//					.forChannel(source.output()).poll(1, TimeUnit.SECONDS);
+//			assertThat(
+//					message.getHeaders().get(MessageHeaders.CONTENT_TYPE, MimeType.class)
+//							.includes(MimeTypeUtils.IMAGE_JPEG));
+//			assertThat(message.getPayload()).isEqualTo(data);
+//		}
+//	}
+//
+//	@Test
+//	public void testSendStringType() throws Exception {
+//		try (ConfigurableApplicationContext context = SpringApplication.run(
+//				SourceApplication.class, "--server.port=0", "--spring.jmx.enabled=false",
+//				"--spring.cloud.stream.bindings.output.contentType=text/plain")) {
+//			MessageCollector collector = context.getBean(MessageCollector.class);
+//			Source source = context.getBean(Source.class);
+//			User user = new User("Alice");
+//			source.output().send(MessageBuilder.withPayload(user).build());
+//			Message<String> message = (Message<String>) collector
+//					.forChannel(source.output()).poll(1, TimeUnit.SECONDS);
+//			assertThat(
+//					message.getHeaders().get(MessageHeaders.CONTENT_TYPE, MimeType.class)
+//							.includes(MimeTypeUtils.TEXT_PLAIN));
+//			assertThat(message.getPayload()).isEqualTo(user.toString());
+//		}
+//	}
+//
+//	@Test
+//	public void testReceiveWithDefaults() throws Exception {
+//		try (ConfigurableApplicationContext context = SpringApplication.run(
+//				SinkApplication.class, "--server.port=0", "--spring.jmx.enabled=false")) {
+//			TestSink testSink = context.getBean(TestSink.class);
+//			SinkApplication sourceApp = context.getBean(SinkApplication.class);
+//			User user = new User("Alice");
+//			testSink.pojo().send(MessageBuilder
+//					.withPayload(this.mapper.writeValueAsBytes(user)).build());
+//			Map<String, Object> headers = (Map<String, Object>) sourceApp.arguments.pop();
+//			User received = (User) sourceApp.arguments.pop();
+//			assertThat(((MimeType) headers.get(MessageHeaders.CONTENT_TYPE))
+//					.includes(MimeTypeUtils.APPLICATION_JSON));
+//			assertThat(user.getName()).isEqualTo(received.getName());
+//		}
+//	}
+//
+//	@Test
+//	public void testReceiveRawWithDifferentContentTypes() {
+//		try (ConfigurableApplicationContext context = SpringApplication.run(
+//				SinkApplication.class, "--server.port=0", "--spring.jmx.enabled=false")) {
+//			TestSink testSink = context.getBean(TestSink.class);
+//			SinkApplication sourceApp = context.getBean(SinkApplication.class);
+//			testSink.raw().send(MessageBuilder.withPayload(new byte[4])
+//					.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.IMAGE_JPEG)
+//					.build());
+//			testSink.raw().send(MessageBuilder.withPayload(new byte[4])
+//					.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.IMAGE_GIF)
+//					.build());
+//			Map<String, Object> headers = (Map<String, Object>) sourceApp.arguments.pop();
+//			sourceApp.arguments.pop();
+//			assertThat(((MimeType) headers.get(MessageHeaders.CONTENT_TYPE))
+//					.includes(MimeTypeUtils.IMAGE_GIF));
+//			headers = (Map<String, Object>) sourceApp.arguments.pop();
+//			sourceApp.arguments.pop();
+//			assertThat(((MimeType) headers.get(MessageHeaders.CONTENT_TYPE))
+//					.includes(MimeTypeUtils.IMAGE_JPEG));
+//		}
+//	}
+//
+//
+//	public interface TestSink {
+//
+//		@Input("POJO_INPUT")
+//		SubscribableChannel pojo();
+//
+//		@Input("STRING_INPUT")
+//		SubscribableChannel string();
+//
+//		@Input("TUPLE_INPUT")
+//		SubscribableChannel tuple();
+//
+//		@Input("RAW_INPUT")
+//		SubscribableChannel raw();
+//
+//	}
+//
+//	@EnableBinding(Source.class)
+//	@SpringBootApplication
+//	public static class SourceApplication {
+//
+//	}
+//
+//	@EnableBinding(TestSink.class)
+//	@SpringBootApplication
+//	public static class SinkApplication {
+//
+//		public LinkedList<? super Object> arguments = new LinkedList<>();
+//
+//		@StreamListener("POJO_INPUT")
+//		public void receive(User user, @Headers Map<String, Object> headers) {
+//			this.arguments.push(user);
+//			this.arguments.push(headers);
+//		}
+//
+//		@StreamListener("STRING_INPUT")
+//		public void receive(String string) {
+//		}
+//
+//		@StreamListener("RAW_INPUT")
+//		public void receive(byte[] data, @Headers Map<String, Object> headers) {
+//			this.arguments.push(data);
+//			this.arguments.push(headers);
+//		}
+//
+//	}
+//
+//}

core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/contentType/User.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2017-2019 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
+ *
+ *      https://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.contentType;
+
+import java.io.Serializable;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * @author Vinicius Carvalho
+ * @author Oleg Zhurakousky
+ */
+@SuppressWarnings("serial")
+public class User implements Serializable {
+
+	private String name;
+
+	public User() {
+	}
+
+	@JsonCreator
+	public User(@JsonProperty("name") String name) {
+		this.name = name;
+	}
+
+	public String getName() {
+		return this.name;
+	}
+
+	public void setName(String name) {
+		this.name = name;
+	}
+
+	@Override
+	public String toString() {
+		final StringBuffer sb = new StringBuffer("User{");
+		sb.append("name='").append(this.name).append('\'');
+		sb.append('}');
+		return sb.toString();
+	}
+
+}

core/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/native-decoding-sink.properties

@@ -0,0 +1,3 @@
+spring.cloud.stream.bindings.input.destination=foobar
+spring.cloud.stream.bindings.input.consumer.useNativeDecoding=true
+spring.cloud.stream.bindings.input.contentType=application/x-java-serialized-object

core/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/native-encoding-source.properties

@@ -0,0 +1,3 @@
+spring.cloud.stream.bindings.output.destination=foobar-x
+spring.cloud.stream.bindings.output.producer.useNativeEncoding=true
+spring.cloud.stream.bindings.output.contentType=application/x-java-serialized-object

core/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/partitioned-configurers.properties

@@ -0,0 +1,4 @@
+spring.cloud.stream.bindings.output.destination=partOut
+spring.cloud.stream.bindings.output.producer.partitionKeyExpression=payload
+spring.cloud.stream.bindings.output.producer.partitionCount=3
+

core/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/sink-channel-configurers.properties

@@ -0,0 +1,3 @@
+spring.cloud.stream.bindings.input.destination=configure1
+spring.cloud.stream.bindings.input.contentType=application/x-spring-tuple
+spring.jackson.serialization.WRITE_DATES_AS_TIMESTAMPS=false

core/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/channel/source-channel-configurers.properties

@@ -0,0 +1,2 @@
+spring.cloud.stream.bindings.output.destination=interceptor-test
+spring.cloud.stream.bindings.output.contentType=application/json

core/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/custom/source-channel-configurers.properties

@@ -0,0 +1,2 @@
+spring.cloud.stream.bindings.output.destination=configure1
+spring.cloud.stream.bindings.output.contentType=test/foo

core/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/fooprocesor/foo-sink.properties

@@ -0,0 +1,2 @@
+spring.cloud.stream.bindings.input.destination=foo-input
+spring.cloud.stream.bindings.input.content-type=application/x-java-object;type=org.springframework.cloud.stream.config.DeserializeJSONToJavaTypeTests.Foo

core/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/inboundjsontuple/inbound-json-tuple.properties

@@ -0,0 +1,2 @@
+spring.cloud.stream.bindings.input.content-type=application/x-spring-tuple
+spring.cloud.stream.bindings.output.content-type=application/x-spring-tuple

core/spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/textplain/text-plain.properties

@@ -0,0 +1,2 @@
+spring.cloud.stream.bindings.input.content-type=text/plain
+spring.cloud.stream.bindings.output.content-type=text/plain