SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Initial removal of EnableBinding

Browse Source
This commit is contained in:
Oleg Zhurakousky
2022-01-12 17:45:04 +01:00
parent c755bfd9fd
commit 1d4c80bed1
53 changed files with 158 additions and 1402 deletions
Download Patch File Download Diff File Expand all files Collapse all files

73
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/ContentTypeOutboundSourceTests.java
View File

@@ -1,73 +0,0 @@
/*
* Copyright 2015-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;
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.binder.BinderFactory;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.test.binder.TestSupportBinder;
import org.springframework.context.annotation.PropertySource;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Ilayaperumal Gopinathan
* @author Vinicius Carvalho
* @author Oleg Zhurakousky
*/
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(classes = { ContentTypeOutboundSourceTests.TestSource.class })
public class ContentTypeOutboundSourceTests {
@Autowired
private Source testSource;
@Autowired
private BinderFactory binderFactory;
@Test
@SuppressWarnings("unchecked")
public void testMessageHeaderWhenNoExplicitContentTypeOnMessage() throws Exception {
this.testSource.output().send(MessageBuilder.withPayload("{\"message\":\"Hi\"}")
.setHeader(MessageHeaders.CONTENT_TYPE, "text/plain").build());
Message<String> received = (Message<String>) ((TestSupportBinder) this.binderFactory
.getBinder(null, MessageChannel.class)).messageCollector()
.forChannel(this.testSource.output()).poll();
assertThat(received.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString())
.contains("text/plain");
assertThat("{\"message\":\"Hi\"}").isEqualTo(received.getPayload());
}
@EnableBinding(Source.class)
@EnableAutoConfiguration
@PropertySource("classpath:/org/springframework/cloud/stream/config/channel/source-channel-configurers.properties")
public static class TestSource {
}
}

94
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/CustomHeaderPropagationTests.java
View File

@@ -1,94 +0,0 @@
/*
* 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;
import java.util.concurrent.TimeUnit;
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.binder.BinderFactory;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.TestSupportBinder;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Oleg Zhurakousky
*/
@RunWith(SpringJUnit4ClassRunner.class)
// @checkstyle:off
@SpringBootTest(classes = CustomHeaderPropagationTests.HeaderPropagationProcessor.class, webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = {
"spring.cloud.stream.integration.messageHandlerNotPropagatedHeaders=bar,contentType" })
public class CustomHeaderPropagationTests {
// @checkstyle:on
@Autowired
private Processor testProcessor;
@Autowired
private BinderFactory binderFactory;
@Test
/**
* @since 2.0 The behavior of content type handling has changed. All input/output
* channels have a default content type of application/json When a processor or a
* source returns a String, and if the content type is json it will be quoted
*/
public void testCustomHeaderPropagation() throws Exception {
this.testProcessor.input().send(MessageBuilder.withPayload("{'name':'foo'}")
.setHeader(MessageHeaders.CONTENT_TYPE, "application/json")
.setHeader("foo", "fooValue").setHeader("bar", "barValue").build());
@SuppressWarnings("unchecked")
Message<String> received = (Message<String>) ((TestSupportBinder) this.binderFactory
.getBinder(null, MessageChannel.class)).messageCollector()
.forChannel(this.testProcessor.output())
.poll(10, TimeUnit.SECONDS);
assertThat(received).isNotNull();
assertThat(received.getHeaders()).containsEntry("foo", "fooValue");
assertThat(received.getHeaders()).doesNotContainKey("bar");
assertThat(received.getHeaders()).containsKeys(MessageHeaders.CONTENT_TYPE);
assertThat(new String(received.getPayload())).isEqualTo("{'name':'foo'}");
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class HeaderPropagationProcessor {
@ServiceActivator(inputChannel = "input", outputChannel = "output")
public Message<String> consume(String data) {
// if we don't force content to be String, it will be quoted on the outbound
// channel
return MessageBuilder.withPayload(data)
.setHeader(MessageHeaders.CONTENT_TYPE, "text/plain").build();
}
}
}

175
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/CustomMessageConverterTests.java
View File

@@ -1,175 +0,0 @@
/*
* Copyright 2015-2017 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.List;
import java.util.concurrent.TimeUnit;
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.binder.BinderFactory;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.test.binder.TestSupportBinder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
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.converter.AbstractMessageConverter;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.MimeType;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Ilayaperumal Gopinathan
* @author Janne Valkealahti
*/
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(classes = CustomMessageConverterTests.TestSource.class)
public class CustomMessageConverterTests {
@Autowired
private Source testSource;
@Autowired
private BinderFactory binderFactory;
@Autowired
private List<MessageConverter> customMessageConverters;
@Test
public void testCustomMessageConverter() throws Exception {
assertThat(this.customMessageConverters).extracting("class")
.contains(FooConverter.class, BarConverter.class);
this.testSource.output().send(MessageBuilder.withPayload(new Foo("hi")).build());
@SuppressWarnings("unchecked")
Message<String> received = (Message<String>) ((TestSupportBinder) this.binderFactory
.getBinder(null, MessageChannel.class)).messageCollector()
.forChannel(this.testSource.output()).poll(1, TimeUnit.SECONDS);
assertThat(received).isNotNull();
assertThat(received.getHeaders().get(MessageHeaders.CONTENT_TYPE))
.isEqualTo(MimeType.valueOf("test/foo"));
}
@EnableBinding(Source.class)
@EnableAutoConfiguration
@PropertySource("classpath:/org/springframework/cloud/stream/config/custom/source-channel-configurers.properties")
@Configuration
public static class TestSource {
@Bean
public MessageConverter fooConverter() {
return new FooConverter();
}
@Bean
public MessageConverter barConverter() {
return new BarConverter();
}
}
public static class FooConverter extends AbstractMessageConverter {
public FooConverter() {
super(MimeType.valueOf("test/foo"));
}
@Override
protected boolean supports(Class<?> clazz) {
return clazz.equals(Foo.class);
}
@Override
protected Object convertToInternal(Object payload, MessageHeaders headers,
Object conversionHint) {
Object result = null;
try {
if (payload instanceof Foo) {
Foo fooPayload = (Foo) payload;
result = fooPayload.test.getBytes();
}
}
catch (Exception e) {
this.logger.error(e.getMessage(), e);
return null;
}
return result;
}
}
public static class BarConverter extends AbstractMessageConverter {
public BarConverter() {
super(MimeType.valueOf("test/bar"));
}
@Override
protected boolean supports(Class<?> clazz) {
return clazz.equals(Bar.class);
}
@Override
protected Object convertToInternal(Object payload, MessageHeaders headers,
Object conversionHint) {
Object result = null;
try {
if (payload instanceof Bar) {
Bar barPayload = (Bar) payload;
result = barPayload.testing.getBytes();
}
}
catch (Exception e) {
this.logger.error(e.getMessage(), e);
return null;
}
return result;
}
}
public static class Foo {
final String test;
public Foo(String test) {
this.test = test;
}
}
public static class Bar {
final String testing;
public Bar(String testing) {
this.testing = testing;
}
}
}

85
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DefaultHeaderPropagationTests.java
View File

@@ -1,85 +0,0 @@
/*
* 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;
import java.util.concurrent.TimeUnit;
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.binder.BinderFactory;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.TestSupportBinder;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Oleg Zhurakousky
*/
@RunWith(SpringJUnit4ClassRunner.class)
// @checkstyle:off
@SpringBootTest(classes = DefaultHeaderPropagationTests.HeaderPropagationProcessor.class, webEnvironment = SpringBootTest.WebEnvironment.NONE)
public class DefaultHeaderPropagationTests {
// @checkstyle:on
@Autowired
private Processor testProcessor;
@Autowired
private BinderFactory binderFactory;
@Test
public void testDefaultHeaderPropagation() throws Exception {
this.testProcessor.input().send(MessageBuilder.withPayload("{'name':'foo'}")
.setHeader(MessageHeaders.CONTENT_TYPE, "application/json")
.setHeader("foo", "fooValue").setHeader("bar", "barValue").build());
@SuppressWarnings("unchecked")
Message<String> received = (Message<String>) ((TestSupportBinder) this.binderFactory
.getBinder(null, MessageChannel.class)).messageCollector()
.forChannel(this.testProcessor.output())
.poll(1, TimeUnit.SECONDS);
assertThat(received).isNotNull();
assertThat(received.getHeaders()).containsEntry("foo", "fooValue");
assertThat(received.getHeaders()).containsEntry("bar", "barValue");
assertThat(received.getHeaders()).containsKeys(MessageHeaders.CONTENT_TYPE);
assertThat(received.getPayload()).isEqualTo("{'name':'foo'}");
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class HeaderPropagationProcessor {
@ServiceActivator(inputChannel = "input", outputChannel = "output")
public Message<String> consume(String data) {
return MessageBuilder.withPayload(data)
.setHeader(MessageHeaders.CONTENT_TYPE, "text/plain").build();
}
}
}

83
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DefaultHeaderPropagationWithApplicationProvidedHeaderTests.java
View File

@@ -1,83 +0,0 @@
/*
* 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 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.binder.BinderFactory;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.TestSupportBinder;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Oleg Zhurakousky
*/
@RunWith(SpringJUnit4ClassRunner.class)
// @checkstyle:off
@SpringBootTest(classes = DefaultHeaderPropagationWithApplicationProvidedHeaderTests.HeaderPropagationProcessor.class, webEnvironment = SpringBootTest.WebEnvironment.NONE)
public class DefaultHeaderPropagationWithApplicationProvidedHeaderTests {
// @checkstyle:on
@Autowired
private Processor testProcessor;
@Autowired
private BinderFactory binderFactory;
@Test
public void testHeaderPropagationIfSetByApplication() throws Exception {
this.testProcessor.input().send(MessageBuilder.withPayload("{'name':'foo'}")
.setHeader(MessageHeaders.CONTENT_TYPE, "application/json")
.setHeader("foo", "fooValue").setHeader("bar", "barValue").build());
@SuppressWarnings("unchecked")
Message<String> received = (Message<String>) ((TestSupportBinder) this.binderFactory
.getBinder(null, MessageChannel.class)).messageCollector()
.forChannel(this.testProcessor.output())
.poll(1, TimeUnit.SECONDS);
assertThat(received.getHeaders().get("foo")).isEqualTo("fooValue");
assertThat(received.getHeaders().get("bar")).isEqualTo("barValue");
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class HeaderPropagationProcessor {
@ServiceActivator(inputChannel = "input", outputChannel = "output")
public Message<?> consume(String data) {
return MessageBuilder.withPayload(data)
.setHeader(MessageHeaders.CONTENT_TYPE, "text/plain").build();
}
}
}

95
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DeserializeJSONToJavaTypeTests.java
View File

@@ -1,95 +0,0 @@
/*
* Copyright 2016-2017 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 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.binder.BinderFactory;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.TestSupportBinder;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Oleg Zhurakousky
*/
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(classes = DeserializeJSONToJavaTypeTests.FooProcessor.class)
public class DeserializeJSONToJavaTypeTests {
@Autowired
private Processor testProcessor;
@Autowired
private BinderFactory binderFactory;
@Test
public void testMessageDeserialized() throws Exception {
this.testProcessor.input().send(MessageBuilder.withPayload("{\"name\":\"Bar\"}")
.setHeader("contentType", "application/json").build());
@SuppressWarnings("unchecked")
Message<String> received = (Message<String>) ((TestSupportBinder) this.binderFactory
.getBinder(null, MessageChannel.class)).messageCollector()
.forChannel(this.testProcessor.output())
.poll(1, TimeUnit.SECONDS);
assertThat(received).isNotNull();
assertThat(received.getPayload()).isEqualTo("{\"name\":\"Bar\"}");
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
@PropertySource("classpath:/org/springframework/cloud/stream/config/fooprocesor/foo-sink.properties")
@Configuration
public static class FooProcessor {
@ServiceActivator(inputChannel = "input", outputChannel = "output")
public Foo consume(Foo foo) {
return foo;
}
}
public static class Foo {
private String name;
public String getName() {
return this.name;
}
public void setName(String name) {
this.name = name;
}
}
}

121
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java
View File

@@ -1,121 +0,0 @@
/*
* Copyright 2015-2017 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.CountDownLatch;
import java.util.concurrent.TimeUnit;
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.binder.BinderHeaders;
import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.annotation.PropertySource;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Ilayaperumal Gopinathan
* @author Gary Russell
* @author Soby Chacko
*/
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(classes = { MessageChannelConfigurerTests.TestSink.class,
MessageChannelConfigurerTests.TestSource.class,
SpelExpressionConverterConfiguration.class })
public class MessageChannelConfigurerTests {
@Autowired
private Sink testSink;
@Autowired
private Source testSource;
@Autowired
private MessageCollector messageCollector;
@Test
public void testChannelTypes() throws Exception {
DirectWithAttributesChannel inputChannel = (DirectWithAttributesChannel) this.testSink
.input();
DirectWithAttributesChannel outputChannel = (DirectWithAttributesChannel) this.testSource
.output();
assertThat(inputChannel.getAttribute("type")).isEqualTo(Sink.INPUT);
assertThat(outputChannel.getAttribute("type")).isEqualTo(Source.OUTPUT);
}
@Test
public void testMessageConverterConfigurer() throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
MessageHandler messageHandler = message -> {
assertThat(message.getPayload()).isInstanceOf(byte[].class);
assertThat(message.getPayload()).isEqualTo("{\"message\":\"Hi\"}".getBytes());
latch.countDown();
};
this.testSink.input().subscribe(messageHandler);
this.testSink.input().send(
MessageBuilder.withPayload("{\"message\":\"Hi\"}".getBytes()).build());
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
this.testSink.input().unsubscribe(messageHandler);
}
@Test
public void testPartitionHeader() throws Exception {
this.testSource.output()
.send(MessageBuilder.withPayload("{\"message\":\"Hi\"}").build());
Message<?> message = this.messageCollector.forChannel(this.testSource.output())
.poll(1, TimeUnit.SECONDS);
assertThat(message.getHeaders().get(BinderHeaders.PARTITION_HEADER).equals(0));
assertThat(message.getHeaders().get(BinderHeaders.PARTITION_OVERRIDE)).isNull();
}
@Test
public void testPartitionHeaderWithPartitionOverride() throws Exception {
this.testSource.output().send(MessageBuilder.withPayload("{\"message\":\"Hi\"}")
.setHeader(BinderHeaders.PARTITION_OVERRIDE, 123).build());
Message<?> message = this.messageCollector.forChannel(this.testSource.output())
.poll(1, TimeUnit.SECONDS);
assertThat(message.getHeaders().get(BinderHeaders.PARTITION_HEADER).equals(123));
assertThat(message.getHeaders().get(BinderHeaders.PARTITION_OVERRIDE)).isNull();
}
@EnableBinding(Sink.class)
@EnableAutoConfiguration
@PropertySource("classpath:/org/springframework/cloud/stream/config/channel/sink-channel-configurers.properties")
public static class TestSink {
}
@EnableBinding(Source.class)
@EnableAutoConfiguration
@PropertySource("classpath:/org/springframework/cloud/stream/config/channel/partitioned-configurers.properties")
public static class TestSource {
}
}

87
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelWithNativeDecodingTests.java
View File

@@ -1,87 +0,0 @@
/*
* Copyright 2015-2017 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.io.ByteArrayOutputStream;
import java.io.ObjectOutput;
import java.io.ObjectOutputStream;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
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.messaging.Sink;
import org.springframework.context.annotation.PropertySource;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.MessageHandler;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Soby Chacko
*/
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(classes = {
MessageChannelWithNativeDecodingTests.NativeDecodingSink.class })
public class MessageChannelWithNativeDecodingTests {
@Autowired
private Sink nativeDecodingSink;
@Test
public void testMessageConverterInterceptorsAreSkippedWhenNativeDecodingIsEnabled()
throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
byte[] serializedData;
ObjectOutput out;
try (ByteArrayOutputStream bos = new ByteArrayOutputStream()) {
out = new ObjectOutputStream(bos);
out.writeObject(123);
out.flush();
serializedData = bos.toByteArray();
}
MessageHandler messageHandler = message -> {
// ensure that the data is not deserialized becasue of native decoding
// and the content type set in the properties file didn't take any effect
assertThat(message.getPayload()).isInstanceOf(byte[].class);
assertThat(message.getPayload()).isEqualTo(serializedData);
latch.countDown();
};
this.nativeDecodingSink.input().subscribe(messageHandler);
this.nativeDecodingSink.input()
.send(MessageBuilder.withPayload(serializedData).build());
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
this.nativeDecodingSink.input().unsubscribe(messageHandler);
}
@EnableBinding(Sink.class)
@EnableAutoConfiguration
@PropertySource("classpath:/org/springframework/cloud/stream/config/channel/native-decoding-sink.properties")
public static class NativeDecodingSink {
}
}

74
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelWithNativeEncodingTests.java
View File

@@ -1,74 +0,0 @@
/*
* Copyright 2015-2017 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 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.messaging.Source;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.context.annotation.PropertySource;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Soby Chacko
*/
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(classes = {
MessageChannelWithNativeEncodingTests.NativeEncodingSource.class })
public class MessageChannelWithNativeEncodingTests {
@Autowired
private Source nativeEncodingSource;
@Autowired
private MessageCollector messageCollector;
@Test
public void testOutboundContentTypeInterceptorIsSkippedWhenNativeEncodingIsEnabled()
throws Exception {
this.nativeEncodingSource.output()
.send(MessageBuilder.withPayload("hello foobar!").build());
Message<?> message = this.messageCollector
.forChannel(this.nativeEncodingSource.output()).poll(1, TimeUnit.SECONDS);
// should not convert the payload to byte[] even though we set a contentType on
// the channel.
// This is becasue, we are using native encoding.
assertThat(message.getPayload()).isInstanceOf(String.class);
assertThat(message.getPayload()).isEqualTo("hello foobar!");
assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE)).isNull();
}
@EnableBinding(Source.class)
@EnableAutoConfiguration
@PropertySource("classpath:/org/springframework/cloud/stream/config/channel/native-encoding-source.properties")
public static class NativeEncodingSource {
}
}

130
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/TextPlainConversionTest.java
View File

@@ -1,130 +0,0 @@
/*
* 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;
import java.util.concurrent.TimeUnit;
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.binder.BinderFactory;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.TestSupportBinder;
import org.springframework.context.annotation.PropertySource;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Oleg Zhurakousky
* @since 1.2
*/
@RunWith(SpringJUnit4ClassRunner.class)
// @checkstyle:off
@SpringBootTest(classes = TextPlainConversionTest.FooProcessor.class, webEnvironment = SpringBootTest.WebEnvironment.NONE)
// @checkstyle:on
public class TextPlainConversionTest {
@Autowired
private Processor testProcessor;
@Autowired
private BinderFactory binderFactory;
@Test
public void testTextPlainConversionOnOutput() throws Exception {
this.testProcessor.input().send(MessageBuilder.withPayload("Bar").build());
@SuppressWarnings("unchecked")
Message<String> received = (Message<String>) ((TestSupportBinder) this.binderFactory
.getBinder(null, MessageChannel.class)).messageCollector()
.forChannel(this.testProcessor.output())
.poll(1, TimeUnit.SECONDS);
assertThat(received).isNotNull();
assertThat(received.getPayload()).isEqualTo("Foo{name='Bar'}");
}
@Test
public void testByteArrayConversionOnOutput() throws Exception {
this.testProcessor.output()
.send(MessageBuilder.withPayload("Bar".getBytes()).build());
@SuppressWarnings("unchecked")
Message<String> received = (Message<String>) ((TestSupportBinder) this.binderFactory
.getBinder(null, MessageChannel.class)).messageCollector()
.forChannel(this.testProcessor.output())
.poll(1, TimeUnit.SECONDS);
assertThat(received).isNotNull();
assertThat(received.getPayload()).isEqualTo("Bar");
}
@Test
public void testTextPlainConversionOnInputAndOutput() throws Exception {
this.testProcessor.input()
.send(MessageBuilder.withPayload(new Foo("Bar")).build());
@SuppressWarnings("unchecked")
Message<String> received = (Message<String>) ((TestSupportBinder) this.binderFactory
.getBinder(null, MessageChannel.class)).messageCollector()
.forChannel(this.testProcessor.output())
.poll(1, TimeUnit.SECONDS);
assertThat(received).isNotNull();
assertThat(received.getPayload()).isEqualTo("Foo{name='Foo{name='Bar'}'}");
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
@PropertySource("classpath:/org/springframework/cloud/stream/config/textplain/text-plain.properties")
public static class FooProcessor {
@ServiceActivator(inputChannel = "input", outputChannel = "output")
public Foo consume(String foo) {
return new Foo(foo);
}
}
public static class Foo {
private String name;
public Foo(String name) {
this.name = name;
}
public String getName() {
return this.name;
}
public void setName(String name) {
this.name = name;
}
@Override
public String toString() {
return "Foo{name='" + this.name + "'}";
}
}
}