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

Move tests to AssertJ

Browse Source
This commit is contained in:
Marius Bogoevici
2016-05-29 08:32:29 -04:00
committed by Soby Chacko
parent fbe9bbac7e
commit aef08a28a3
29 changed files with 780 additions and 946 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-cloud-stream-binders/pom.xml
View File

@@ -1,5 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>spring-cloud-stream-binders-parent</artifactId>
<packaging>pom</packaging>
@@ -26,10 +27,5 @@
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-logging</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

265
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
View File

@@ -26,7 +26,6 @@ import java.util.concurrent.TimeUnit;
import kafka.admin.AdminUtils;
import kafka.api.TopicMetadata;
import org.hamcrest.collection.IsCollectionWithSize;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Test;
@@ -59,22 +58,9 @@ import org.springframework.retry.backoff.FixedBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import static org.hamcrest.Matchers.arrayWithSize;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasItemInArray;
import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.not;
import static org.hamcrest.Matchers.nullValue;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.fail;
/**
* Integration tests for the {@link KafkaMessageChannelBinder}.
* @author Eric Bottard
@@ -151,7 +137,7 @@ public class KafkaBinderTests extends
KafkaTestBinder binder = getBinder();
DirectChannel moduleOutputChannel = new DirectChannel();
DirectChannel moduleInputChannel = new DirectChannel();
QueueChannel dlqChannel = new QueueChannel();
QueueChannel dlqChannel = new QueueChannel();
FailingInvocationCountingMessageHandler handler = new FailingInvocationCountingMessageHandler();
moduleInputChannel.subscribe(handler);
ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
@@ -164,8 +150,8 @@ public class KafkaBinderTests extends
long uniqueBindingId = System.currentTimeMillis();
Binding<MessageChannel> producerBinding = binder.bindProducer("retryTest." + uniqueBindingId + ".0",
moduleOutputChannel, producerProperties);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("retryTest." + uniqueBindingId + ".0", "testGroup",
moduleInputChannel, consumerProperties);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("retryTest." + uniqueBindingId + ".0",
"testGroup", moduleInputChannel, consumerProperties);
ExtendedConsumerProperties<KafkaConsumerProperties> dlqConsumerProperties = createConsumerProperties();
dlqConsumerProperties.setMaxAttempts(1);
@@ -178,9 +164,9 @@ public class KafkaBinderTests extends
moduleOutputChannel.send(testMessage);
Message<?> receivedMessage = receive(dlqChannel, 3);
assertNotNull(receivedMessage);
assertEquals(testMessagePayload, receivedMessage.getPayload());
assertThat(handler.getInvocationCount(), equalTo(consumerProperties.getMaxAttempts()));
assertThat(receivedMessage).isNotNull();
assertThat(receivedMessage.getPayload()).isEqualTo(testMessagePayload);
assertThat(handler.getInvocationCount()).isEqualTo(consumerProperties.getMaxAttempts());
dlqConsumerBinding.unbind();
consumerBinding.unbind();
producerBinding.unbind();
@@ -209,13 +195,13 @@ public class KafkaBinderTests extends
Message<String> testMessage = MessageBuilder.withPayload(testMessagePayload).build();
moduleOutputChannel.send(testMessage);
assertTrue(handler.getLatch().await((int) (timeoutMultiplier * 1000), TimeUnit.MILLISECONDS));
assertThat(handler.getLatch().await((int) (timeoutMultiplier * 1000), TimeUnit.MILLISECONDS));
// first attempt fails
assertThat(handler.getReceivedMessages().entrySet(), hasSize(1));
assertThat(handler.getReceivedMessages().entrySet()).hasSize(1);
Message<?> receivedMessage = handler.getReceivedMessages().entrySet().iterator().next().getValue();
assertNotNull(receivedMessage);
assertEquals(testMessagePayload, receivedMessage.getPayload());
assertThat(handler.getInvocationCount(), equalTo(consumerProperties.getMaxAttempts()));
assertThat(receivedMessage).isNotNull();
assertThat(receivedMessage.getPayload()).isEqualTo(testMessagePayload);
assertThat(handler.getInvocationCount()).isEqualTo(consumerProperties.getMaxAttempts());
consumerBinding.unbind();
// on the second attempt the message is redelivered
@@ -227,9 +213,9 @@ public class KafkaBinderTests extends
moduleOutputChannel.send(testMessage2);
Message<?> firstReceived = receive(successfulInputChannel);
assertEquals(testMessagePayload, firstReceived.getPayload());
assertThat(firstReceived.getPayload()).isEqualTo(testMessagePayload);
Message<?> secondReceived = receive(successfulInputChannel);
assertEquals(testMessage2Payload, secondReceived.getPayload());
assertThat(secondReceived.getPayload()).isEqualTo(testMessage2Payload);
consumerBinding.unbind();
producerBinding.unbind();
}
@@ -264,20 +250,19 @@ public class KafkaBinderTests extends
moduleOutputChannel.send(testMessage);
Message<?> dlqMessage = receive(dlqChannel, 3);
assertNotNull(dlqMessage);
assertEquals(testMessagePayload, dlqMessage.getPayload());
assertThat(dlqMessage).isNotNull();
assertThat(dlqMessage.getPayload()).isEqualTo(testMessagePayload);
// first attempt fails
assertThat(handler.getReceivedMessages().entrySet(), hasSize(1));
assertThat(handler.getReceivedMessages().entrySet()).hasSize(1);
Message<?> handledMessage = handler.getReceivedMessages().entrySet().iterator().next().getValue();
assertNotNull(handledMessage);
assertEquals(testMessagePayload, handledMessage.getPayload());
assertThat(handler.getInvocationCount(), equalTo(consumerProperties.getMaxAttempts()));
assertThat(handledMessage).isNotNull();
assertThat(handledMessage.getPayload()).isEqualTo(testMessagePayload);
assertThat(handler.getInvocationCount()).isEqualTo(consumerProperties.getMaxAttempts());
dlqConsumerBinding.unbind();
consumerBinding.unbind();
// on the second attempt the message is not redelivered because the DLQ is set
QueueChannel successfulInputChannel = new QueueChannel();
consumerBinding = binder.bindConsumer("retryTest." + uniqueBindingId + ".0", "testGroup",
@@ -287,7 +272,7 @@ public class KafkaBinderTests extends
moduleOutputChannel.send(testMessage2);
Message<?> receivedMessage = receive(successfulInputChannel);
assertEquals(testMessage2Payload, receivedMessage.getPayload());
assertThat(receivedMessage.getPayload()).isEqualTo(testMessage2Payload);
consumerBinding.unbind();
producerBinding.unbind();
@@ -301,28 +286,28 @@ public class KafkaBinderTests extends
@Test
public void testCompression() throws Exception {
final ProducerMetadata.CompressionType[] codecs = new ProducerMetadata.CompressionType[] {
ProducerMetadata.CompressionType.none,
ProducerMetadata.CompressionType.gzip,
ProducerMetadata.CompressionType.none, ProducerMetadata.CompressionType.gzip,
ProducerMetadata.CompressionType.snappy };
byte[] ratherBigPayload = new byte[2048];
Arrays.fill(ratherBigPayload, (byte) 65);
byte[] testPayload = new byte[2048];
Arrays.fill(testPayload, (byte) 65);
KafkaTestBinder binder = getBinder();
for (ProducerMetadata.CompressionType codec : codecs) {
DirectChannel moduleOutputChannel = new DirectChannel();
QueueChannel moduleInputChannel = new QueueChannel();
ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
producerProperties.getExtension().setCompressionType(codec);
Binding<MessageChannel> producerBinding = binder.bindProducer("foo.0", moduleOutputChannel, producerProperties);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.0", "test", moduleInputChannel, createConsumerProperties());
Message<?> message = org.springframework.integration.support.MessageBuilder.withPayload(ratherBigPayload).build();
Binding<MessageChannel> producerBinding = binder.bindProducer("foo.0", moduleOutputChannel,
producerProperties);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.0", "test", moduleInputChannel,
createConsumerProperties());
Message<?> message = org.springframework.integration.support.MessageBuilder.withPayload(testPayload)
.build();
// Let the consumer actually bind to the producer before sending a msg
binderBindUnbindLatency();
moduleOutputChannel.send(message);
Message<?> inbound = receive(moduleInputChannel);
assertNotNull(inbound);
assertArrayEquals(ratherBigPayload, (byte[]) inbound.getPayload());
assertThat(inbound).isNotNull();
assertThat((byte[]) inbound.getPayload()).containsExactly(testPayload);
producerBinding.unbind();
consumerBinding.unbind();
}
@@ -331,8 +316,8 @@ public class KafkaBinderTests extends
@Test
public void testCustomPartitionCountOverridesDefaultIfLarger() throws Exception {
byte[] ratherBigPayload = new byte[2048];
Arrays.fill(ratherBigPayload, (byte) 65);
byte[] testPayload = new byte[2048];
Arrays.fill(testPayload, (byte) 65);
KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties();
binderConfiguration.setMinPartitionCount(10);
KafkaTestBinder binder = new KafkaTestBinder(binderConfiguration);
@@ -343,18 +328,21 @@ public class KafkaBinderTests extends
producerProperties.setPartitionCount(10);
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
long uniqueBindingId = System.currentTimeMillis();
Binding<MessageChannel> producerBinding = binder.bindProducer("foo" + uniqueBindingId + ".0", moduleOutputChannel, producerProperties);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo" + uniqueBindingId + ".0", null, moduleInputChannel, consumerProperties);
Message<?> message = org.springframework.integration.support.MessageBuilder.withPayload(ratherBigPayload).build();
Binding<MessageChannel> producerBinding = binder.bindProducer("foo" + uniqueBindingId + ".0",
moduleOutputChannel, producerProperties);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo" + uniqueBindingId + ".0", null,
moduleInputChannel, consumerProperties);
Message<?> message = org.springframework.integration.support.MessageBuilder.withPayload(testPayload)
.build();
// Let the consumer actually bind to the producer before sending a msg
binderBindUnbindLatency();
moduleOutputChannel.send(message);
Message<?> inbound = receive(moduleInputChannel);
assertNotNull(inbound);
assertArrayEquals(ratherBigPayload, (byte[]) inbound.getPayload());
Collection<Partition> partitions = binder.getCoreBinder().getConnectionFactory().getPartitions(
"foo" + uniqueBindingId + ".0");
assertThat(partitions, hasSize(10));
assertThat(inbound).isNotNull();
assertThat((byte[]) inbound.getPayload()).containsExactly(testPayload);
Collection<Partition> partitions = binder.getCoreBinder().getConnectionFactory()
.getPartitions("foo" + uniqueBindingId + ".0");
assertThat(partitions).hasSize(10);
producerBinding.unbind();
consumerBinding.unbind();
}
@@ -362,8 +350,8 @@ public class KafkaBinderTests extends
@Test
public void testCustomPartitionCountDoesNotOverridePartitioningIfSmaller() throws Exception {
byte[] ratherBigPayload = new byte[2048];
Arrays.fill(ratherBigPayload, (byte) 65);
byte[] testPayload = new byte[2048];
Arrays.fill(testPayload, (byte) 65);
KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties();
binderConfiguration.setMinPartitionCount(6);
KafkaTestBinder binder = new KafkaTestBinder(binderConfiguration);
@@ -374,18 +362,21 @@ public class KafkaBinderTests extends
producerProperties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload"));
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
long uniqueBindingId = System.currentTimeMillis();
Binding<MessageChannel> producerBinding = binder.bindProducer("foo" + uniqueBindingId + ".0", moduleOutputChannel, producerProperties);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo" + uniqueBindingId + ".0", null, moduleInputChannel, consumerProperties);
Message<?> message = org.springframework.integration.support.MessageBuilder.withPayload(ratherBigPayload).build();
Binding<MessageChannel> producerBinding = binder.bindProducer("foo" + uniqueBindingId + ".0",
moduleOutputChannel, producerProperties);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo" + uniqueBindingId + ".0", null,
moduleInputChannel, consumerProperties);
Message<?> message = org.springframework.integration.support.MessageBuilder.withPayload(testPayload)
.build();
// Let the consumer actually bind to the producer before sending a msg
binderBindUnbindLatency();
moduleOutputChannel.send(message);
Message<?> inbound = receive(moduleInputChannel);
assertNotNull(inbound);
assertArrayEquals(ratherBigPayload, (byte[]) inbound.getPayload());
Collection<Partition> partitions = binder.getCoreBinder().getConnectionFactory().getPartitions(
"foo" + uniqueBindingId + ".0");
assertThat(partitions, hasSize(6));
assertThat(inbound).isNotNull();
assertThat((byte[]) inbound.getPayload()).containsExactly(testPayload);
Collection<Partition> partitions = binder.getCoreBinder().getConnectionFactory()
.getPartitions("foo" + uniqueBindingId + ".0");
assertThat(partitions).hasSize(6);
producerBinding.unbind();
consumerBinding.unbind();
}
@@ -393,8 +384,8 @@ public class KafkaBinderTests extends
@Test
public void testCustomPartitionCountOverridesPartitioningIfLarger() throws Exception {
byte[] ratherBigPayload = new byte[2048];
Arrays.fill(ratherBigPayload, (byte) 65);
byte[] testPayload = new byte[2048];
Arrays.fill(testPayload, (byte) 65);
KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties();
binderConfiguration.setMinPartitionCount(4);
KafkaTestBinder binder = new KafkaTestBinder(binderConfiguration);
@@ -406,18 +397,21 @@ public class KafkaBinderTests extends
producerProperties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload"));
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
long uniqueBindingId = System.currentTimeMillis();
Binding<MessageChannel> producerBinding = binder.bindProducer("foo" + uniqueBindingId + ".0", moduleOutputChannel, producerProperties);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo" + uniqueBindingId + ".0", null, moduleInputChannel, consumerProperties);
Message<?> message = org.springframework.integration.support.MessageBuilder.withPayload(ratherBigPayload).build();
Binding<MessageChannel> producerBinding = binder.bindProducer("foo" + uniqueBindingId + ".0",
moduleOutputChannel, producerProperties);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo" + uniqueBindingId + ".0", null,
moduleInputChannel, consumerProperties);
Message<?> message = org.springframework.integration.support.MessageBuilder.withPayload(testPayload)
.build();
// Let the consumer actually bind to the producer before sending a msg
binderBindUnbindLatency();
moduleOutputChannel.send(message);
Message<?> inbound = receive(moduleInputChannel);
assertNotNull(inbound);
assertArrayEquals(ratherBigPayload, (byte[]) inbound.getPayload());
Collection<Partition> partitions = binder.getCoreBinder().getConnectionFactory().getPartitions(
"foo" + uniqueBindingId + ".0");
assertThat(partitions, hasSize(5));
assertThat(inbound).isNotNull();
assertThat((byte[]) inbound.getPayload()).containsExactly(testPayload);
Collection<Partition> partitions = binder.getCoreBinder().getConnectionFactory()
.getPartitions("foo" + uniqueBindingId + ".0");
assertThat(partitions).hasSize(5);
producerBinding.unbind();
consumerBinding.unbind();
}
@@ -439,13 +433,13 @@ public class KafkaBinderTests extends
output.send(new GenericMessage<>(testPayload1.getBytes()));
binder.bindConsumer(testTopicName, "startOffsets", input1, createConsumerProperties());
Message<byte[]> receivedMessage1 = (Message<byte[]>) receive(input1);
assertThat(receivedMessage1, not(nullValue()));
assertThat(new String(receivedMessage1.getPayload()), equalTo(testPayload1));
assertThat(receivedMessage1).isNotNull();
assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload1);
String testPayload2 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload2.getBytes()));
Message<byte[]> receivedMessage2 = (Message<byte[]>) receive(input1);
assertThat(receivedMessage2, not(nullValue()));
assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload2));
assertThat(receivedMessage2).isNotNull();
assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload2);
}
@Test
@@ -463,12 +457,12 @@ public class KafkaBinderTests extends
properties.getExtension().setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest);
binder.bindConsumer(testTopicName, "startOffsets", input1, properties);
Message<byte[]> receivedMessage1 = (Message<byte[]>) receive(input1);
assertThat(receivedMessage1, not(nullValue()));
assertThat(receivedMessage1).isNotNull();
String testPayload2 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload2.getBytes()));
Message<byte[]> receivedMessage2 = (Message<byte[]>) receive(input1);
assertThat(receivedMessage2, not(nullValue()));
assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload2));
assertThat(receivedMessage2).isNotNull();
assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload2);
}
@Test
@@ -480,21 +474,22 @@ public class KafkaBinderTests extends
String testTopicName = UUID.randomUUID().toString();
Binding<MessageChannel> producerBinding = binder.bindProducer(testTopicName, output, createProducerProperties());
Binding<MessageChannel> producerBinding = binder.bindProducer(testTopicName, output,
createProducerProperties());
String testPayload1 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload1.getBytes()));
ExtendedConsumerProperties<KafkaConsumerProperties> properties = createConsumerProperties();
properties.getExtension().setResetOffsets(true);
properties.getExtension().setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest);
Binding<MessageChannel> consumerBinding =
binder.bindConsumer(testTopicName, "startOffsets", input1, properties);
Binding<MessageChannel> consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1,
properties);
Message<byte[]> receivedMessage1 = (Message<byte[]>) receive(input1);
assertThat(receivedMessage1, not(nullValue()));
assertThat(receivedMessage1).isNotNull();
String testPayload2 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload2.getBytes()));
Message<byte[]> receivedMessage2 = (Message<byte[]>) receive(input1);
assertThat(receivedMessage2, not(nullValue()));
assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload2));
assertThat(receivedMessage2).isNotNull();
assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload2);
consumerBinding.unbind();
String testPayload3 = "foo-" + UUID.randomUUID().toString();
@@ -503,17 +498,16 @@ public class KafkaBinderTests extends
ExtendedConsumerProperties<KafkaConsumerProperties> properties2 = createConsumerProperties();
properties2.getExtension().setResetOffsets(true);
properties2.getExtension().setStartOffset(KafkaMessageChannelBinder.StartOffset.earliest);
consumerBinding =
binder.bindConsumer(testTopicName, "startOffsets", input1, properties2);
consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, properties2);
Message<byte[]> receivedMessage4 = (Message<byte[]>) receive(input1);
assertThat(receivedMessage4, not(nullValue()));
assertThat(new String(receivedMessage4.getPayload()), equalTo(testPayload1));
assertThat(receivedMessage4).isNotNull();
assertThat(new String(receivedMessage4.getPayload())).isEqualTo(testPayload1);
Message<byte[]> receivedMessage5 = (Message<byte[]>) receive(input1);
assertThat(receivedMessage5, not(nullValue()));
assertThat(new String(receivedMessage5.getPayload()), equalTo(testPayload2));
assertThat(receivedMessage5).isNotNull();
assertThat(new String(receivedMessage5.getPayload())).isEqualTo(testPayload2);
Message<byte[]> receivedMessage6 = (Message<byte[]>) receive(input1);
assertThat(receivedMessage6, not(nullValue()));
assertThat(new String(receivedMessage6.getPayload()), equalTo(testPayload3));
assertThat(receivedMessage6).isNotNull();
assertThat(new String(receivedMessage6.getPayload())).isEqualTo(testPayload3);
consumerBinding.unbind();
producerBinding.unbind();
}
@@ -531,28 +525,29 @@ public class KafkaBinderTests extends
QueueChannel input1 = new QueueChannel();
String testTopicName = UUID.randomUUID().toString();
Binding<MessageChannel> producerBinding = binder.bindProducer(testTopicName, output, createProducerProperties());
Binding<MessageChannel> producerBinding = binder.bindProducer(testTopicName, output,
createProducerProperties());
String testPayload1 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload1.getBytes()));
ExtendedConsumerProperties<KafkaConsumerProperties> firstConsumerProperties = createConsumerProperties();
Binding<MessageChannel> consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, firstConsumerProperties);
Binding<MessageChannel> consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1,
firstConsumerProperties);
Message<byte[]> receivedMessage1 = (Message<byte[]>) receive(input1);
assertThat(receivedMessage1, not(nullValue()));
assertThat(receivedMessage1).isNotNull();
String testPayload2 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload2.getBytes()));
Message<byte[]> receivedMessage2 = (Message<byte[]>) receive(input1);
assertThat(receivedMessage2, not(nullValue()));
assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload2));
assertThat(receivedMessage2).isNotNull();
assertThat(new String(receivedMessage2.getPayload())).isNotNull();
consumerBinding.unbind();
String testPayload3 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload3.getBytes()));
consumerBinding =
binder.bindConsumer(testTopicName, "startOffsets", input1, createConsumerProperties());
consumerBinding = binder.bindConsumer(testTopicName, "startOffsets", input1, createConsumerProperties());
Message<byte[]> receivedMessage3 = (Message<byte[]>) receive(input1);
assertThat(receivedMessage3, not(nullValue()));
assertThat(new String(receivedMessage3.getPayload()), equalTo(testPayload3));
assertThat(receivedMessage3).isNotNull();
assertThat(new String(receivedMessage3.getPayload())).isEqualTo(testPayload3);
consumerBinding.unbind();
producerBinding.unbind();
}
@@ -572,8 +567,10 @@ public class KafkaBinderTests extends
DirectFieldAccessor accessor = new DirectFieldAccessor(extractEndpoint(producerBinding));
MessageHandler handler = (MessageHandler) accessor.getPropertyValue("handler");
DirectFieldAccessor accessor1 = new DirectFieldAccessor(handler);
ProducerConfiguration producerConfiguration = (ProducerConfiguration) accessor1.getPropertyValue("producerConfiguration");
assertTrue("Kafka Sync Producer should have been enabled.", producerConfiguration.getProducerMetadata().isSync());
ProducerConfiguration producerConfiguration = (ProducerConfiguration) accessor1
.getPropertyValue("producerConfiguration");
assertThat(producerConfiguration.getProducerMetadata().isSync())
.withFailMessage("Kafka Sync Producer should have been enabled.");
producerBinding.unbind();
}
@@ -594,14 +591,14 @@ public class KafkaBinderTests extends
binder.setMetadataRetryOperations(metatadataRetrievalRetryOperations);
DirectChannel output = new DirectChannel();
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
String testTopicName = "nonexisting" + System.currentTimeMillis();
String testTopicName = "nonexisting" + System.currentTimeMillis();
try {
binder.doBindConsumer(testTopicName, "test", output, consumerProperties);
fail();
}
catch (Exception e) {
assertTrue(e instanceof BinderException);
assertThat(e.getMessage(), containsString("Topic " + testTopicName + " does not exist"));
assertThat(e).isInstanceOf(BinderException.class);
assertThat(e).hasMessageContaining("Topic " + testTopicName + " does not exist");
}
try {
@@ -609,13 +606,13 @@ public class KafkaBinderTests extends
fail();
}
catch (Exception e) {
assertThat(e, instanceOf(TopicNotFoundException.class));
assertThat(e).isInstanceOf(TopicNotFoundException.class);
}
}
@Test
public void testAutoConfigureTopicsDisabledSucceedsIfTopicExisting() throws Exception {
String testTopicName = "existing" + System.currentTimeMillis();
String testTopicName = "existing" + System.currentTimeMillis();
AdminUtils.createTopic(kafkaTestSupport.getZkClient(), testTopicName, 5, 1, new Properties());
KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
configurationProperties.setAutoCreateTopics(false);
@@ -632,7 +629,7 @@ public class KafkaBinderTests extends
@Test
public void testAutoAddPartitionsDisabledFailsIfTopicUnderpartitioned() throws Exception {
String testTopicName = "existing" + System.currentTimeMillis();
String testTopicName = "existing" + System.currentTimeMillis();
AdminUtils.createTopic(kafkaTestSupport.getZkClient(), testTopicName, 1, 1, new Properties());
KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
configurationProperties.setAutoAddPartitions(false);
@@ -650,16 +647,16 @@ public class KafkaBinderTests extends
binder.doBindConsumer(testTopicName, "test", output, consumerProperties);
}
catch (Exception e) {
assertThat(e, instanceOf(BinderException.class));
assertThat(e.getMessage(),
containsString("The number of expected partitions was: 3, but 1 has been found instead"));
assertThat(e).isInstanceOf(BinderException.class);
assertThat(e)
.hasMessageContaining("The number of expected partitions was: 3, but 1 has been found instead");
}
}
@Test
public void testAutoAddPartitionsDisabledSucceedsIfTopicPartitionedCorrectly() throws Exception {
String testTopicName = "existing" + System.currentTimeMillis();
String testTopicName = "existing" + System.currentTimeMillis();
AdminUtils.createTopic(kafkaTestSupport.getZkClient(), testTopicName, 6, 1, new Properties());
KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
configurationProperties.setAutoAddPartitions(false);
@@ -682,14 +679,13 @@ public class KafkaBinderTests extends
Binding<?> binding = binder.doBindConsumer(testTopicName, "test", output, consumerProperties);
Partition[] listenedPartitions
= TestUtils.getPropertyValue(binding, "endpoint.val$messageListenerContainer.partitions", Partition[].class);
Partition[] listenedPartitions = TestUtils.getPropertyValue(binding,
"endpoint.val$messageListenerContainer.partitions", Partition[].class);
assertThat(listenedPartitions, arrayWithSize(2));
assertThat(listenedPartitions, hasItemInArray(new Partition(testTopicName, 2)));
assertThat(listenedPartitions, hasItemInArray(new Partition(testTopicName, 5)));
assertThat(listenedPartitions).hasSize(2);
assertThat(listenedPartitions).contains(new Partition(testTopicName, 2), new Partition(testTopicName, 5));
Collection<Partition> partitions = binder.getConnectionFactory().getPartitions(testTopicName);
assertThat(partitions, IsCollectionWithSize.hasSize(6));
assertThat(partitions).hasSize(6);
binding.unbind();
}
@@ -710,14 +706,14 @@ public class KafkaBinderTests extends
binder.setMetadataRetryOperations(metatadataRetrievalRetryOperations);
DirectChannel output = new DirectChannel();
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
String testTopicName = "nonexisting" + System.currentTimeMillis();
String testTopicName = "nonexisting" + System.currentTimeMillis();
Binding<?> binding = binder.doBindConsumer(testTopicName, "test", output, consumerProperties);
binding.unbind();
}
@Test
public void testPartitionCountNotReduced() throws Exception {
String testTopicName = "existing" + System.currentTimeMillis();
String testTopicName = "existing" + System.currentTimeMillis();
AdminUtils.createTopic(kafkaTestSupport.getZkClient(), testTopicName, 6, 1, new Properties());
KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
configurationProperties.setAutoAddPartitions(true);
@@ -736,13 +732,14 @@ public class KafkaBinderTests extends
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
Binding<?> binding = binder.doBindConsumer(testTopicName, "test", output, consumerProperties);
binding.unbind();
TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(testTopicName, kafkaTestSupport.getZkClient());
assertThat(topicMetadata.partitionsMetadata().size(), equalTo(6));
TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(testTopicName,
kafkaTestSupport.getZkClient());
assertThat(topicMetadata.partitionsMetadata().size()).isEqualTo(6);
}
@Test
public void testPartitionCountIncreasedIfAutoAddPartitionsSet() throws Exception {
String testTopicName = "existing" + System.currentTimeMillis();
String testTopicName = "existing" + System.currentTimeMillis();
AdminUtils.createTopic(kafkaTestSupport.getZkClient(), testTopicName, 1, 1, new Properties());
KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
configurationProperties.setMinPartitionCount(6);
@@ -762,8 +759,9 @@ public class KafkaBinderTests extends
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
Binding<?> binding = binder.doBindConsumer(testTopicName, "test", output, consumerProperties);
binding.unbind();
TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(testTopicName, kafkaTestSupport.getZkClient());
assertThat(topicMetadata.partitionsMetadata().size(), equalTo(6));
TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(testTopicName,
kafkaTestSupport.getZkClient());
assertThat(topicMetadata.partitionsMetadata().size()).isEqualTo(6);
}
private static final class FailingInvocationCountingMessageHandler implements MessageHandler {
@@ -786,7 +784,8 @@ public class KafkaBinderTests extends
public void handleMessage(Message<?> message) throws MessagingException {
invocationCount++;
Long offset = message.getHeaders().get(KafkaHeaders.OFFSET, Long.class);
// using the offset as key allows to ensure that we don't store duplicate messages on retry
// using the offset as key allows to ensure that we don't store duplicate
// messages on retry
if (!receivedMessages.containsKey(offset)) {
receivedMessages.put(offset, message);
latch.countDown();

93
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java
View File

@@ -21,7 +21,6 @@ import java.util.Arrays;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.cloud.stream.binder.BinderHeaders;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
@@ -35,13 +34,7 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.containsString;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
@@ -88,14 +81,14 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
output.send(new GenericMessage<>(new byte[] { (byte) 2 }));
Message<?> receive0 = receive(input0);
assertNotNull(receive0);
assertThat(receive0).isNotNull();
Message<?> receive1 = receive(input1);
assertNotNull(receive1);
assertThat(receive1).isNotNull();
Message<?> receive2 = receive(input2);
assertNotNull(receive2);
assertThat(receive2).isNotNull();
assertThat(Arrays.asList(((byte[]) receive0.getPayload())[0], ((byte[]) receive1.getPayload())[0],
((byte[]) receive2.getPayload())[0]), containsInAnyOrder((byte) 0, (byte) 1, (byte) 2));
((byte[]) receive2.getPayload())[0])).containsExactlyInAnyOrder((byte) 0, (byte) 1, (byte) 2);
input0Binding.unbind();
input1Binding.unbind();
@@ -118,7 +111,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
Binding<MessageChannel> outputBinding = binder.bindProducer("part.0", output, properties);
try {
AbstractEndpoint endpoint = extractEndpoint(outputBinding);
assertThat(getEndpointRouting(endpoint), containsString("part.0-' + headers['partition']"));
assertThat(getEndpointRouting(endpoint)).contains("part.0-' + headers['partition']");
}
catch (UnsupportedOperationException ignored) {
@@ -142,29 +135,21 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
input2.setBeanName("test.input2S");
Binding<MessageChannel> input2Binding = binder.bindConsumer("part.0", "test", input2, consumerProperties);
Message<byte[]> message2 = MessageBuilder.withPayload(new byte[]{2})
Message<byte[]> message2 = MessageBuilder.withPayload(new byte[] { 2 })
.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "foo")
.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 42)
.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 43)
.build();
.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 43).build();
output.send(message2);
output.send(new GenericMessage<>(new byte[]{1}));
output.send(new GenericMessage<>(new byte[]{0}));
output.send(new GenericMessage<>(new byte[] { 1 }));
output.send(new GenericMessage<>(new byte[] { 0 }));
Message<?> receive0 = receive(input0);
assertNotNull(receive0);
assertThat(receive0).isNotNull();
Message<?> receive1 = receive(input1);
assertNotNull(receive1);
assertThat(receive1).isNotNull();
Message<?> receive2 = receive(input2);
assertNotNull(receive2);
assertThat(Arrays.asList(
((byte[]) receive0.getPayload())[0],
((byte[]) receive1.getPayload())[0],
((byte[]) receive2.getPayload())[0]),
containsInAnyOrder((byte) 0, (byte) 1, (byte) 2));
assertThat(receive2).isNotNull();
assertThat(Arrays.asList(((byte[]) receive0.getPayload())[0], ((byte[]) receive1.getPayload())[0],
((byte[]) receive2.getPayload())[0])).containsExactlyInAnyOrder((byte) 0, (byte) 1, (byte) 2);
input0Binding.unbind();
input1Binding.unbind();
input2Binding.unbind();
@@ -182,14 +167,15 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
Binding<MessageChannel> producerBinding = binder.bindProducer("foo.0", moduleOutputChannel, producerProperties);
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
consumerProperties.setHeaderMode(HeaderMode.raw);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.0", "test", moduleInputChannel, consumerProperties);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.0", "test", moduleInputChannel,
consumerProperties);
Message<?> message = MessageBuilder.withPayload("foo".getBytes()).build();
// Let the consumer actually bind to the producer before sending a msg
binderBindUnbindLatency();
moduleOutputChannel.send(message);
Message<?> inbound = receive(moduleInputChannel);
assertNotNull(inbound);
assertEquals("foo", new String((byte[]) inbound.getPayload()));
assertThat(inbound).isNotNull();
assertThat(new String((byte[]) inbound.getPayload())).isEqualTo("foo");
producerBinding.unbind();
consumerBinding.unbind();
}
@@ -215,13 +201,16 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
Binding<MessageChannel> producerBinding = binder.bindProducer("baz.0", moduleOutputChannel, producerProperties);
ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
consumerProperties.setHeaderMode(HeaderMode.raw);
Binding<MessageChannel> input1Binding = binder.bindConsumer("baz.0", "test", module1InputChannel, consumerProperties);
Binding<MessageChannel> input1Binding = binder.bindConsumer("baz.0", "test", module1InputChannel,
consumerProperties);
// A new module is using the tap as an input channel
String fooTapName = "baz.0";
Binding<MessageChannel> input2Binding = binder.bindConsumer(fooTapName, "tap1", module2InputChannel, consumerProperties);
Binding<MessageChannel> input2Binding = binder.bindConsumer(fooTapName, "tap1", module2InputChannel,
consumerProperties);
// Another new module is using tap as an input channel
String barTapName = "baz.0";
Binding<MessageChannel> input3Binding = binder.bindConsumer(barTapName, "tap2", module3InputChannel, consumerProperties);
Binding<MessageChannel> input3Binding = binder.bindConsumer(barTapName, "tap2", module3InputChannel,
consumerProperties);
Message<?> message = MessageBuilder.withPayload("foo".getBytes()).build();
boolean success = false;
@@ -229,20 +218,20 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
while (!success) {
moduleOutputChannel.send(message);
Message<?> inbound = receive(module1InputChannel);
assertNotNull(inbound);
assertEquals("foo", new String((byte[]) inbound.getPayload()));
assertThat(inbound).isNotNull();
assertThat(new String((byte[]) inbound.getPayload())).isEqualTo("foo");
Message<?> tapped1 = receive(module2InputChannel);
Message<?> tapped2 = receive(module3InputChannel);
if (tapped1 == null || tapped2 == null) {
// listener may not have started
assertFalse("Failed to receive tap after retry", retried);
assertThat(retried).isFalse().withFailMessage("Failed to receive tap after retry");
retried = true;
continue;
}
success = true;
assertEquals("foo", new String((byte[]) tapped1.getPayload()));
assertEquals("foo", new String((byte[]) tapped2.getPayload()));
assertThat(new String((byte[]) tapped1.getPayload())).isEqualTo("foo");
assertThat(new String((byte[]) tapped2.getPayload())).isEqualTo("foo");
}
// delete one tap stream is deleted
input3Binding.unbind();
@@ -251,31 +240,23 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
// other tap still receives messages
Message<?> tapped = receive(module2InputChannel);
assertNotNull(tapped);
assertThat(tapped).isNotNull();
// removed tap does not
assertNull(receive(module3InputChannel));
assertThat(receive(module3InputChannel)).isNull();
// re-subscribed tap does receive the message
input3Binding = binder.bindConsumer(barTapName, "tap2", module3InputChannel, createConsumerProperties());
assertNotNull(receive(module3InputChannel));
assertThat(receive(module3InputChannel)).isNotNull();
// clean up
input1Binding.unbind();
input2Binding.unbind();
input3Binding.unbind();
producerBinding.unbind();
assertFalse(extractEndpoint(input1Binding).isRunning());
assertFalse(extractEndpoint(input2Binding).isRunning());
assertFalse(extractEndpoint(input3Binding).isRunning());
assertFalse(extractEndpoint(producerBinding).isRunning());
assertThat(extractEndpoint(input1Binding).isRunning()).isFalse();
assertThat(extractEndpoint(input2Binding).isRunning()).isFalse();
assertThat(extractEndpoint(input3Binding).isRunning()).isFalse();
assertThat(extractEndpoint(producerBinding).isRunning()).isFalse();
}
private void assertMessageReceive(QueueChannel moduleInputChannel, String payload) {
Message<?> inbound = receive(moduleInputChannel);
assertNotNull(inbound);
assertEquals(payload, new String((byte[]) inbound.getPayload()));
assertNull(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
}
}

4
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactoryIntegrationTests.java
View File

@@ -30,7 +30,7 @@ import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.cloud.stream.test.junit.rabbit.RabbitTestSupport;
import static org.junit.Assert.assertEquals;
import static org.assertj.core.api.Assertions.assertThat;
/**
@@ -65,7 +65,7 @@ public class LocalizedQueueConnectionFactoryIntegrationTests {
ConnectionFactory targetConnectionFactory = this.lqcf.getTargetConnectionFactory("[" + queue.getName() + "]");
RabbitTemplate template = new RabbitTemplate(targetConnectionFactory);
template.convertAndSend("", queue.getName(), "foo");
assertEquals("foo", template.receiveAndConvert(queue.getName()));
assertThat(template.receiveAndConvert(queue.getName())).isEqualTo("foo");
((CachingConnectionFactory) targetConnectionFactory).destroy();
admin.deleteQueue(queue.getName());
}

31
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java
View File

@@ -39,10 +39,7 @@ import org.springframework.cloud.stream.test.junit.rabbit.RabbitTestSupport;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;
import static org.hamcrest.Matchers.containsString;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.fail;
/**
@@ -116,7 +113,7 @@ public class RabbitBinderCleanerTests {
fail("Expected exception");
}
catch (RabbitAdminException e) {
assertEquals("Queue " + queueName + " is in use", e.getMessage());
assertThat(e).hasMessageContaining("Queue " + queueName + " is in use");
}
channel.basicCancel(consumerTag);
waitForConsumerStateNot(queueName, 1);
@@ -125,8 +122,8 @@ public class RabbitBinderCleanerTests {
fail("Expected exception");
}
catch (RabbitAdminException e) {
assertThat(e.getMessage(), containsString("Cannot delete exchange "));
assertThat(e.getMessage(), containsString("; it has bindings:"));
assertThat(e).hasMessageContaining("Cannot delete exchange ");
assertThat(e).hasMessageContaining("; it has bindings:");
}
return null;
}
@@ -144,7 +141,7 @@ public class RabbitBinderCleanerTests {
}
Thread.sleep(100);
}
assertTrue("Consumer state remained at " + state + " after 10 seconds", n < 100);
assertThat(n < 100).withFailMessage("Consumer state remained at " + state + " after 10 seconds");
}
});
@@ -153,27 +150,27 @@ public class RabbitBinderCleanerTests {
rabbitAdmin.deleteQueue(foreignQueue);
connectionFactory.destroy();
Map<String, List<String>> cleanedMap = cleaner.clean(stream1, false);
assertEquals(2, cleanedMap.size());
assertThat(cleanedMap).hasSize(2);
List<String> cleanedQueues = cleanedMap.get("queues");
// should *not* clean stream2
assertEquals(10, cleanedQueues.size());
assertThat(cleanedQueues).hasSize(10);
for (int i = 0; i < 5; i++) {
assertEquals(BINDER_PREFIX + stream1 + ".default." + i, cleanedQueues.get(i * 2));
assertEquals(BINDER_PREFIX + stream1 + ".default." + i + ".dlq", cleanedQueues.get(i * 2 + 1));
assertThat(cleanedQueues.get(i * 2)).isEqualTo(BINDER_PREFIX + stream1 + ".default." + i);
assertThat(cleanedQueues.get(i * 2 + 1)).isEqualTo(BINDER_PREFIX + stream1 + ".default." + i + ".dlq");
}
List<String> cleanedExchanges = cleanedMap.get("exchanges");
assertEquals(6, cleanedExchanges.size());
assertThat(cleanedExchanges).hasSize(6);
// wild card *should* clean stream2
cleanedMap = cleaner.clean(stream1 + "*", false);
assertEquals(2, cleanedMap.size());
assertThat(cleanedMap).hasSize(2);
cleanedQueues = cleanedMap.get("queues");
assertEquals(5, cleanedQueues.size());
assertThat(cleanedQueues).hasSize(5);
for (int i = 0; i < 5; i++) {
assertEquals(BINDER_PREFIX + stream2 + ".default." + i, cleanedQueues.get(i));
assertThat(cleanedQueues.get(i)).isEqualTo(BINDER_PREFIX + stream2 + ".default." + i);
}
cleanedExchanges = cleanedMap.get("exchanges");
assertEquals(6, cleanedExchanges.size());
assertThat(cleanedExchanges).hasSize(6);
}
public static class AmqpQueue {

302
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
View File

@@ -63,15 +63,7 @@ import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.startsWith;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -81,7 +73,8 @@ import static org.mockito.Mockito.when;
* @author Gary Russell
* @author David Turanski
*/
public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBinder, ExtendedConsumerProperties<RabbitConsumerProperties>, ExtendedProducerProperties<RabbitProducerProperties>> {
public class RabbitBinderTests extends
PartitionCapableBinderTests<RabbitTestBinder, ExtendedConsumerProperties<RabbitConsumerProperties>, ExtendedProducerProperties<RabbitProducerProperties>> {
private final String CLASS_UNDER_TEST_NAME = RabbitMessageChannelBinder.class.getSimpleName();
@@ -118,10 +111,12 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
RabbitTestBinder binder = getBinder();
DirectChannel moduleOutputChannel = new DirectChannel();
DirectChannel moduleInputChannel = new DirectChannel();
Binding<MessageChannel> producerBinding = binder.bindProducer("bad.0", moduleOutputChannel, createProducerProperties());
Binding<MessageChannel> consumerBinding = binder.bindConsumer("bad.0", "test", moduleInputChannel, createConsumerProperties());
Message<?> message = MessageBuilder.withPayload("bad").setHeader(MessageHeaders.CONTENT_TYPE,
"foo/bar").build();
Binding<MessageChannel> producerBinding = binder.bindProducer("bad.0", moduleOutputChannel,
createProducerProperties());
Binding<MessageChannel> consumerBinding = binder.bindConsumer("bad.0", "test", moduleInputChannel,
createConsumerProperties());
Message<?> message = MessageBuilder.withPayload("bad").setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar")
.build();
final CountDownLatch latch = new CountDownLatch(3);
moduleInputChannel.subscribe(new MessageHandler() {
@@ -132,7 +127,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
}
});
moduleOutputChannel.send(message);
assertTrue(latch.await(10, TimeUnit.SECONDS));
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
producerBinding.unbind();
consumerBinding.unbind();
}
@@ -146,22 +141,21 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
AbstractEndpoint endpoint = extractEndpoint(consumerBinding);
SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint, "messageListenerContainer",
SimpleMessageListenerContainer.class);
assertEquals(AcknowledgeMode.AUTO, container.getAcknowledgeMode());
assertThat(container.getQueueNames()[0],
startsWith(properties.getExtension().getPrefix()));
assertTrue(TestUtils.getPropertyValue(container, "transactional", Boolean.class));
assertEquals(1, TestUtils.getPropertyValue(container, "concurrentConsumers"));
assertNull(TestUtils.getPropertyValue(container, "maxConcurrentConsumers"));
assertTrue(TestUtils.getPropertyValue(container, "defaultRequeueRejected", Boolean.class));
assertEquals(1, TestUtils.getPropertyValue(container, "prefetchCount"));
assertEquals(1, TestUtils.getPropertyValue(container, "txSize"));
assertThat(container.getAcknowledgeMode()).isEqualTo(AcknowledgeMode.AUTO);
assertThat(container.getQueueNames()[0]).startsWith(properties.getExtension().getPrefix());
assertThat(TestUtils.getPropertyValue(container, "transactional", Boolean.class)).isTrue();
assertThat(TestUtils.getPropertyValue(container, "concurrentConsumers")).isEqualTo(1);
assertThat(TestUtils.getPropertyValue(container, "maxConcurrentConsumers")).isNull();
assertThat(TestUtils.getPropertyValue(container, "defaultRequeueRejected", Boolean.class)).isTrue();
assertThat(TestUtils.getPropertyValue(container, "prefetchCount")).isEqualTo(1);
assertThat(TestUtils.getPropertyValue(container, "txSize")).isEqualTo(1);
Advice retry = TestUtils.getPropertyValue(container, "adviceChain", Advice[].class)[0];
assertEquals(3, TestUtils.getPropertyValue(retry, "retryOperations.retryPolicy.maxAttempts"));
assertEquals(1000L, TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.initialInterval"));
assertEquals(10000L, TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.maxInterval"));
assertEquals(2.0, TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.multiplier"));
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.retryPolicy.maxAttempts")).isEqualTo(3);
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.initialInterval")).isEqualTo(1000L);
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.maxInterval")).isEqualTo(10000L);
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.multiplier")).isEqualTo(2.0);
consumerBinding.unbind();
assertFalse(endpoint.isRunning());
assertThat(endpoint.isRunning()).isFalse();
properties = createConsumerProperties();
properties.getExtension().setAcknowledgeMode(AcknowledgeMode.NONE);
@@ -173,7 +167,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
properties.getExtension().setMaxConcurrency(3);
properties.getExtension().setPrefix("foo.");
properties.getExtension().setPrefetch(20);
properties.getExtension().setRequestHeaderPatterns(new String[] {"foo"});
properties.getExtension().setRequestHeaderPatterns(new String[] { "foo" });
properties.getExtension().setRequeueRejected(false);
properties.getExtension().setTxSize(10);
properties.setInstanceIndex(0);
@@ -182,32 +176,34 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
endpoint = extractEndpoint(consumerBinding);
container = verifyContainer(endpoint);
assertEquals("foo.props.0.test", container.getQueueNames()[0]);
assertThat(container.getQueueNames()[0]).isEqualTo("foo.props.0.test");
consumerBinding.unbind();
assertFalse(endpoint.isRunning());
assertThat(endpoint.isRunning()).isFalse();
}
@Test
public void testProducerProperties() throws Exception {
RabbitTestBinder binder = getBinder();
Binding<MessageChannel> producerBinding = binder.bindProducer("props.0", new DirectChannel(), createProducerProperties());
Binding<MessageChannel> producerBinding = binder.bindProducer("props.0", new DirectChannel(),
createProducerProperties());
@SuppressWarnings("unchecked")
AbstractEndpoint endpoint = extractEndpoint(producerBinding);
MessageDeliveryMode mode = TestUtils.getPropertyValue(endpoint, "handler.delegate.defaultDeliveryMode",
MessageDeliveryMode.class);
assertEquals(MessageDeliveryMode.PERSISTENT, mode);
assertThat(mode).isEqualTo(MessageDeliveryMode.PERSISTENT);
List<?> requestHeaders = TestUtils.getPropertyValue(endpoint,
"handler.delegate.headerMapper.requestHeaderMatcher.matchers", List.class);
assertEquals(2, requestHeaders.size());
assertThat(requestHeaders).hasSize(2);
producerBinding.unbind();
assertFalse(endpoint.isRunning());
assertFalse(TestUtils.getPropertyValue(endpoint, "handler.delegate.amqpTemplate.transactional", Boolean.class));
assertThat(endpoint.isRunning()).isFalse();
assertThat(TestUtils.getPropertyValue(endpoint, "handler.delegate.amqpTemplate.transactional", Boolean.class))
.isFalse();
ExtendedProducerProperties<RabbitProducerProperties> properties = createProducerProperties();
properties.getExtension().setPrefix("foo.");
properties.getExtension().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
properties.getExtension().setRequestHeaderPatterns(new String[] {"foo"});
properties.getExtension().setRequestHeaderPatterns(new String[] { "foo" });
properties.setPartitionKeyExpression(spelExpressionParser.parseExpression("'foo'"));
properties.setPartitionKeyExtractorClass(TestPartitionKeyExtractorClass.class);
properties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("0"));
@@ -217,18 +213,16 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
producerBinding = binder.bindProducer("props.0", new DirectChannel(), properties);
endpoint = extractEndpoint(producerBinding);
assertEquals(
"'props.0-' + headers['partition']",
TestUtils.getPropertyValue(endpoint, "handler.delegate.routingKeyExpression",
SpelExpression.class).getExpressionString());
mode = TestUtils.getPropertyValue(endpoint, "handler.delegate.defaultDeliveryMode",
MessageDeliveryMode.class);
assertEquals(MessageDeliveryMode.NON_PERSISTENT, mode);
assertTrue(TestUtils.getPropertyValue(endpoint, "handler.delegate.amqpTemplate.transactional", Boolean.class));
assertThat(TestUtils.getPropertyValue(endpoint, "handler.delegate.routingKeyExpression", SpelExpression.class)
.getExpressionString()).isEqualTo("'props.0-' + headers['partition']");
mode = TestUtils.getPropertyValue(endpoint, "handler.delegate.defaultDeliveryMode", MessageDeliveryMode.class);
assertThat(mode).isEqualTo(MessageDeliveryMode.NON_PERSISTENT);
assertThat(TestUtils.getPropertyValue(endpoint, "handler.delegate.amqpTemplate.transactional", Boolean.class))
.isTrue();
verifyFooRequestProducer(endpoint);
producerBinding.unbind();
assertFalse(endpoint.isRunning());
assertThat(endpoint.isRunning()).isFalse();
}
@Test
@@ -253,7 +247,8 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
}
});
Binding<MessageChannel> consumerBinding = binder.bindConsumer("durabletest.0", "tgroup", moduleInputChannel, properties);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("durabletest.0", "tgroup", moduleInputChannel,
properties);
RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
template.convertAndSend(TEST_PREFIX + "durabletest.0", "", "foo");
@@ -262,15 +257,15 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
while (n++ < 100) {
Object deadLetter = template.receiveAndConvert(TEST_PREFIX + "durabletest.0.tgroup.dlq");
if (deadLetter != null) {
assertEquals("foo", deadLetter);
assertThat(deadLetter).isEqualTo("foo");
break;
}
Thread.sleep(100);
}
assertTrue(n < 100);
assertThat(n).isLessThan(100);
consumerBinding.unbind();
assertNotNull(admin.getQueueProperties(TEST_PREFIX + "durabletest.0.tgroup.dlq"));
assertThat(admin.getQueueProperties(TEST_PREFIX + "durabletest.0.tgroup.dlq")).isNotNull();
}
@Test
@@ -294,10 +289,11 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
}
});
Binding<MessageChannel> consumerBinding = binder.bindConsumer("nondurabletest.0", "tgroup", moduleInputChannel, properties);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("nondurabletest.0", "tgroup", moduleInputChannel,
properties);
consumerBinding.unbind();
assertNull(admin.getQueueProperties(TEST_PREFIX + "nondurabletest.0.dlq"));
assertThat(admin.getQueueProperties(TEST_PREFIX + "nondurabletest.0.dlq")).isNull();
}
@Test
@@ -319,7 +315,8 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
}
});
Binding<MessageChannel> consumerBinding = binder.bindConsumer("dlqtest", "default", moduleInputChannel, properties);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("dlqtest", "default", moduleInputChannel,
properties);
RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
template.convertAndSend("", TEST_PREFIX + "dlqtest.default", "foo");
@@ -328,21 +325,21 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
while (n++ < 100) {
Object deadLetter = template.receiveAndConvert(TEST_PREFIX + "dlqtest.default.dlq");
if (deadLetter != null) {
assertEquals("foo", deadLetter);
assertThat(deadLetter).isEqualTo("foo");
break;
}
Thread.sleep(100);
}
assertTrue(n < 100);
assertThat(n).isLessThan(100);
consumerBinding.unbind();
ApplicationContext context = TestUtils.getPropertyValue(binder, "binder.autoDeclareContext",
ApplicationContext.class);
assertFalse(context.containsBean(TEST_PREFIX + "dlqtest.default.binding"));
assertFalse(context.containsBean(TEST_PREFIX + "dlqtest.default"));
assertFalse(context.containsBean(TEST_PREFIX + "dlqtest.default.dlq.binding"));
assertFalse(context.containsBean(TEST_PREFIX + "dlqtest.default.dlq"));
assertThat(context.containsBean(TEST_PREFIX + "dlqtest.default.binding")).isFalse();
assertThat(context.containsBean(TEST_PREFIX + "dlqtest.default")).isFalse();
assertThat(context.containsBean(TEST_PREFIX + "dlqtest.default.dlq.binding")).isFalse();
assertThat(context.containsBean(TEST_PREFIX + "dlqtest.default.dlq")).isFalse();
}
@Test
@@ -358,13 +355,14 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
DirectChannel input0 = new DirectChannel();
input0.setBeanName("test.input0DLQ");
Binding<MessageChannel> input0Binding = binder.bindConsumer("partDLQ.0", "dlqPartGrp", input0, properties);
Binding<MessageChannel> defaultConsumerBinding1 =
binder.bindConsumer("partDLQ.0", "default", new QueueChannel(), properties);
Binding<MessageChannel> defaultConsumerBinding1 = binder.bindConsumer("partDLQ.0", "default",
new QueueChannel(), properties);
properties.setInstanceIndex(1);
DirectChannel input1 = new DirectChannel();
input1.setBeanName("test.input1DLQ");
Binding<MessageChannel> input1Binding = binder.bindConsumer("partDLQ.0", "dlqPartGrp", input1, properties);
Binding<MessageChannel> defaultConsumerBinding2 = binder.bindConsumer("partDLQ.0", "default", new QueueChannel(), properties);
Binding<MessageChannel> defaultConsumerBinding2 = binder.bindConsumer("partDLQ.0", "default",
new QueueChannel(), properties);
ExtendedProducerProperties<RabbitProducerProperties> producerProperties = createProducerProperties();
producerProperties.getExtension().setPrefix("bindertest.");
@@ -402,13 +400,13 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
});
output.send(new GenericMessage<Integer>(1));
assertTrue(latch1.await(10, TimeUnit.SECONDS));
output.send(new GenericMessage<>(1));
assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue();
output.send(new GenericMessage<Integer>(0));
assertTrue(latch0.await(10, TimeUnit.SECONDS));
output.send(new GenericMessage<>(0));
assertThat(latch0.await(10, TimeUnit.SECONDS)).isTrue();
output.send(new GenericMessage<Integer>(1));
output.send(new GenericMessage<>(1));
RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
template.setReceiveTimeout(10000);
@@ -416,13 +414,13 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
String streamDLQName = "bindertest.partDLQ.0.dlqPartGrp.dlq";
org.springframework.amqp.core.Message received = template.receive(streamDLQName);
assertNotNull(received);
assertEquals(1, received.getMessageProperties().getHeaders().get("partition"));
assertThat(received).isNotNull();
assertThat(received.getMessageProperties().getHeaders()).containsEntry("partition", 1);
output.send(new GenericMessage<Integer>(0));
output.send(new GenericMessage<>(0));
received = template.receive(streamDLQName);
assertNotNull(received);
assertEquals(0, received.getMessageProperties().getHeaders().get("partition"));
assertThat(received).isNotNull();
assertThat(received.getMessageProperties().getHeaders()).containsEntry("partition", 0);
input0Binding.unbind();
input1Binding.unbind();
@@ -455,13 +453,17 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
consumerProperties.setInstanceIndex(0);
DirectChannel input0 = new DirectChannel();
input0.setBeanName("test.input0DLQ");
Binding<MessageChannel> input0Binding = binder.bindConsumer("partDLQ.1", "dlqPartGrp", input0, consumerProperties);
Binding<MessageChannel> defaultConsumerBinding1 = binder.bindConsumer("partDLQ.1", "defaultConsumer", new QueueChannel(), consumerProperties);
Binding<MessageChannel> input0Binding = binder.bindConsumer("partDLQ.1", "dlqPartGrp", input0,
consumerProperties);
Binding<MessageChannel> defaultConsumerBinding1 = binder.bindConsumer("partDLQ.1", "defaultConsumer",
new QueueChannel(), consumerProperties);
consumerProperties.setInstanceIndex(1);
DirectChannel input1 = new DirectChannel();
input1.setBeanName("test.input1DLQ");
Binding<MessageChannel> input1Binding = binder.bindConsumer("partDLQ.1", "dlqPartGrp", input1, consumerProperties);
Binding<MessageChannel> defaultConsumerBinding2 = binder.bindConsumer("partDLQ.1", "defaultConsumer", new QueueChannel(), consumerProperties);
Binding<MessageChannel> input1Binding = binder.bindConsumer("partDLQ.1", "dlqPartGrp", input1,
consumerProperties);
Binding<MessageChannel> defaultConsumerBinding2 = binder.bindConsumer("partDLQ.1", "defaultConsumer",
new QueueChannel(), consumerProperties);
final CountDownLatch latch0 = new CountDownLatch(1);
input0.subscribe(new MessageHandler() {
@@ -490,10 +492,10 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
});
output.send(new GenericMessage<Integer>(1));
assertTrue(latch1.await(10, TimeUnit.SECONDS));
assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue();
output.send(new GenericMessage<Integer>(0));
assertTrue(latch0.await(10, TimeUnit.SECONDS));
assertThat(latch0.await(10, TimeUnit.SECONDS)).isTrue();
output.send(new GenericMessage<Integer>(1));
@@ -503,13 +505,13 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
String streamDLQName = "bindertest.partDLQ.1.dlqPartGrp.dlq";
org.springframework.amqp.core.Message received = template.receive(streamDLQName);
assertNotNull(received);
assertEquals(1, received.getMessageProperties().getHeaders().get("partition"));
assertThat(received).isNotNull();
assertThat(received.getMessageProperties().getHeaders()).containsEntry("partition", 1);
output.send(new GenericMessage<Integer>(0));
received = template.receive(streamDLQName);
assertNotNull(received);
assertEquals(0, received.getMessageProperties().getHeaders().get("partition"));
assertThat(received).isNotNull();
assertThat(received.getMessageProperties().getHeaders()).containsEntry("partition", 0);
input0Binding.unbind();
input1Binding.unbind();
@@ -546,7 +548,8 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
}
});
Binding<MessageChannel> consumerBinding = binder.bindConsumer("dlqpubtest", "default", moduleInputChannel, properties);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("dlqpubtest", "default", moduleInputChannel,
properties);
RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
template.convertAndSend("", TEST_PREFIX + "dlqpubtest.default", "foo");
@@ -555,13 +558,13 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
while (n++ < 100) {
org.springframework.amqp.core.Message deadLetter = template.receive(TEST_PREFIX + "dlqpubtest.default.dlq");
if (deadLetter != null) {
assertEquals("foo", new String(deadLetter.getBody()));
assertNotNull(deadLetter.getMessageProperties().getHeaders().get("x-exception-stacktrace"));
assertThat(new String(deadLetter.getBody())).isEqualTo("foo");
assertThat(deadLetter.getMessageProperties().getHeaders()).containsKey(("x-exception-stacktrace"));
break;
}
Thread.sleep(100);
}
assertTrue(n < 100);
assertThat(n).isLessThan(100);
consumerBinding.unbind();
}
@@ -589,41 +592,43 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
.setPropertyValue("logger", logger);
when(logger.isTraceEnabled()).thenReturn(true);
assertEquals(Deflater.BEST_SPEED, TestUtils.getPropertyValue(binder, "binder.compressingPostProcessor.level"));
assertThat(TestUtils.getPropertyValue(binder, "binder.compressingPostProcessor.level"))
.isEqualTo(Deflater.BEST_SPEED);
output.send(new GenericMessage<>("foo".getBytes()));
output.send(new GenericMessage<>("bar".getBytes()));
Object out = spyOn("batching.0.default").receive(false);
assertThat(out, instanceOf(byte[].class));
assertEquals("\u0000\u0000\u0000\u0003foo\u0000\u0000\u0000\u0003bar", new String((byte[]) out));
assertThat(out).isInstanceOf(byte[].class);
assertThat(new String((byte[]) out)).isEqualTo("\u0000\u0000\u0000\u0003foo\u0000\u0000\u0000\u0003bar");
ArgumentCaptor<Object> captor = ArgumentCaptor.forClass(Object.class);
verify(logger).trace(captor.capture());
assertThat(captor.getValue().toString(), containsString("Compressed 14 to "));
assertThat(captor.getValue().toString()).contains(("Compressed 14 to "));
QueueChannel input = new QueueChannel();
input.setBeanName("batchingConsumer");
Binding<MessageChannel> consumerBinding = binder.bindConsumer("batching.0", "test", input, createConsumerProperties());
Binding<MessageChannel> consumerBinding = binder.bindConsumer("batching.0", "test", input,
createConsumerProperties());
output.send(new GenericMessage<>("foo".getBytes()));
output.send(new GenericMessage<>("bar".getBytes()));
Message<byte[]> in = (Message<byte[]>) input.receive(10000);
assertNotNull(in);
assertEquals("foo", new String(in.getPayload()));
assertThat(in).isNotNull();
assertThat(new String(in.getPayload())).isEqualTo("foo");
in = (Message<byte[]>) input.receive(10000);
assertNotNull(in);
assertEquals("bar", new String(in.getPayload()));
assertNull(in.getHeaders().get(AmqpHeaders.DELIVERY_MODE));
assertThat(in).isNotNull();
assertThat(new String(in.getPayload())).isEqualTo("bar");
assertThat(in.getHeaders().get(AmqpHeaders.DELIVERY_MODE)).isNull();
producerBinding.unbind();
consumerBinding.unbind();
}
/*
* Test late binding due to broker down; queues with and without DLQs, and
* partitioned queues.
* Test late binding due to broker down; queues with and without DLQs, and partitioned
* queues.
*/
@Test
public void testLateBinding() throws Exception {
@@ -642,14 +647,16 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
QueueChannel moduleInputChannel = new QueueChannel();
ExtendedConsumerProperties<RabbitConsumerProperties> rabbitConsumerProperties = createConsumerProperties();
rabbitConsumerProperties.getExtension().setPrefix("latebinder.");
Binding<MessageChannel> late0ConsumerBinding = binder.bindConsumer("late.0", "test", moduleInputChannel, rabbitConsumerProperties);
Binding<MessageChannel> late0ConsumerBinding = binder.bindConsumer("late.0", "test", moduleInputChannel,
rabbitConsumerProperties);
properties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload.equals('0') ? 0 : 1"));
properties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("hashCode()"));
properties.setPartitionCount(2);
MessageChannel partOutputChannel = new DirectChannel();
Binding<MessageChannel> partlate0ProducerBinding = binder.bindProducer("partlate.0", partOutputChannel, properties);
Binding<MessageChannel> partlate0ProducerBinding = binder.bindProducer("partlate.0", partOutputChannel,
properties);
QueueChannel partInputChannel0 = new QueueChannel();
QueueChannel partInputChannel1 = new QueueChannel();
@@ -658,57 +665,64 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
partLateConsumerProperties.getExtension().setPrefix("latebinder.");
partLateConsumerProperties.setPartitioned(true);
partLateConsumerProperties.setInstanceIndex(0);
Binding<MessageChannel> partlate0Consumer0Binding = binder.bindConsumer("partlate.0", "test", partInputChannel0, partLateConsumerProperties);
Binding<MessageChannel> partlate0Consumer0Binding = binder.bindConsumer("partlate.0", "test", partInputChannel0,
partLateConsumerProperties);
partLateConsumerProperties.setInstanceIndex(1);
Binding<MessageChannel> partlate0Consumer1Binding = binder.bindConsumer("partlate.0", "test", partInputChannel1, partLateConsumerProperties);
Binding<MessageChannel> partlate0Consumer1Binding = binder.bindConsumer("partlate.0", "test", partInputChannel1,
partLateConsumerProperties);
ExtendedProducerProperties<RabbitProducerProperties> noDlqProducerProperties = createProducerProperties();
noDlqProducerProperties.getExtension().setPrefix("latebinder.");
MessageChannel noDLQOutputChannel = new DirectChannel();
Binding<MessageChannel> noDlqProducerBinding = binder.bindProducer("lateNoDLQ.0", noDLQOutputChannel, noDlqProducerProperties);
Binding<MessageChannel> noDlqProducerBinding = binder.bindProducer("lateNoDLQ.0", noDLQOutputChannel,
noDlqProducerProperties);
QueueChannel noDLQInputChannel = new QueueChannel();
ExtendedConsumerProperties<RabbitConsumerProperties> noDlqConsumerProperties = createConsumerProperties();
noDlqConsumerProperties.getExtension().setPrefix("latebinder.");
Binding<MessageChannel> noDlqConsumerBinding = binder.bindConsumer("lateNoDLQ.0", "test", noDLQInputChannel, noDlqConsumerProperties);
Binding<MessageChannel> noDlqConsumerBinding = binder.bindConsumer("lateNoDLQ.0", "test", noDLQInputChannel,
noDlqConsumerProperties);
MessageChannel outputChannel = new DirectChannel();
Binding<MessageChannel> pubSubProducerBinding = binder.bindProducer("latePubSub", outputChannel, noDlqProducerProperties);
Binding<MessageChannel> pubSubProducerBinding = binder.bindProducer("latePubSub", outputChannel,
noDlqProducerProperties);
QueueChannel pubSubInputChannel = new QueueChannel();
noDlqConsumerProperties.getExtension().setDurableSubscription(false);
Binding<MessageChannel> nonDurableConsumerBinding = binder.bindConsumer("latePubSub", "lategroup", pubSubInputChannel, noDlqConsumerProperties);
Binding<MessageChannel> nonDurableConsumerBinding = binder.bindConsumer("latePubSub", "lategroup",
pubSubInputChannel, noDlqConsumerProperties);
QueueChannel durablePubSubInputChannel = new QueueChannel();
noDlqConsumerProperties.getExtension().setDurableSubscription(true);
Binding<MessageChannel> durableConsumerBinding = binder.bindConsumer("latePubSub", "lateDurableGroup", durablePubSubInputChannel, noDlqConsumerProperties);
Binding<MessageChannel> durableConsumerBinding = binder.bindConsumer("latePubSub", "lateDurableGroup",
durablePubSubInputChannel, noDlqConsumerProperties);
proxy.start();
moduleOutputChannel.send(new GenericMessage<>("foo"));
Message<?> message = moduleInputChannel.receive(10000);
assertNotNull(message);
assertEquals("foo", message.getPayload());
assertThat(message).isNotNull();
assertThat(message.getPayload()).isNotNull();
noDLQOutputChannel.send(new GenericMessage<>("bar"));
message = noDLQInputChannel.receive(10000);
assertNotNull(message);
assertEquals("bar", message.getPayload());
assertThat(message);
assertThat(message.getPayload()).isEqualTo("bar");
outputChannel.send(new GenericMessage<>("baz"));
message = pubSubInputChannel.receive(10000);
assertNotNull(message);
assertEquals("baz", message.getPayload());
assertThat(message);
assertThat(message.getPayload()).isEqualTo("baz");
message = durablePubSubInputChannel.receive(10000);
assertNotNull(message);
assertEquals("baz", message.getPayload());
assertThat(message).isNotNull();
assertThat(message.getPayload()).isEqualTo("baz");
partOutputChannel.send(new GenericMessage<>("0"));
partOutputChannel.send(new GenericMessage<>("1"));
message = partInputChannel0.receive(10000);
assertNotNull(message);
assertEquals("0", message.getPayload());
assertThat(message).isNotNull();
assertThat(message.getPayload()).isEqualTo("0");
message = partInputChannel1.receive(10000);
assertNotNull(message);
assertEquals("1", message.getPayload());
assertThat(message).isNotNull();
assertThat(message.getPayload()).isEqualTo("1");
late0ProducerBinding.unbind();
late0ConsumerBinding.unbind();
@@ -734,41 +748,39 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
Advice retry;
container = TestUtils.getPropertyValue(endpoint, "messageListenerContainer",
SimpleMessageListenerContainer.class);
assertEquals(AcknowledgeMode.NONE, container.getAcknowledgeMode());
assertThat(container.getQueueNames()[0], startsWith("foo.props.0"));
assertFalse(TestUtils.getPropertyValue(container, "transactional", Boolean.class));
assertEquals(2, TestUtils.getPropertyValue(container, "concurrentConsumers"));
assertEquals(3, TestUtils.getPropertyValue(container, "maxConcurrentConsumers"));
assertFalse(TestUtils.getPropertyValue(container, "defaultRequeueRejected", Boolean.class));
assertEquals(20, TestUtils.getPropertyValue(container, "prefetchCount"));
assertEquals(10, TestUtils.getPropertyValue(container, "txSize"));
assertThat(container.getAcknowledgeMode()).isEqualTo(AcknowledgeMode.NONE);
assertThat(container.getQueueNames()[0]).startsWith("foo.props.0");
assertThat(TestUtils.getPropertyValue(container, "transactional", Boolean.class)).isFalse();
assertThat(TestUtils.getPropertyValue(container, "concurrentConsumers")).isEqualTo(2);
assertThat(TestUtils.getPropertyValue(container, "maxConcurrentConsumers")).isEqualTo(3);
assertThat(TestUtils.getPropertyValue(container, "defaultRequeueRejected", Boolean.class)).isFalse();
assertThat(TestUtils.getPropertyValue(container, "prefetchCount")).isEqualTo(20);
assertThat(TestUtils.getPropertyValue(container, "txSize")).isEqualTo(10);
retry = TestUtils.getPropertyValue(container, "adviceChain", Advice[].class)[0];
assertEquals(23, TestUtils.getPropertyValue(retry, "retryOperations.retryPolicy.maxAttempts"));
assertEquals(2000L, TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.initialInterval"));
assertEquals(20000L, TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.maxInterval"));
assertEquals(5.0, TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.multiplier"));
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.retryPolicy.maxAttempts")).isEqualTo(23);
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.initialInterval")).isEqualTo(2000L);
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.maxInterval")).isEqualTo(20000L);
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.multiplier")).isEqualTo(5.0);
List<?> requestMatchers = TestUtils.getPropertyValue(endpoint,
"headerMapper.requestHeaderMatcher.matchers",
List<?> requestMatchers = TestUtils.getPropertyValue(endpoint, "headerMapper.requestHeaderMatcher.matchers",
List.class);
assertEquals(1, requestMatchers.size());
assertEquals("foo", TestUtils.getPropertyValue(requestMatchers.get(0), "pattern"));
assertThat(requestMatchers).hasSize(1);
assertThat(TestUtils.getPropertyValue(requestMatchers.get(0), "pattern")).isEqualTo("foo");
return container;
}
private void verifyFooRequestProducer(AbstractEndpoint endpoint) {
List<?> requestMatchers = TestUtils.getPropertyValue(endpoint,
"handler.delegate.headerMapper.requestHeaderMatcher.matchers",
List.class);
assertEquals(1, requestMatchers.size());
assertEquals("foo", TestUtils.getPropertyValue(requestMatchers.get(0), "pattern"));
"handler.delegate.headerMapper.requestHeaderMatcher.matchers", List.class);
assertThat(requestMatchers).hasSize(1);
assertThat(TestUtils.getPropertyValue(requestMatchers.get(0), "pattern")).isEqualTo("foo");
}
@Override
protected String getEndpointRouting(AbstractEndpoint endpoint) {
return TestUtils.getPropertyValue(endpoint, "handler.delegate.routingKeyExpression",
SpelExpression.class).getExpressionString();
return TestUtils.getPropertyValue(endpoint, "handler.delegate.routingKeyExpression", SpelExpression.class)
.getExpressionString();
}
@Override
@@ -778,8 +790,8 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
@Override
protected String getPubSubEndpointRouting(AbstractEndpoint endpoint) {
return TestUtils.getPropertyValue(endpoint, "handler.delegate.exchangeNameExpression",
SpelExpression.class).getExpressionString();
return TestUtils.getPropertyValue(endpoint, "handler.delegate.exchangeNameExpression", SpelExpression.class)
.getExpressionString();
}
@Override
@@ -805,7 +817,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
bar = template.receiveAndConvert(new RabbitConsumerProperties().getPrefix() + queue);
Thread.sleep(100);
}
assertTrue("Message did not arrive in RabbitMQ", n < 100);
assertThat(n).isLessThan(100).withFailMessage("Message did not arrive in RabbitMQ");
return bar;
}

61
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java
View File

@@ -48,14 +48,7 @@ import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.MessageChannel;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasKey;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.not;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
@@ -89,22 +82,22 @@ public class RabbitBinderModuleTests {
context = SpringApplication.run(SimpleProcessor.class, "--server.port=0");
BinderFactory<?> binderFactory = context.getBean(BinderFactory.class);
Binder binder = binderFactory.getBinder(null);
assertThat(binder, instanceOf(RabbitMessageChannelBinder.class));
assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class);
DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor
.getPropertyValue("connectionFactory");
assertThat(binderConnectionFactory, instanceOf(CachingConnectionFactory.class));
assertThat(binderConnectionFactory).isInstanceOf(CachingConnectionFactory.class);
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
assertThat(binderConnectionFactory, is(connectionFactory));
assertThat(binderConnectionFactory).isSameAs(connectionFactory);
CompositeHealthIndicator bindersHealthIndicator = context.getBean("bindersHealthIndicator",
CompositeHealthIndicator.class);
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
assertNotNull(bindersHealthIndicator);
assertThat(bindersHealthIndicator).isNotNull();
@SuppressWarnings("unchecked")
Map<String, HealthIndicator> healthIndicators = (Map<String, HealthIndicator>) directFieldAccessor
.getPropertyValue("indicators");
assertThat(healthIndicators, hasKey("rabbit"));
assertThat(healthIndicators.get("rabbit").health().getStatus(), equalTo(Status.UP));
assertThat(healthIndicators).containsKey(("rabbit"));
assertThat(healthIndicators.get("rabbit").health().getStatus()).isEqualTo((Status.UP));
}
@Test
@@ -115,7 +108,7 @@ public class RabbitBinderModuleTests {
"--spring.cloud.stream.rabbit.bindings.output.producer.transacted=true");
BinderFactory<?> binderFactory = context.getBean(BinderFactory.class);
Binder binder = binderFactory.getBinder(null);
assertThat(binder, instanceOf(RabbitMessageChannelBinder.class));
assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class);
ChannelBindingService channelBindingService = context.getBean(ChannelBindingService.class);
DirectFieldAccessor channelBindingServiceAccessor = new DirectFieldAccessor(channelBindingService);
Map<String, List<Binding<MessageChannel>>> consumerBindings = (Map<String, List<Binding<MessageChannel>>>) channelBindingServiceAccessor
@@ -123,27 +116,27 @@ public class RabbitBinderModuleTests {
Binding<MessageChannel> inputBinding = consumerBindings.get("input").get(0);
SimpleMessageListenerContainer container = TestUtils.getPropertyValue(inputBinding,
"endpoint.messageListenerContainer", SimpleMessageListenerContainer.class);
assertTrue(TestUtils.getPropertyValue(container, "transactional", Boolean.class));
assertThat(TestUtils.getPropertyValue(container, "transactional", Boolean.class)).isTrue();
Map<String, Binding<MessageChannel>> producerBindings = (Map<String, Binding<MessageChannel>>) TestUtils
.getPropertyValue(channelBindingService, "producerBindings");
Binding<MessageChannel> outputBinding = producerBindings.get("output");
assertTrue(TestUtils.getPropertyValue(outputBinding, "endpoint.handler.delegate.amqpTemplate.transactional",
Boolean.class));
assertThat(TestUtils.getPropertyValue(outputBinding, "endpoint.handler.delegate.amqpTemplate.transactional",
Boolean.class)).isTrue();
DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor
.getPropertyValue("connectionFactory");
assertThat(binderConnectionFactory, instanceOf(CachingConnectionFactory.class));
assertThat(binderConnectionFactory).isInstanceOf(CachingConnectionFactory.class);
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
assertThat(binderConnectionFactory, is(connectionFactory));
assertThat(binderConnectionFactory).isSameAs(connectionFactory);
CompositeHealthIndicator bindersHealthIndicator = context.getBean("bindersHealthIndicator",
CompositeHealthIndicator.class);
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
assertNotNull(bindersHealthIndicator);
assertThat(bindersHealthIndicator).isNotNull();
@SuppressWarnings("unchecked")
Map<String, HealthIndicator> healthIndicators = (Map<String, HealthIndicator>) directFieldAccessor
.getPropertyValue("indicators");
assertThat(healthIndicators, hasKey("rabbit"));
assertThat(healthIndicators.get("rabbit").health().getStatus(), equalTo(Status.UP));
assertThat(healthIndicators).containsKey("rabbit");
assertThat(healthIndicators.get("rabbit").health().getStatus()).isEqualTo(Status.UP);
}
@Test
@@ -152,23 +145,23 @@ public class RabbitBinderModuleTests {
.run("--server.port=0");
BinderFactory<?> binderFactory = context.getBean(BinderFactory.class);
Binder binder = binderFactory.getBinder(null);
assertThat(binder, instanceOf(RabbitMessageChannelBinder.class));
assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class);
DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor
.getPropertyValue("connectionFactory");
assertThat(binderConnectionFactory, is(MOCK_CONNECTION_FACTORY));
assertThat(binderConnectionFactory).isSameAs(MOCK_CONNECTION_FACTORY);
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
assertThat(binderConnectionFactory, is(connectionFactory));
assertThat(binderConnectionFactory).isSameAs(connectionFactory);
CompositeHealthIndicator bindersHealthIndicator = context.getBean("bindersHealthIndicator",
CompositeHealthIndicator.class);
assertNotNull(bindersHealthIndicator);
assertThat(bindersHealthIndicator).isNotNull();
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
@SuppressWarnings("unchecked")
Map<String, HealthIndicator> healthIndicators = (Map<String, HealthIndicator>) directFieldAccessor
.getPropertyValue("indicators");
assertThat(healthIndicators, hasKey("rabbit"));
assertThat(healthIndicators).containsKey("rabbit");
// mock connection factory behaves as if down
assertThat(healthIndicators.get("rabbit").health().getStatus(), equalTo(Status.DOWN));
assertThat(healthIndicators.get("rabbit").health().getStatus()).isEqualTo(Status.DOWN);
}
@Test
@@ -182,21 +175,21 @@ public class RabbitBinderModuleTests {
context = SpringApplication.run(SimpleProcessor.class, params.toArray(new String[params.size()]));
BinderFactory<?> binderFactory = context.getBean(BinderFactory.class);
Binder binder = binderFactory.getBinder(null);
assertThat(binder, instanceOf(RabbitMessageChannelBinder.class));
assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class);
DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor
.getPropertyValue("connectionFactory");
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
assertThat(binderConnectionFactory, not(is(connectionFactory)));
assertThat(binderConnectionFactory).isNotSameAs(connectionFactory);
CompositeHealthIndicator bindersHealthIndicator = context.getBean("bindersHealthIndicator",
CompositeHealthIndicator.class);
assertNotNull(bindersHealthIndicator);
assertThat(bindersHealthIndicator);
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
@SuppressWarnings("unchecked")
Map<String, HealthIndicator> healthIndicators = (Map<String, HealthIndicator>) directFieldAccessor
.getPropertyValue("indicators");
assertThat(healthIndicators, hasKey("custom"));
assertThat(healthIndicators.get("custom").health().getStatus(), equalTo(Status.UP));
assertThat(healthIndicators).containsKey("custom");
assertThat(healthIndicators.get("custom").health().getStatus()).isEqualTo(Status.UP);
}
@EnableBinding(Processor.class)

8
spring-cloud-stream-binders/spring-cloud-stream-binder-test/pom.xml
View File

@@ -1,5 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>spring-cloud-stream-binder-test</artifactId>
@@ -33,5 +34,10 @@
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>compile</scope>
</dependency>
</dependencies>
</project>

76
spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java
View File

@@ -18,7 +18,6 @@ package org.springframework.cloud.stream.binder;
import java.util.UUID;
import org.hamcrest.collection.IsArrayContainingInAnyOrder;
import org.junit.After;
import org.junit.Test;
@@ -32,19 +31,14 @@ import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.PollableChannel;
import org.springframework.util.MimeTypeUtils;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasProperty;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Gary Russell
* @author Ilayaperumal Gopinathan
* @author David Turanski
* @author Mark Fisher
* @author Marius Bogoevici
*/
@SuppressWarnings("unchecked")
public abstract class AbstractBinderTests<B extends AbstractTestBinder<? extends AbstractBinder<MessageChannel, CP, PP>, CP, PP>, CP extends ConsumerProperties, PP extends ProducerProperties> {
@@ -79,21 +73,31 @@ public abstract class AbstractBinderTests<B extends AbstractTestBinder<? extends
@Test
public void testClean() throws Exception {
Binder binder = getBinder();
Binding<MessageChannel> foo0ProducerBinding = binder.bindProducer("foo.0", new DirectChannel(), createProducerProperties());
Binding<MessageChannel> foo0ConsumerBinding = binder.bindConsumer("foo.0", "test", new DirectChannel(), createConsumerProperties());
Binding<MessageChannel> foo1ProducerBinding = binder.bindProducer("foo.1", new DirectChannel(), createProducerProperties());
Binding<MessageChannel> foo1ConsumerBinding = binder.bindConsumer("foo.1", "test", new DirectChannel(), createConsumerProperties());
Binding<MessageChannel> foo2ProducerBinding = binder.bindProducer("foo.2", new DirectChannel(), createProducerProperties());
Binding<MessageChannel> foo0ProducerBinding = binder.bindProducer("foo.0", new DirectChannel(),
createProducerProperties());
Binding<MessageChannel> foo0ConsumerBinding = binder.bindConsumer("foo.0", "test", new DirectChannel(),
createConsumerProperties());
Binding<MessageChannel> foo1ProducerBinding = binder.bindProducer("foo.1", new DirectChannel(),
createProducerProperties());
Binding<MessageChannel> foo1ConsumerBinding = binder.bindConsumer("foo.1", "test", new DirectChannel(),
createConsumerProperties());
Binding<MessageChannel> foo2ProducerBinding = binder.bindProducer("foo.2", new DirectChannel(),
createProducerProperties());
foo0ProducerBinding.unbind();
assertFalse(TestUtils.getPropertyValue(foo0ProducerBinding, "endpoint", AbstractEndpoint.class).isRunning());
assertThat(TestUtils.getPropertyValue(foo0ProducerBinding, "endpoint", AbstractEndpoint.class).isRunning())
.isFalse();
foo0ConsumerBinding.unbind();
foo1ProducerBinding.unbind();
assertFalse(TestUtils.getPropertyValue(foo0ConsumerBinding, "endpoint", AbstractEndpoint.class).isRunning());
assertFalse(TestUtils.getPropertyValue(foo1ProducerBinding, "endpoint", AbstractEndpoint.class).isRunning());
assertThat(TestUtils.getPropertyValue(foo0ConsumerBinding, "endpoint", AbstractEndpoint.class).isRunning())
.isFalse();
assertThat(TestUtils.getPropertyValue(foo1ProducerBinding, "endpoint", AbstractEndpoint.class).isRunning())
.isFalse();
foo1ConsumerBinding.unbind();
foo2ProducerBinding.unbind();
assertFalse(TestUtils.getPropertyValue(foo1ConsumerBinding, "endpoint", AbstractEndpoint.class).isRunning());
assertFalse(TestUtils.getPropertyValue(foo2ProducerBinding, "endpoint", AbstractEndpoint.class).isRunning());
assertThat(TestUtils.getPropertyValue(foo1ConsumerBinding, "endpoint", AbstractEndpoint.class).isRunning())
.isFalse();
assertThat(TestUtils.getPropertyValue(foo2ProducerBinding, "endpoint", AbstractEndpoint.class).isRunning())
.isFalse();
}
@Test
@@ -101,18 +105,20 @@ public abstract class AbstractBinderTests<B extends AbstractTestBinder<? extends
Binder binder = getBinder();
DirectChannel moduleOutputChannel = new DirectChannel();
QueueChannel moduleInputChannel = new QueueChannel();
Binding<MessageChannel> producerBinding = binder.bindProducer("foo.0", moduleOutputChannel, createProducerProperties());
Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.0", "test", moduleInputChannel, createConsumerProperties());
Message<?> message = MessageBuilder.withPayload("foo").setHeader(MessageHeaders.CONTENT_TYPE,
"foo/bar").build();
Binding<MessageChannel> producerBinding = binder.bindProducer("foo.0", moduleOutputChannel,
createProducerProperties());
Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.0", "test", moduleInputChannel,
createConsumerProperties());
Message<?> message = MessageBuilder.withPayload("foo").setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar")
.build();
// Let the consumer actually bind to the producer before sending a msg
binderBindUnbindLatency();
moduleOutputChannel.send(message);
Message<?> inbound = receive(moduleInputChannel);
assertNotNull(inbound);
assertEquals("foo", inbound.getPayload());
assertNull(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
assertEquals("foo/bar", inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE));
assertThat(inbound).isNotNull();
assertThat(inbound.getPayload()).isEqualTo("foo");
assertThat(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull();
assertThat(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE)).isEqualTo("foo/bar");
producerBinding.unbind();
consumerBinding.unbind();
}
@@ -147,12 +153,10 @@ public abstract class AbstractBinderTests<B extends AbstractTestBinder<? extends
messages[0] = receive(moduleInputChannel);
messages[1] = receive(moduleInputChannel);
assertNotNull(messages[0]);
assertNotNull(messages[1]);
assertThat(messages, IsArrayContainingInAnyOrder.arrayContainingInAnyOrder(
hasProperty("payload", equalTo(testPayload1.getBytes())),
hasProperty("payload", equalTo(testPayload2.getBytes()))));
assertThat(messages[0]).isNotNull();
assertThat(messages[1]).isNotNull();
assertThat(messages).extracting("payload").containsExactlyInAnyOrder(testPayload1.getBytes(),
testPayload2.getBytes());
producerBinding1.unbind();
producerBinding2.unbind();
@@ -174,10 +178,10 @@ public abstract class AbstractBinderTests<B extends AbstractTestBinder<? extends
Message<?> message = MessageBuilder.withPayload("foo").build();
moduleOutputChannel.send(message);
Message<?> inbound = receive(moduleInputChannel);
assertNotNull(inbound);
assertEquals("foo", inbound.getPayload());
assertNull(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
assertEquals(MimeTypeUtils.TEXT_PLAIN_VALUE, inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE));
assertThat(inbound).isNotNull();
assertThat(inbound.getPayload()).isEqualTo("foo");
assertThat(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull();
assertThat(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE)).isEqualTo(MimeTypeUtils.TEXT_PLAIN_VALUE);
producerBinding.unbind();
consumerBinding.unbind();
}

111
spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java
View File

@@ -17,10 +17,10 @@
package org.springframework.cloud.stream.binder;
import java.util.Arrays;
import java.util.List;
import java.util.UUID;
import org.hamcrest.CustomMatcher;
import org.hamcrest.Matcher;
import org.assertj.core.api.Condition;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
@@ -34,15 +34,7 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasProperty;
import static org.hamcrest.Matchers.not;
import static org.hamcrest.Matchers.nullValue;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests for binders that support partitioning.
@@ -76,12 +68,12 @@ abstract public class PartitionCapableBinderTests<B extends AbstractTestBinder<?
output.send(new GenericMessage<>(testPayload1.getBytes()));
Message<byte[]> receivedMessage1 = (Message<byte[]>) receive(input1);
assertThat(receivedMessage1, not(nullValue()));
assertThat(new String(receivedMessage1.getPayload()), equalTo(testPayload1));
assertThat(receivedMessage1).isNotNull();
assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload1);
Message<byte[]> receivedMessage2 = (Message<byte[]>) receive(input2);
assertThat(receivedMessage2, not(nullValue()));
assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload1));
assertThat(receivedMessage2).isNotNull();
assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload1);
binding2.unbind();
@@ -93,15 +85,15 @@ abstract public class PartitionCapableBinderTests<B extends AbstractTestBinder<?
output.send(new GenericMessage<>(testPayload3.getBytes()));
receivedMessage1 = (Message<byte[]>) receive(input1);
assertThat(receivedMessage1, not(nullValue()));
assertThat(new String(receivedMessage1.getPayload()), equalTo(testPayload2));
assertThat(receivedMessage1).isNotNull();
assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload2);
receivedMessage1 = (Message<byte[]>) receive(input1);
assertThat(receivedMessage1, not(nullValue()));
assertThat(new String(receivedMessage1.getPayload()), equalTo(testPayload3));
assertThat(receivedMessage1).isNotNull();
assertThat(new String(receivedMessage1.getPayload())).isNotNull();
receivedMessage2 = (Message<byte[]>) receive(input2);
assertThat(receivedMessage2, not(nullValue()));
assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload3));
assertThat(receivedMessage2).isNotNull();
assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload3);
producerBinding.unbind();
binding1.unbind();
@@ -128,8 +120,8 @@ abstract public class PartitionCapableBinderTests<B extends AbstractTestBinder<?
createConsumerProperties());
Message<?> receivedMessage1 = receive(inbound1);
assertThat(receivedMessage1, not(nullValue()));
assertThat(new String((byte[]) receivedMessage1.getPayload()), equalTo(testPayload));
assertThat(receivedMessage1).isNotNull();
assertThat(new String((byte[]) receivedMessage1.getPayload())).isEqualTo(testPayload);
producerBinding.unbind();
consumerBinding.unbind();
@@ -157,11 +149,11 @@ abstract public class PartitionCapableBinderTests<B extends AbstractTestBinder<?
createConsumerProperties());
Message<?> receivedMessage1 = receive(inbound1);
assertThat(receivedMessage1, not(nullValue()));
assertThat(new String((byte[]) receivedMessage1.getPayload()), equalTo(testPayload));
assertThat(receivedMessage1).isNotNull();
assertThat(new String((byte[]) receivedMessage1.getPayload())).isEqualTo(testPayload);
Message<?> receivedMessage2 = receive(inbound2);
assertThat(receivedMessage2, not(nullValue()));
assertThat(new String((byte[]) receivedMessage2.getPayload()), equalTo(testPayload));
assertThat(receivedMessage2).isNotNull();
assertThat(new String((byte[]) receivedMessage2.getPayload())).isEqualTo(testPayload);
consumerBinding1.unbind();
consumerBinding2.unbind();
@@ -199,8 +191,8 @@ abstract public class PartitionCapableBinderTests<B extends AbstractTestBinder<?
Binding<MessageChannel> outputBinding = binder.bindProducer("part.0", output, producerProperties);
try {
AbstractEndpoint endpoint = extractEndpoint(outputBinding);
assertThat(getEndpointRouting(endpoint),
containsString(getExpectedRoutingBaseDestination("part.0", "test") + "-' + headers['partition']"));
assertThat(getEndpointRouting(endpoint))
.contains(getExpectedRoutingBaseDestination("part.0", "test") + "-' + headers['partition']");
}
catch (UnsupportedOperationException ignored) {
}
@@ -214,36 +206,38 @@ abstract public class PartitionCapableBinderTests<B extends AbstractTestBinder<?
output.send(new GenericMessage<>(0));
Message<?> receive0 = receive(input0);
assertNotNull(receive0);
assertThat(receive0).isNotNull();
Message<?> receive1 = receive(input1);
assertNotNull(receive1);
assertThat(receive1).isNotNull();
Message<?> receive2 = receive(input2);
assertNotNull(receive2);
Matcher<Message<?>> fooMatcher = new CustomMatcher<Message<?>>("the message with 'foo' as its correlationId") {
assertThat(receive2).isNotNull();
Condition<Message<?>> correlationHeadersForPayload2 = new Condition<Message<?>>() {
@Override
public boolean matches(Object item) {
IntegrationMessageHeaderAccessor accessor = new IntegrationMessageHeaderAccessor((Message<?>) item);
boolean result = "foo".equals(accessor.getCorrelationId()) && 42 == accessor.getSequenceNumber()
public boolean matches(Message<?> value) {
IntegrationMessageHeaderAccessor accessor = new IntegrationMessageHeaderAccessor(value);
return "foo".equals(accessor.getCorrelationId()) && 42 == accessor.getSequenceNumber()
&& 43 == accessor.getSequenceSize();
return result;
}
};
if (usesExplicitRouting()) {
assertEquals(0, receive0.getPayload());
assertEquals(1, receive1.getPayload());
assertEquals(2, receive2.getPayload());
assertThat(receive2, fooMatcher);
assertThat(receive0.getPayload()).isEqualTo(0);
assertThat(receive1.getPayload()).isEqualTo(1);
assertThat(receive2.getPayload()).isEqualTo(2);
assertThat(receive2).has(correlationHeadersForPayload2);
}
else {
assertThat(Arrays.asList((Integer) receive0.getPayload(), (Integer) receive1.getPayload(),
(Integer) receive2.getPayload()), containsInAnyOrder(0, 1, 2));
List<Message<?>> receivedMessages = Arrays.asList(receive0, receive1, receive2);
assertThat(receivedMessages).extracting("payload").containsExactlyInAnyOrder(0, 1, 2);
Condition<Message<?>> payloadIs2 = new Condition<Message<?>>() {
@SuppressWarnings("unchecked")
Matcher<Iterable<? extends Message<?>>> containsOur3Messages = containsInAnyOrder(fooMatcher,
hasProperty("payload", equalTo(0)), hasProperty("payload", equalTo(1)));
assertThat(Arrays.asList(receive0, receive1, receive2), containsOur3Messages);
@Override
public boolean matches(Message<?> value) {
return value.getPayload().equals(2);
}
};
assertThat(receivedMessages).filteredOn(payloadIs2).areExactly(1, correlationHeadersForPayload2);
}
input0Binding.unbind();
@@ -282,8 +276,8 @@ abstract public class PartitionCapableBinderTests<B extends AbstractTestBinder<?
Binding<MessageChannel> outputBinding = binder.bindProducer("partJ.0", output, producerProperties);
if (usesExplicitRouting()) {
AbstractEndpoint endpoint = extractEndpoint(outputBinding);
assertThat(getEndpointRouting(endpoint),
containsString(getExpectedRoutingBaseDestination("partJ.0", "test") + "-' + headers['partition']"));
assertThat(getEndpointRouting(endpoint)).
contains(getExpectedRoutingBaseDestination("partJ.0", "test") + "-' + headers['partition']");
}
output.send(new GenericMessage<>(2));
@@ -291,21 +285,20 @@ abstract public class PartitionCapableBinderTests<B extends AbstractTestBinder<?
output.send(new GenericMessage<>(0));
Message<?> receive0 = receive(input0);
assertNotNull(receive0);
assertThat(receive0).isNotNull();
Message<?> receive1 = receive(input1);
assertNotNull(receive1);
assertThat(receive1).isNotNull();
Message<?> receive2 = receive(input2);
assertNotNull(receive2);
assertThat(receive2).isNotNull();
if (usesExplicitRouting()) {
assertEquals(0, receive0.getPayload());
assertEquals(1, receive1.getPayload());
assertEquals(2, receive2.getPayload());
assertThat(receive0.getPayload()).isEqualTo(0);
assertThat(receive1.getPayload()).isEqualTo(1);
assertThat(receive2.getPayload()).isEqualTo(2);
}
else {
assertThat(Arrays.asList((Integer) receive0.getPayload(), (Integer) receive1.getPayload(),
(Integer) receive2.getPayload()), containsInAnyOrder(0, 1, 2));
List<Message<?>> receivedMessages = Arrays.asList(receive0, receive1, receive2);
assertThat(receivedMessages).extracting("payload").containsExactlyInAnyOrder(0, 1, 2);
}
input0Binding.unbind();

146
spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/test/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupportTests.java
View File

@@ -41,9 +41,8 @@ import org.springframework.tuple.TupleBuilder;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Gary Russell
@@ -66,61 +65,57 @@ public class MessageChannelBinderSupportTests {
byte[] payload = "foo".getBytes();
Message<byte[]> message = MessageBuilder.withPayload(payload).build();
MessageValues converted = binder.serializePayloadIfNecessary(message);
assertSame(payload, converted.getPayload());
assertThat(converted.getPayload()).isSameAs(payload);
Message<?> convertedMessage = converted.toMessage();
assertSame(payload, convertedMessage.getPayload());
assertEquals(MimeTypeUtils.APPLICATION_OCTET_STREAM,
contentTypeResolver.resolve(convertedMessage.getHeaders()));
assertThat(convertedMessage.getPayload()).isSameAs(payload);
assertThat(contentTypeResolver.resolve(convertedMessage.getHeaders()))
.isEqualTo(MimeTypeUtils.APPLICATION_OCTET_STREAM);
MessageValues reconstructed = binder.deserializePayloadIfNecessary(convertedMessage);
payload = (byte[]) reconstructed.getPayload();
assertSame(converted.getPayload(), payload);
assertNull(reconstructed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
assertThat(converted.getPayload()).isSameAs(payload);
assertThat(reconstructed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull();
}
@Test
public void testBytesPassThruContentType() {
byte[] payload = "foo".getBytes();
Message<byte[]> message = MessageBuilder.withPayload(payload)
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_OCTET_STREAM_VALUE)
.build();
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_OCTET_STREAM_VALUE).build();
MessageValues messageValues = binder.serializePayloadIfNecessary(message);
Message<?> converted = messageValues.toMessage();
assertSame(payload, converted.getPayload());
assertEquals(MimeTypeUtils.APPLICATION_OCTET_STREAM,
contentTypeResolver.resolve(converted.getHeaders()));
assertThat(converted.getPayload()).isSameAs(payload);
assertThat(contentTypeResolver.resolve(converted.getHeaders()))
.isEqualTo(MimeTypeUtils.APPLICATION_OCTET_STREAM);
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
payload = (byte[]) reconstructed.getPayload();
assertSame(converted.getPayload(), payload);
assertEquals(MimeTypeUtils.APPLICATION_OCTET_STREAM_VALUE,
reconstructed.get(MessageHeaders.CONTENT_TYPE));
assertNull(reconstructed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
assertThat(converted.getPayload()).isSameAs(payload);
assertThat(reconstructed.get(MessageHeaders.CONTENT_TYPE))
.isEqualTo(MimeTypeUtils.APPLICATION_OCTET_STREAM_VALUE);
assertThat(reconstructed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull();
}
@Test
public void testString() throws IOException {
MessageValues convertedValues = binder.serializePayloadIfNecessary(
new GenericMessage<String>("foo"));
MessageValues convertedValues = binder.serializePayloadIfNecessary(new GenericMessage<>("foo"));
Message<?> converted = convertedValues.toMessage();
assertEquals(MimeTypeUtils.TEXT_PLAIN,
contentTypeResolver.resolve(converted.getHeaders()));
assertThat(contentTypeResolver.resolve(converted.getHeaders())).isEqualTo(MimeTypeUtils.TEXT_PLAIN);
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
assertEquals("foo", reconstructed.getPayload());
assertNull(reconstructed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
assertEquals(MimeTypeUtils.TEXT_PLAIN_VALUE, reconstructed.get(MessageHeaders.CONTENT_TYPE));
assertThat(reconstructed.getPayload()).isEqualTo("foo");
assertThat(reconstructed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull();
assertThat(reconstructed.get(MessageHeaders.CONTENT_TYPE)).isEqualTo(MimeTypeUtils.TEXT_PLAIN_VALUE);
}
@Test
public void testStringXML() throws IOException {
Message<?> message = MessageBuilder
.withPayload("<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"no\"?><test></test>")
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_XML)
.build();
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_XML).build();
Message<?> converted = binder.serializePayloadIfNecessary(message).toMessage();
assertEquals(MimeTypeUtils.TEXT_PLAIN, contentTypeResolver.resolve(converted.getHeaders()));
assertThat(contentTypeResolver.resolve(converted.getHeaders())).isEqualTo(MimeTypeUtils.TEXT_PLAIN);
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
assertEquals("<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"no\"?><test></test>", reconstructed.getPayload());
assertEquals(MimeTypeUtils.TEXT_XML.toString(), reconstructed.get(MessageHeaders.CONTENT_TYPE));
assertThat(reconstructed.getPayload())
.isEqualTo("<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"no\"?><test></test>");
assertThat(reconstructed.get(MessageHeaders.CONTENT_TYPE)).isEqualTo(MimeTypeUtils.TEXT_XML.toString());
}
@Test
@@ -133,13 +128,11 @@ public class MessageChannelBinderSupportTests {
Message<?> converted = convertedValues.toMessage();
assertEquals(MimeTypeUtils.TEXT_PLAIN,
contentTypeResolver.resolve(converted.getHeaders()));
assertEquals(MimeTypeUtils.APPLICATION_JSON.toString(),
converted.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
assertThat(contentTypeResolver.resolve(converted.getHeaders())).isEqualTo(MimeTypeUtils.TEXT_PLAIN);
assertThat(converted.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isEqualTo(MimeTypeUtils.APPLICATION_JSON.toString());
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
assertEquals("{\"foo\":\"foo\"}", reconstructed.getPayload());
assertEquals(MimeTypeUtils.APPLICATION_JSON_VALUE, reconstructed.get(MessageHeaders.CONTENT_TYPE));
assertThat(reconstructed.getPayload()).isEqualTo("{\"foo\":\"foo\"}");
assertThat(reconstructed.get(MessageHeaders.CONTENT_TYPE)).isEqualTo(MimeTypeUtils.APPLICATION_JSON_VALUE);
}
@Test
@@ -148,59 +141,24 @@ public class MessageChannelBinderSupportTests {
.copyHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON))
.build();
MessageValues reconstructed = binder.deserializePayloadIfNecessary(inbound);
assertEquals("{\"foo\":\"bar\"}", reconstructed.getPayload());
assertEquals(MimeTypeUtils.APPLICATION_JSON, reconstructed.get(MessageHeaders.CONTENT_TYPE));
assertThat(reconstructed.getPayload()).isEqualTo("{\"foo\":\"bar\"}");
assertThat(reconstructed.get(MessageHeaders.CONTENT_TYPE)).isEqualTo(MimeTypeUtils.APPLICATION_JSON);
}
@Test
public void testPojoSerialization() {
MessageValues convertedValues = binder.serializePayloadIfNecessary(
new GenericMessage<Foo>(new Foo("bar")));
MessageValues convertedValues = binder.serializePayloadIfNecessary(new GenericMessage<>(new Foo("bar")));
Message<?> converted = convertedValues.toMessage();
MimeType mimeType = contentTypeResolver.resolve(converted.getHeaders());
assertEquals("application", mimeType.getType());
assertEquals("x-java-object", mimeType.getSubtype());
assertEquals(Foo.class.getName(), mimeType.getParameter("type"));
assertThat(mimeType.getType()).isEqualTo("application");
assertThat(mimeType.getSubtype()).isEqualTo("x-java-object");
assertThat(mimeType.getParameter("type")).isEqualTo(Foo.class.getName());
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
assertEquals("bar", ((Foo) reconstructed.getPayload()).getBar());
assertNull(reconstructed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
assertEquals("application/x-java-object;type=org.springframework.cloud.stream.binder.MessageChannelBinderSupportTests$Foo",
reconstructed.get(MessageHeaders.CONTENT_TYPE));
}
@Test
public void testPojoWithXJavaObjectMimeTypeNoType() {
MessageValues convertedValues = binder.serializePayloadIfNecessary(
new GenericMessage<Foo>(new Foo("bar")));
Message<?> converted = convertedValues.toMessage();
MimeType mimeType = contentTypeResolver.resolve(converted.getHeaders());
assertEquals("application", mimeType.getType());
assertEquals("x-java-object", mimeType.getSubtype());
assertEquals(Foo.class.getName(), mimeType.getParameter("type"));
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
assertEquals("bar", ((Foo) reconstructed.getPayload()).getBar());
assertNull(reconstructed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
assertEquals("application/x-java-object;type=org.springframework.cloud.stream.binder.MessageChannelBinderSupportTests$Foo",
reconstructed.get(MessageHeaders.CONTENT_TYPE));
}
@Test
public void testPojoWithXJavaObjectMimeTypeExplicitType() {
MessageValues convertedValues = binder.serializePayloadIfNecessary(
new GenericMessage<Foo>(new Foo("bar")));
Message<?> converted = convertedValues.toMessage();
MimeType mimeType = contentTypeResolver.resolve(converted.getHeaders());
assertEquals("application", mimeType.getType());
assertEquals("x-java-object", mimeType.getSubtype());
assertEquals(Foo.class.getName(), mimeType.getParameter("type"));
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
assertEquals("bar", ((Foo) reconstructed.getPayload()).getBar());
assertNull(reconstructed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
assertEquals("application/x-java-object;type=org.springframework.cloud.stream.binder.MessageChannelBinderSupportTests$Foo",
reconstructed.get(MessageHeaders.CONTENT_TYPE));
assertThat(((Foo) reconstructed.getPayload()).getBar()).isEqualTo("bar");
assertThat(reconstructed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull();
assertThat(reconstructed.get(MessageHeaders.CONTENT_TYPE)).isEqualTo(
"application/x-java-object;type=org.springframework.cloud.stream.binder.MessageChannelBinderSupportTests$Foo");
}
@Test
@@ -209,50 +167,50 @@ public class MessageChannelBinderSupportTests {
MessageValues convertedValues = binder.serializePayloadIfNecessary(new GenericMessage<>(payload));
Message<?> converted = convertedValues.toMessage();
MimeType mimeType = contentTypeResolver.resolve(converted.getHeaders());
assertEquals("application", mimeType.getType());
assertEquals("x-java-object", mimeType.getSubtype());
assertEquals(DefaultTuple.class.getName(), mimeType.getParameter("type"));
assertThat(mimeType.getType()).isEqualTo("application");
assertThat(mimeType.getSubtype()).isEqualTo("x-java-object");
assertThat(mimeType.getParameter("type")).isEqualTo(DefaultTuple.class.getName());
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
assertEquals("bar", ((Tuple) reconstructed.getPayload()).getString("foo"));
assertNull(reconstructed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
assertEquals("application/x-java-object;type=org.springframework.tuple.DefaultTuple",
reconstructed.get(MessageHeaders.CONTENT_TYPE));
assertThat(((Tuple) reconstructed.getPayload()).getString("foo")).isEqualTo("bar");
assertThat(reconstructed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull();
assertThat(reconstructed.get(MessageHeaders.CONTENT_TYPE))
.isEqualTo("application/x-java-object;type=org.springframework.tuple.DefaultTuple");
}
@Test
public void mimeTypeIsSimpleObject() throws ClassNotFoundException {
MimeType mt = JavaClassMimeTypeConversion.mimeTypeFromObject(new Object());
String className = JavaClassMimeTypeConversion.classNameFromMimeType(mt);
assertEquals(Object.class, Class.forName(className));
assertThat(Class.forName(className)).isEqualTo(Object.class);
}
@Test
public void mimeTypeIsObjectArray() throws ClassNotFoundException {
MimeType mt = JavaClassMimeTypeConversion.mimeTypeFromObject(new String[0]);
String className = JavaClassMimeTypeConversion.classNameFromMimeType(mt);
assertEquals(String[].class, Class.forName(className));
assertThat(Class.forName(className)).isEqualTo(String[].class);
}
@Test
public void mimeTypeIsMultiDimensionalObjectArray() throws ClassNotFoundException {
MimeType mt = JavaClassMimeTypeConversion.mimeTypeFromObject(new String[0][0][0]);
String className = JavaClassMimeTypeConversion.classNameFromMimeType(mt);
assertEquals(String[][][].class, Class.forName(className));
assertThat(Class.forName(className)).isEqualTo(String[][][].class);
}
@Test
public void mimeTypeIsPrimitiveArray() throws ClassNotFoundException {
MimeType mt = JavaClassMimeTypeConversion.mimeTypeFromObject(new int[0]);
String className = JavaClassMimeTypeConversion.classNameFromMimeType(mt);
assertEquals(int[].class, Class.forName(className));
assertThat(Class.forName(className)).isEqualTo(int[].class);
}
@Test
public void mimeTypeIsMultiDimensionalPrimitiveArray() throws ClassNotFoundException {
MimeType mt = JavaClassMimeTypeConversion.mimeTypeFromObject(new int[0][0][0]);
String className = JavaClassMimeTypeConversion.classNameFromMimeType(mt);
assertEquals(int[][][].class, Class.forName(className));
assertThat(Class.forName(className)).isEqualTo(int[][][].class);
}
public static class Foo {