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

Implement remainder of Spring Boot Checkstyle code rules

Browse Source 
Fixes #532

Implements most of the remaining rules from Spring Boot, except JavaDoc + a few additional ones
This commit is contained in:
Marius Bogoevici
2016-05-12 15:55:03 -04:00
parent bbe5b3ee56
commit 18dc0a707d
97 changed files with 546 additions and 365 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java
View File

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.kafka;
import java.util.Collection;

6
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaConsumerProperties.java
View File

@@ -23,11 +23,11 @@ public class KafkaConsumerProperties {
private boolean autoCommitOffset = true;
private boolean resetOffsets = false;
private boolean resetOffsets;
private KafkaMessageChannelBinder.StartOffset startOffset = null;
private KafkaMessageChannelBinder.StartOffset startOffset;
private boolean enableDlq = false;
private boolean enableDlq;
public boolean isAutoCommitOffset() {
return autoCommitOffset;

27
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -129,7 +129,7 @@ public class KafkaMessageChannelBinder
private boolean autoCreateTopics = true;
private boolean autoAddPartitions = false;
private boolean autoAddPartitions;
private RetryOperations metadataRetryOperations;
@@ -163,7 +163,7 @@ public class KafkaMessageChannelBinder
private int offsetUpdateTimeWindow = 10000;
private int offsetUpdateCount = 0;
private int offsetUpdateCount;
private int offsetUpdateShutdownTimeout = 2000;
@@ -173,7 +173,7 @@ public class KafkaMessageChannelBinder
private ProducerListener producerListener;
private volatile Producer<byte[],byte[]> dlqProducer;
private volatile Producer<byte[], byte[]> dlqProducer;
private KafkaExtendedBindingProperties extendedBindingProperties = new KafkaExtendedBindingProperties();
@@ -783,11 +783,11 @@ public class KafkaMessageChannelBinder
}
}
private class ReceivingHandler extends AbstractReplyProducingMessageHandler {
private final class ReceivingHandler extends AbstractReplyProducingMessageHandler {
private final ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties;
public ReceivingHandler(ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties) {
private ReceivingHandler(ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties) {
this.consumerProperties = consumerProperties;
}
@@ -802,22 +802,21 @@ public class KafkaMessageChannelBinder
}
}
@SuppressWarnings("serial")
private final class KafkaBinderHeaders extends MessageHeaders {
KafkaBinderHeaders(Map<String, Object> headers) {
super(headers, MessageHeaders.ID_VALUE_NONE, -1L);
}
}
@Override
protected boolean shouldCopyRequestHeaders() {
// prevent the message from being copied again in superclass
return false;
}
@SuppressWarnings("serial")
private final class KafkaBinderHeaders extends MessageHeaders {
KafkaBinderHeaders(Map<String, Object> headers) {
super(headers, MessageHeaders.ID_VALUE_NONE, -1L);
}
}
}
private class SendingHandler extends AbstractMessageHandler {
private final class SendingHandler extends AbstractMessageHandler {
private final AtomicInteger roundRobinCount = new AtomicInteger();

4
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaProducerProperties.java
View File

@@ -29,9 +29,9 @@ public class KafkaProducerProperties {
private ProducerMetadata.CompressionType compressionType = ProducerMetadata.CompressionType.none;
private boolean sync = false;
private boolean sync;
private int batchTimeout = 0;
private int batchTimeout;
public int getBufferSize() {
return bufferSize;

6
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/WindowingOffsetManager.java
View File

@@ -68,7 +68,7 @@ public class WindowingOffsetManager implements OffsetManager, InitializingBean,
private long timespan = 10 * 1000;
private int count = 0;
private int count;
private Subject<PartitionAndOffset, PartitionAndOffset> offsets;
@@ -187,13 +187,13 @@ public class WindowingOffsetManager implements OffsetManager, InitializingBean,
delegate.flush();
}
class PartitionAndOffset {
private final class PartitionAndOffset {
private final Partition partition;
private final Long offset;
public PartitionAndOffset(Partition partition, Long offset) {
private PartitionAndOffset(Partition partition, Long offset) {
this.partition = partition;
this.offset = offset;
}

4
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java
View File

@@ -39,7 +39,7 @@ public class KafkaBinderConfigurationProperties {
private int offsetUpdateTimeWindow = 10000;
private int offsetUpdateCount = 0;
private int offsetUpdateCount;
private int offsetUpdateShutdownTimeout = 2000;
@@ -47,7 +47,7 @@ public class KafkaBinderConfigurationProperties {
private boolean autoCreateTopics = true;
private boolean autoAddPartitions = false;
private boolean autoAddPartitions;
/**
* ZK session timeout in milliseconds.

2
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/package-info.java
View File

@@ -5,7 +5,7 @@
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,

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

@@ -79,7 +79,8 @@ import static org.junit.Assert.fail;
* @author Mark Fisher
* @author Ilayaperumal Gopinathan
*/
public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinder, ExtendedConsumerProperties<KafkaConsumerProperties>, ExtendedProducerProperties<KafkaProducerProperties>> {
public class KafkaBinderTests extends
PartitionCapableBinderTests<KafkaTestBinder, ExtendedConsumerProperties<KafkaConsumerProperties>, ExtendedProducerProperties<KafkaProducerProperties>> {
private final String CLASS_UNDER_TEST_NAME = KafkaMessageChannelBinder.class.getSimpleName();
@@ -614,24 +615,6 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
binding.unbind();
}
private static class FailingInvocationCountingMessageHandler implements MessageHandler {
private int invocationCount = 0;
public FailingInvocationCountingMessageHandler() {
}
@Override
public void handleMessage(Message<?> message) throws MessagingException {
invocationCount++;
throw new RuntimeException();
}
public int getInvocationCount() {
return invocationCount;
}
}
@Test
public void testPartitionCountNotReduced() throws Exception {
String testTopicName = "existing" + System.currentTimeMillis();
@@ -689,4 +672,22 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(testTopicName, kafkaTestSupport.getZkClient());
assertThat(topicMetadata.partitionsMetadata().size(), equalTo(6));
}
private static final class FailingInvocationCountingMessageHandler implements MessageHandler {
private int invocationCount;
private FailingInvocationCountingMessageHandler() {
}
@Override
public void handleMessage(Message<?> message) throws MessagingException {
invocationCount++;
throw new RuntimeException();
}
public int getInvocationCount() {
return invocationCount;
}
}
}

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

@@ -20,16 +20,14 @@ import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy;
import org.springframework.cloud.stream.binder.PartitionSelectorStrategy;
import org.springframework.messaging.Message;
/**
*
* @author Marius Bogoevici
*/
public class RawKafkaPartitionTestSupport implements PartitionKeyExtractorStrategy, PartitionSelectorStrategy {
@Override
public int selectPartition(Object key, int divisor) {
return ((byte[])key)[0] % divisor;
return ((byte[]) key)[0] % divisor;
}
@Override

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

@@ -83,9 +83,9 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
input2.setBeanName("test.input2J");
Binding<MessageChannel> input2Binding = binder.bindConsumer("partJ.0", "test", input2, consumerProperties);
output.send(new GenericMessage<>(new byte[]{(byte)0}));
output.send(new GenericMessage<>(new byte[]{(byte)1}));
output.send(new GenericMessage<>(new byte[]{(byte)2}));
output.send(new GenericMessage<>(new byte[] { (byte) 0 }));
output.send(new GenericMessage<>(new byte[] { (byte) 1 }));
output.send(new GenericMessage<>(new byte[] { (byte) 2 }));
Message<?> receive0 = receive(input0);
assertNotNull(receive0);
@@ -94,11 +94,8 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
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(Arrays.asList(((byte[]) receive0.getPayload())[0], ((byte[]) receive1.getPayload())[0],
((byte[]) receive2.getPayload())[0]), containsInAnyOrder((byte) 0, (byte) 1, (byte) 2));
input0Binding.unbind();
input1Binding.unbind();
@@ -166,7 +163,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
((byte[]) receive0.getPayload())[0],
((byte[]) receive1.getPayload())[0],
((byte[]) receive2.getPayload())[0]),
containsInAnyOrder((byte)0, (byte)1, (byte)2));
containsInAnyOrder((byte) 0, (byte) 1, (byte) 2));
input0Binding.unbind();
input1Binding.unbind();
@@ -192,7 +189,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
moduleOutputChannel.send(message);
Message<?> inbound = receive(moduleInputChannel);
assertNotNull(inbound);
assertEquals("foo", new String((byte[])inbound.getPayload()));
assertEquals("foo", new String((byte[]) inbound.getPayload()));
producerBinding.unbind();
consumerBinding.unbind();
}
@@ -233,7 +230,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
moduleOutputChannel.send(message);
Message<?> inbound = receive(module1InputChannel);
assertNotNull(inbound);
assertEquals("foo", new String((byte[])inbound.getPayload()));
assertEquals("foo", new String((byte[]) inbound.getPayload()));
Message<?> tapped1 = receive(module2InputChannel);
Message<?> tapped2 = receive(module3InputChannel);
@@ -277,7 +274,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
private void assertMessageReceive(QueueChannel moduleInputChannel, String payload) {
Message<?> inbound = receive(moduleInputChannel);
assertNotNull(inbound);
assertEquals(payload, new String((byte[])inbound.getPayload()));
assertEquals(payload, new String((byte[]) inbound.getPayload()));
assertNull(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE));
}

2
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitBindingCleaner.java
View File

@@ -47,7 +47,7 @@ public class RabbitBindingCleaner implements BindingCleaner {
@Override
public Map<String, List<String>> clean(String entity, boolean isJob) {
return clean("http://localhost:15672", "guest", "guest", "/", BINDER_PREFIX , entity, isJob);
return clean("http://localhost:15672", "guest", "guest", "/", BINDER_PREFIX, entity, isJob);
}
public Map<String, List<String>> clean(String adminUri, String user, String pw, String vhost,

6
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitConsumerProperties.java
View File

@@ -28,7 +28,7 @@ public class RabbitConsumerProperties {
private String prefix = "";
private boolean transacted = false;
private boolean transacted;
private AcknowledgeMode acknowledgeMode = AcknowledgeMode.AUTO;
@@ -40,11 +40,11 @@ public class RabbitConsumerProperties {
private int txSize = 1;
private boolean autoBindDlq = false;
private boolean autoBindDlq;
private boolean durableSubscription = true;
private boolean republishToDlq = false;
private boolean republishToDlq;
private boolean requeueRejected = true;

2
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitManagementUtils.java
View File

@@ -42,7 +42,7 @@ import org.springframework.web.client.RestTemplate;
* @author Gary Russell
* @since 1.2
*/
public class RabbitManagementUtils {
public abstract class RabbitManagementUtils {
public static RestTemplate buildRestTemplate(String adminUri, String user, String password) {
BasicCredentialsProvider credsProvider = new BasicCredentialsProvider();

14
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
View File

@@ -524,10 +524,10 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, E
}
public void cleanAutoDeclareContext(String prefix, String name) {
synchronized(this.autoDeclareContext) {
removeSingleton(applyPrefix(prefix,name) + ".binding");
removeSingleton(applyPrefix(prefix,name));
String dlq = applyPrefix(prefix,name) + ".dlq";
synchronized (this.autoDeclareContext) {
removeSingleton(applyPrefix(prefix, name) + ".binding");
removeSingleton(applyPrefix(prefix, name));
String dlq = applyPrefix(prefix, name) + ".dlq";
removeSingleton(dlq + ".binding");
removeSingleton(dlq);
}
@@ -564,7 +564,7 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, E
}
}
private class SendingHandler extends AbstractMessageHandler implements Lifecycle {
private final class SendingHandler extends AbstractMessageHandler implements Lifecycle {
private final MessageHandler delegate;
@@ -626,9 +626,9 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, E
}
private class ReceivingHandler extends AbstractReplyProducingMessageHandler {
private final class ReceivingHandler extends AbstractReplyProducingMessageHandler {
public ReceivingHandler() {
private ReceivingHandler() {
super();
this.setBeanFactory(RabbitMessageChannelBinder.this.getBeanFactory());
}

6
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java
View File

@@ -30,11 +30,11 @@ public class RabbitProducerProperties {
private String[] requestHeaderPatterns = new String[] {"STANDARD_REQUEST_HEADERS", "*"};
private boolean autoBindDlq = false;
private boolean autoBindDlq;
private boolean compress = false;
private boolean compress;
private boolean batchingEnabled = false;
private boolean batchingEnabled;
private int batchSize = 100;

10
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitServiceAutoConfiguration.java
View File

@@ -48,6 +48,11 @@ import org.springframework.context.annotation.Profile;
@AutoConfigureBefore({CloudAutoConfiguration.class, RabbitAutoConfiguration.class})
public class RabbitServiceAutoConfiguration {
@Bean
public HealthIndicator binderHealthIndicator(RabbitTemplate rabbitTemplate) {
return new RabbitHealthIndicator(rabbitTemplate);
}
@Configuration
@Profile("cloud")
@ConditionalOnClass(Cloud.class)
@@ -70,9 +75,4 @@ public class RabbitServiceAutoConfiguration {
@Import(RabbitAutoConfiguration.class)
protected static class NoCloudConfig {
}
@Bean
public HealthIndicator binderHealthIndicator(RabbitTemplate rabbitTemplate) {
return new RabbitHealthIndicator(rabbitTemplate);
}
}

18
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/package-info.java
View File

@@ -1,5 +1,21 @@
/*
* Copyright 2015-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
/**
* This package contains an implementation of the {@link org.springframework.cloud.stream.binder.Binder} for RabbitMQ.
*/
package org.springframework.cloud.stream.binder.rabbit;
package org.springframework.cloud.stream.binder.rabbit;

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

@@ -584,9 +584,6 @@ public class RabbitBinderTests extends PartitionCapableBinderTests<RabbitTestBin
output.setBeanName("batchingProducer");
Binding<MessageChannel> producerBinding = binder.bindProducer("batching.0", output, properties);
while (template.receive(properties.getExtension().getPrefix() + "batching.0.default") != null) {
}
Log logger = spy(TestUtils.getPropertyValue(binder, "binder.compressingPostProcessor.logger", Log.class));
new DirectFieldAccessor(TestUtils.getPropertyValue(binder, "binder.compressingPostProcessor"))
.setPropertyValue("logger", logger);

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

@@ -66,10 +66,10 @@ public class RabbitBinderModuleTests {
@ClassRule
public static RabbitTestSupport rabbitTestSupport = new RabbitTestSupport();
private ConfigurableApplicationContext context = null;
private ConfigurableApplicationContext context;
public static final ConnectionFactory MOCK_CONNECTION_FACTORY =
Mockito.mock(ConnectionFactory.class, Mockito.RETURNS_MOCKS);
public static final ConnectionFactory MOCK_CONNECTION_FACTORY = Mockito.mock(ConnectionFactory.class,
Mockito.RETURNS_MOCKS);
@After
public void tearDown() {
@@ -91,26 +91,26 @@ public class RabbitBinderModuleTests {
Binder binder = binderFactory.getBinder(null);
assertThat(binder, instanceOf(RabbitMessageChannelBinder.class));
DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
ConnectionFactory binderConnectionFactory =
(ConnectionFactory) binderFieldAccessor.getPropertyValue("connectionFactory");
ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor
.getPropertyValue("connectionFactory");
assertThat(binderConnectionFactory, instanceOf(CachingConnectionFactory.class));
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
assertThat(binderConnectionFactory, is(connectionFactory));
CompositeHealthIndicator bindersHealthIndicator =
context.getBean("bindersHealthIndicator", CompositeHealthIndicator.class);
CompositeHealthIndicator bindersHealthIndicator = context.getBean("bindersHealthIndicator",
CompositeHealthIndicator.class);
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
assertNotNull(bindersHealthIndicator);
@SuppressWarnings("unchecked")
Map<String,HealthIndicator> healthIndicators =
(Map<String, HealthIndicator>) directFieldAccessor.getPropertyValue("indicators");
Map<String, HealthIndicator> healthIndicators = (Map<String, HealthIndicator>) directFieldAccessor
.getPropertyValue("indicators");
assertThat(healthIndicators, hasKey("rabbit"));
assertThat(healthIndicators.get("rabbit").health().getStatus(), equalTo(Status.UP));
}
@Test
@SuppressWarnings("unchecked")
public void testParentConnectionFactoryInheritedByDefaultAndRabbitSettingsPropagated() {
context = SpringApplication.run(SimpleProcessor.class,
"--server.port=0",
context = SpringApplication.run(SimpleProcessor.class, "--server.port=0",
"--spring.cloud.stream.rabbit.bindings.input.consumer.transacted=true",
"--spring.cloud.stream.rabbit.bindings.output.producer.transacted=true");
BinderFactory<?> binderFactory = context.getBean(BinderFactory.class);
@@ -118,53 +118,54 @@ public class RabbitBinderModuleTests {
assertThat(binder, instanceOf(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.getPropertyValue("consumerBindings");
Map<String, List<Binding<MessageChannel>>> consumerBindings = (Map<String, List<Binding<MessageChannel>>>) channelBindingServiceAccessor
.getPropertyValue("consumerBindings");
Binding<MessageChannel> inputBinding = consumerBindings.get("input").get(0);
SimpleMessageListenerContainer container = TestUtils.getPropertyValue(inputBinding,
"endpoint.messageListenerContainer",
SimpleMessageListenerContainer.class);
"endpoint.messageListenerContainer", SimpleMessageListenerContainer.class);
assertTrue(TestUtils.getPropertyValue(container, "transactional", Boolean.class));
Map<String, Binding<MessageChannel>> producerBindings =
(Map<String, Binding<MessageChannel>>) TestUtils.getPropertyValue(channelBindingService, "producerBindings");
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));
assertTrue(TestUtils.getPropertyValue(outputBinding, "endpoint.handler.delegate.amqpTemplate.transactional",
Boolean.class));
DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
ConnectionFactory binderConnectionFactory =
(ConnectionFactory) binderFieldAccessor.getPropertyValue("connectionFactory");
ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor
.getPropertyValue("connectionFactory");
assertThat(binderConnectionFactory, instanceOf(CachingConnectionFactory.class));
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
assertThat(binderConnectionFactory, is(connectionFactory));
CompositeHealthIndicator bindersHealthIndicator =
context.getBean("bindersHealthIndicator", CompositeHealthIndicator.class);
CompositeHealthIndicator bindersHealthIndicator = context.getBean("bindersHealthIndicator",
CompositeHealthIndicator.class);
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
assertNotNull(bindersHealthIndicator);
@SuppressWarnings("unchecked")
Map<String, HealthIndicator> healthIndicators =
(Map<String, HealthIndicator>) directFieldAccessor.getPropertyValue("indicators");
Map<String, HealthIndicator> healthIndicators = (Map<String, HealthIndicator>) directFieldAccessor
.getPropertyValue("indicators");
assertThat(healthIndicators, hasKey("rabbit"));
assertThat(healthIndicators.get("rabbit").health().getStatus(), equalTo(Status.UP));
}
@Test
public void testParentConnectionFactoryInheritedIfOverridden() {
context = new SpringApplication(SimpleProcessor.class, ConnectionFactoryConfiguration.class).run("--server.port=0");
context = new SpringApplication(SimpleProcessor.class, ConnectionFactoryConfiguration.class)
.run("--server.port=0");
BinderFactory<?> binderFactory = context.getBean(BinderFactory.class);
Binder binder = binderFactory.getBinder(null);
assertThat(binder, instanceOf(RabbitMessageChannelBinder.class));
DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
ConnectionFactory binderConnectionFactory =
(ConnectionFactory) binderFieldAccessor.getPropertyValue("connectionFactory");
ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor
.getPropertyValue("connectionFactory");
assertThat(binderConnectionFactory, is(MOCK_CONNECTION_FACTORY));
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
assertThat(binderConnectionFactory, is(connectionFactory));
CompositeHealthIndicator bindersHealthIndicator =
context.getBean("bindersHealthIndicator", CompositeHealthIndicator.class);
CompositeHealthIndicator bindersHealthIndicator = context.getBean("bindersHealthIndicator",
CompositeHealthIndicator.class);
assertNotNull(bindersHealthIndicator);
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
@SuppressWarnings("unchecked")
Map<String,HealthIndicator> healthIndicators =
(Map<String, HealthIndicator>) directFieldAccessor.getPropertyValue("indicators");
Map<String, HealthIndicator> healthIndicators = (Map<String, HealthIndicator>) directFieldAccessor
.getPropertyValue("indicators");
assertThat(healthIndicators, hasKey("rabbit"));
// mock connection factory behaves as if down
assertThat(healthIndicators.get("rabbit").health().getStatus(), equalTo(Status.DOWN));
@@ -183,17 +184,17 @@ public class RabbitBinderModuleTests {
Binder binder = binderFactory.getBinder(null);
assertThat(binder, instanceOf(RabbitMessageChannelBinder.class));
DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
ConnectionFactory binderConnectionFactory =
(ConnectionFactory) binderFieldAccessor.getPropertyValue("connectionFactory");
ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor
.getPropertyValue("connectionFactory");
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
assertThat(binderConnectionFactory, not(is(connectionFactory)));
CompositeHealthIndicator bindersHealthIndicator =
context.getBean("bindersHealthIndicator", CompositeHealthIndicator.class);
CompositeHealthIndicator bindersHealthIndicator = context.getBean("bindersHealthIndicator",
CompositeHealthIndicator.class);
assertNotNull(bindersHealthIndicator);
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
@SuppressWarnings("unchecked")
Map<String,HealthIndicator> healthIndicators =
(Map<String, HealthIndicator>) directFieldAccessor.getPropertyValue("indicators");
Map<String, HealthIndicator> healthIndicators = (Map<String, HealthIndicator>) directFieldAccessor
.getPropertyValue("indicators");
assertThat(healthIndicators, hasKey("custom"));
assertThat(healthIndicators.get("custom").health().getStatus(), equalTo(Status.UP));
}

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

@@ -73,7 +73,7 @@ public abstract class AbstractBinderTests<B extends AbstractTestBinder<? extends
* Allows accomodating tests which are slower than normal (e.g. retry).
*/
protected Message<?> receive(PollableChannel channel, int additionalMultiplier) {
return channel.receive((int)(1000 * timeoutMultiplier * additionalMultiplier));
return channel.receive((int) (1000 * timeoutMultiplier * additionalMultiplier));
}
@Test
@@ -143,7 +143,7 @@ public abstract class AbstractBinderTests<B extends AbstractTestBinder<? extends
moduleOutputChannel2.send(message2);
Message<?> messages[] = new Message[2];
Message<?>[] messages = new Message[2];
messages[0] = receive(moduleInputChannel);
messages[1] = receive(moduleInputChannel);

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

@@ -53,7 +53,7 @@ public abstract class AbstractTestBinder<C extends AbstractBinder<MessageChannel
@Override
public Binding<MessageChannel> bindProducer(String name, MessageChannel moduleOutputChannel, PP properties) {
queues.add(name);
return binder.bindProducer(name, moduleOutputChannel, properties);
return binder.bindProducer(name, moduleOutputChannel, properties);
}
public C getCoreBinder() {

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

@@ -30,7 +30,7 @@ import static org.mockito.Mockito.when;
*
* @author Gary Russell
*/
public class BinderTestUtils {
public abstract class BinderTestUtils {
private static final MessageBuilderFactory mbf = new DefaultMessageBuilderFactory();

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

@@ -23,12 +23,13 @@ import org.springframework.messaging.MessageChannel;
*
* @author Gary Russell
*/
public abstract class BrokerBinderTests<B extends AbstractTestBinder<? extends AbstractBinder<MessageChannel, CP, PP>, CP, PP>, CP extends ConsumerProperties, PP extends ProducerProperties> extends AbstractBinderTests<B,CP,PP> {
public abstract class BrokerBinderTests<B extends AbstractTestBinder<? extends AbstractBinder<MessageChannel, CP, PP>, CP, PP>, CP extends ConsumerProperties, PP extends ProducerProperties>
extends AbstractBinderTests<B, CP, PP> {
/**
* Create a new spy on the given 'queue'. This allows de-correlating the creation of
* the 'connection' from its actual usage, which may be needed by some implementations to
* see messages sent after connection creation.
* the 'connection' from its actual usage, which may be needed by some implementations
* to see messages sent after connection creation.
*/
public abstract Spy spyOn(final String name);

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

@@ -51,7 +51,8 @@ import static org.junit.Assert.assertThat;
* @author Mark Fisher
* @author Marius Bogoevici
*/
abstract public class PartitionCapableBinderTests<B extends AbstractTestBinder<? extends AbstractBinder<MessageChannel, CP, PP>, CP, PP>, CP extends ConsumerProperties, PP extends ProducerProperties> extends BrokerBinderTests<B,CP,PP> {
abstract public class PartitionCapableBinderTests<B extends AbstractTestBinder<? extends AbstractBinder<MessageChannel, CP, PP>, CP, PP>, CP extends ConsumerProperties, PP extends ProducerProperties>
extends BrokerBinderTests<B, CP, PP> {
protected static final SpelExpressionParser spelExpressionParser = new SpelExpressionParser();
@@ -60,13 +61,16 @@ abstract public class PartitionCapableBinderTests<B extends AbstractTestBinder<?
public void testAnonymousGroup() throws Exception {
B binder = getBinder();
DirectChannel output = new DirectChannel();
Binding<MessageChannel> producerBinding = binder.bindProducer("defaultGroup.0", output, createProducerProperties());
Binding<MessageChannel> producerBinding = binder.bindProducer("defaultGroup.0", output,
createProducerProperties());
QueueChannel input1 = new QueueChannel();
Binding<MessageChannel> binding1 = binder.bindConsumer("defaultGroup.0", null, input1, createConsumerProperties());
Binding<MessageChannel> binding1 = binder.bindConsumer("defaultGroup.0", null, input1,
createConsumerProperties());
QueueChannel input2 = new QueueChannel();
Binding<MessageChannel> binding2 = binder.bindConsumer("defaultGroup.0", null, input2, createConsumerProperties());
Binding<MessageChannel> binding2 = binder.bindConsumer("defaultGroup.0", null, input2,
createConsumerProperties());
String testPayload1 = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload1.getBytes()));
@@ -120,7 +124,8 @@ abstract public class PartitionCapableBinderTests<B extends AbstractTestBinder<?
output.send(new GenericMessage<>(testPayload.getBytes()));
QueueChannel inbound1 = new QueueChannel();
Binding<MessageChannel> consumerBinding = binder.bindConsumer(testDestination, "test1", inbound1, createConsumerProperties());
Binding<MessageChannel> consumerBinding = binder.bindConsumer(testDestination, "test1", inbound1,
createConsumerProperties());
Message<?> receivedMessage1 = receive(inbound1);
assertThat(receivedMessage1, not(nullValue()));
@@ -138,16 +143,18 @@ abstract public class PartitionCapableBinderTests<B extends AbstractTestBinder<?
String testDestination = "testDestination" + UUID.randomUUID().toString().replace("-", "");
PP producerProperties = createProducerProperties();
producerProperties.setRequiredGroups("test1","test2");
producerProperties.setRequiredGroups("test1", "test2");
Binding<MessageChannel> producerBinding = binder.bindProducer(testDestination, output, producerProperties);
String testPayload = "foo-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload.getBytes()));
QueueChannel inbound1 = new QueueChannel();
Binding<MessageChannel> consumerBinding1 = binder.bindConsumer(testDestination, "test1", inbound1, createConsumerProperties());
Binding<MessageChannel> consumerBinding1 = binder.bindConsumer(testDestination, "test1", inbound1,
createConsumerProperties());
QueueChannel inbound2 = new QueueChannel();
Binding<MessageChannel> consumerBinding2 = binder.bindConsumer(testDestination, "test2", inbound2, createConsumerProperties());
Binding<MessageChannel> consumerBinding2 = binder.bindConsumer(testDestination, "test2", inbound2,
createConsumerProperties());
Message<?> receivedMessage1 = receive(inbound1);
assertThat(receivedMessage1, not(nullValue()));
@@ -192,8 +199,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),
containsString(getExpectedRoutingBaseDestination("part.0", "test") + "-' + headers['partition']"));
}
catch (UnsupportedOperationException ignored) {
}
@@ -201,8 +208,7 @@ abstract public class PartitionCapableBinderTests<B extends AbstractTestBinder<?
Message<Integer> message2 = MessageBuilder.withPayload(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<>(1));
output.send(new GenericMessage<>(0));
@@ -219,9 +225,8 @@ abstract public class PartitionCapableBinderTests<B extends AbstractTestBinder<?
@Override
public boolean matches(Object item) {
IntegrationMessageHeaderAccessor accessor = new IntegrationMessageHeaderAccessor((Message<?>) item);
boolean result = "foo".equals(accessor.getCorrelationId()) &&
42 == accessor.getSequenceNumber() &&
43 == accessor.getSequenceSize();
boolean result = "foo".equals(accessor.getCorrelationId()) && 42 == accessor.getSequenceNumber()
&& 43 == accessor.getSequenceSize();
return result;
}
};
@@ -232,21 +237,13 @@ abstract public class PartitionCapableBinderTests<B extends AbstractTestBinder<?
assertThat(receive2, fooMatcher);
}
else {
assertThat(Arrays.asList(
(Integer) receive0.getPayload(),
(Integer) receive1.getPayload(),
(Integer) receive2.getPayload()),
containsInAnyOrder(0, 1, 2));
assertThat(Arrays.asList((Integer) receive0.getPayload(), (Integer) receive1.getPayload(),
(Integer) receive2.getPayload()), containsInAnyOrder(0, 1, 2));
@SuppressWarnings("unchecked")
Matcher<Iterable<? extends Message<?>>> containsOur3Messages = containsInAnyOrder(
fooMatcher,
hasProperty("payload", equalTo(0)),
hasProperty("payload", equalTo(1))
);
assertThat(
Arrays.asList(receive0, receive1, receive2),
containsOur3Messages);
Matcher<Iterable<? extends Message<?>>> containsOur3Messages = containsInAnyOrder(fooMatcher,
hasProperty("payload", equalTo(0)), hasProperty("payload", equalTo(1)));
assertThat(Arrays.asList(receive0, receive1, receive2), containsOur3Messages);
}
input0Binding.unbind();
@@ -285,8 +282,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),
containsString(getExpectedRoutingBaseDestination("partJ.0", "test") + "-' + headers['partition']"));
}
output.send(new GenericMessage<>(2));
@@ -307,11 +304,8 @@ abstract public class PartitionCapableBinderTests<B extends AbstractTestBinder<?
}
else {
assertThat(Arrays.asList(
(Integer) receive0.getPayload(),
(Integer) receive1.getPayload(),
(Integer) receive2.getPayload()),
containsInAnyOrder(0, 1, 2));
assertThat(Arrays.asList((Integer) receive0.getPayload(), (Integer) receive1.getPayload(),
(Integer) receive2.getPayload()), containsInAnyOrder(0, 1, 2));
}
input0Binding.unbind();
@@ -321,10 +315,12 @@ abstract public class PartitionCapableBinderTests<B extends AbstractTestBinder<?
}
/**
* Implementations should return whether the binder under test uses "explicit" routing (e.g. Rabbit)
* whereby Spring Cloud Stream is responsible for assigning a partition and knows which exact consumer will receive the
* message (i.e. honor "partitionIndex") or "implicit" routing (e.g. Kafka) whereby the only guarantee
* is that messages will be spread, but we don't control exactly which consumer gets which message.
* Implementations should return whether the binder under test uses "explicit" routing
* (e.g. Rabbit) whereby Spring Cloud Stream is responsible for assigning a partition
* and knows which exact consumer will receive the message (i.e. honor
* "partitionIndex") or "implicit" routing (e.g. Kafka) whereby the only guarantee is
* that messages will be spread, but we don't control exactly which consumer gets
* which message.
*/
protected abstract boolean usesExplicitRouting();
@@ -336,8 +332,8 @@ abstract public class PartitionCapableBinderTests<B extends AbstractTestBinder<?
}
/**
* For implementations that rely on explicit routing, return the expected base destination
* (the part that precedes '-partition' within the expression).
* For implementations that rely on explicit routing, return the expected base
* destination (the part that precedes '-partition' within the expression).
*/
protected String getExpectedRoutingBaseDestination(String name, String group) {
throw new UnsupportedOperationException();

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

@@ -20,7 +20,6 @@ import org.springframework.messaging.Message;
/**
*
* @author Gary Russell
*/
public class PartitionTestSupport implements PartitionKeyExtractorStrategy, PartitionSelectorStrategy {

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

@@ -25,5 +25,5 @@ package org.springframework.cloud.stream.binder;
*/
public interface Spy {
public Object receive(boolean expectNull) throws Exception;
Object receive(boolean expectNull) throws Exception;
}

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

@@ -22,7 +22,7 @@ import org.springframework.util.Assert;
/**
* Copy of class in org.springframework.amqp.utils.test to avoid dependency on spring-amqp
*/
public class TestUtils {
public abstract class TestUtils {
/**
* Uses nested {@link DirectFieldAccessor}s to obtain a property using dotted notation