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

Cleanup after Binder renaming

Browse Source 
one more...

one more comment change
This commit is contained in:
David Turanski
2015-07-21 14:09:50 -04:00
committed by Marius Bogoevici
parent 46100b2f8f
commit 93f278e839
12 changed files with 33 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -24,9 +24,9 @@ import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.binder.BindingCleaner;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;
import org.springframework.cloud.stream.binder.BinderCleaner;
import org.springframework.cloud.stream.binder.BinderUtils;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport;
import org.springframework.cloud.stream.binder.RabbitAdminException;
@@ -34,14 +34,14 @@ import org.springframework.cloud.stream.binder.RabbitManagementUtils;
/**
* Implementation of {@link org.springframework.cloud.stream.binder.BinderCleaner} for the {@code RabbitBinder}.
* Implementation of {@link org.springframework.cloud.stream.binder.BindingCleaner} for the {@code RabbitBinder}.
* @author Gary Russell
* @author David Turanski
* @since 1.2
*/
public class RabbitBinderCleaner implements BinderCleaner {
public class RabbitBindingCleaner implements BindingCleaner {
private final static Logger logger = LoggerFactory.getLogger(RabbitBinderCleaner.class);
private final static Logger logger = LoggerFactory.getLogger(RabbitBindingCleaner.class);
public static final String BINDER_PREFIX = "binder.rabbit.";

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

@@ -117,7 +117,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
private static final int DEFAULT_PREFETCH_COUNT = 1;
private static final String DEFAULT_RABBIT_PREFIX = "binder.rabbit.";
static final String DEFAULT_RABBIT_PREFIX = "binder.";
private static final int DEFAULT_TX_SIZE = 1;

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

@@ -64,7 +64,7 @@ public class RabbitBinderCleanerTests {
@Test
public void testCleanStream() {
final RabbitBinderCleaner cleaner = new RabbitBinderCleaner();
final RabbitBindingCleaner cleaner = new RabbitBindingCleaner();
final RestTemplate template = RabbitManagementUtils.buildRestTemplate("http://localhost:15672", "guest",
"guest");
final String stream1 = UUID.randomUUID().toString();

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

@@ -80,8 +80,6 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
private final String CLASS_UNDER_TEST_NAME = RabbitMessageChannelBinder.class.getSimpleName();
public static final String BINDER_PREFIX = "binder.rabbit.";
public static final String TEST_PREFIX = "bindertest.";
@Rule
@@ -137,7 +135,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint, "messageListenerContainer",
SimpleMessageListenerContainer.class);
assertEquals(AcknowledgeMode.AUTO, container.getAcknowledgeMode());
assertEquals(BINDER_PREFIX + "props.0", container.getQueueNames()[0]);
assertEquals(RabbitMessageChannelBinder.DEFAULT_RABBIT_PREFIX + "props.0", container.getQueueNames()[0]);
assertTrue(TestUtils.getPropertyValue(container, "transactional", Boolean.class));
assertEquals(1, TestUtils.getPropertyValue(container, "concurrentConsumers"));
assertNull(TestUtils.getPropertyValue(container, "maxConcurrentConsumers"));
@@ -208,7 +206,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
List<Binding> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class);
assertEquals(1, bindings.size());
AbstractEndpoint endpoint = bindings.get(0).getEndpoint();
assertEquals(BINDER_PREFIX + "props.0", TestUtils.getPropertyValue(endpoint, "handler.delegate.routingKey"));
assertEquals(RabbitMessageChannelBinder.DEFAULT_RABBIT_PREFIX + "props.0", TestUtils.getPropertyValue(endpoint, "handler.delegate.routingKey"));
MessageDeliveryMode mode = TestUtils.getPropertyValue(endpoint, "handler.delegate.defaultDeliveryMode",
MessageDeliveryMode.class);
assertEquals(MessageDeliveryMode.PERSISTENT, mode);
@@ -589,7 +587,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
output.setBeanName("batchingProducer");
binder.bindProducer("batching.0", output, properties);
while (template.receive(BINDER_PREFIX + "batching.0") != null) {
while (template.receive(RabbitMessageChannelBinder.DEFAULT_RABBIT_PREFIX + "batching.0") != null) {
}
Log logger = spy(TestUtils.getPropertyValue(binder, "binder.compressingPostProcessor.logger", Log.class));
@@ -714,12 +712,12 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
public Object receive(boolean expectNull) throws Exception {
if (expectNull) {
Thread.sleep(50);
return template.receiveAndConvert(BINDER_PREFIX + queue);
return template.receiveAndConvert(RabbitMessageChannelBinder.DEFAULT_RABBIT_PREFIX + queue);
}
Object bar = null;
int n = 0;
while (n++ < 100 && bar == null) {
bar = template.receiveAndConvert(BINDER_PREFIX + queue);
bar = template.receiveAndConvert(RabbitMessageChannelBinder.DEFAULT_RABBIT_PREFIX + queue);
Thread.sleep(100);
}
assertTrue("Message did not arrive in RabbitMQ", n < 100);

17
spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/Binder.java
View File

@@ -15,13 +15,10 @@ package org.springframework.cloud.stream.binder;
import java.util.Properties;
import org.springframework.messaging.MessageChannel;
/**
* A strategy interface used to bind a module interface to a logical name. The name is intended to identify a
* logical consumer or producer of messages. This may be a queue, a channel adapter, another message channel, a Spring
* bean, etc.
*
* @author Mark Fisher
* @author David Turanski
* @author Gary Russell
@@ -33,7 +30,6 @@ public interface Binder<T> {
/**
* Bind a message consumer on a p2p channel
*
* @param name the logical identity of the message source
* @param inboundBindTarget the module interface to be bound as a point to point consumer
* @param properties arbitrary String key/value pairs that will be used in the binding
@@ -43,7 +39,6 @@ public interface Binder<T> {
/**
* Bind a message consumer on a pub/sub channel
*
* @param name the logical identity of the message source
* @param inboundBindTarget the module interface to be bound as a pub/sub consumer
* @param properties arbitrary String key/value pairs that will be used in the binding
@@ -52,7 +47,6 @@ public interface Binder<T> {
/**
* Bind a message producer on a p2p channel.
*
* @param name the logical identity of the message target
* @param outboundBindTarget the module interface bound as a producer
* @param properties arbitrary String key/value pairs that will be used in the binding
@@ -62,7 +56,6 @@ public interface Binder<T> {
/**
* Bind a message producer on a pub/sub channel.
*
* @param name the logical identity of the message target
* @param outboundBindTarget the module interface bound as a producer
* @param properties arbitrary String key/value pairs that will be used in the binding
@@ -71,21 +64,18 @@ public interface Binder<T> {
/**
* Unbind inbound module components and stop any active components that use the channel.
*
* @param name the channel name
*/
void unbindConsumers(String name);
/**
* Unbind an outbound module components and stop any active components that use the channel.
*
* Unbind outbound module components and stop any active components that use the channel.
* @param name the channel name
*/
void unbindProducers(String name);
/**
* Unbind a specific p2p or pub/sub message consumer
*
* @param name The logical identify of a message source
* @param inboundBindTarget The module interface bound as a consumer
*/
@@ -93,7 +83,6 @@ public interface Binder<T> {
/**
* Unbind a specific p2p or pub/sub message producer
*
* @param name the logical identity of the message target
* @param outboundBindTarget the channel bound as a producer
*/
@@ -101,7 +90,6 @@ public interface Binder<T> {
/**
* Bind a producer that expects async replies. To unbind, invoke unbindProducer() and unbindConsumer().
*
* @param name The name of the requestor.
* @param requests The interface used to send requests.
* @param replies The interface used to receive replies.
@@ -112,7 +100,6 @@ public interface Binder<T> {
/**
* Bind a consumer that handles requests from a requestor and asynchronously sends replies. To unbind, invoke
* unbindProducer() and unbindConsumer().
*
* @param name The name of the requestor for which this replier will handle requests.
* @param requests The interface used to send requests.
* @param replies The interface used to receive replies.
@@ -121,7 +108,7 @@ public interface Binder<T> {
void bindReplier(String name, T requests, T replies, Properties properties);
/**
* Create a object and bind a producer dynamically, creating the infrastructure
* Create an object and bind a producer dynamically, creating the infrastructure
* required by the binder technology.
* @param name The name of the "queue:" channel.
* @param properties arbitrary String key/value pairs that will be used in the binding.

11
spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/BinderHeaders.java
View File

@@ -22,7 +22,6 @@ import org.springframework.messaging.MessageHeaders;
/**
* Spring Integration message headers for XD.
*
* @author Gary Russell
* @author David Turanski
*/
@@ -47,13 +46,13 @@ public final class BinderHeaders {
* that have no inherent header support (by embedding the headers in the payload).
*/
public static final String[] STANDARD_HEADERS = new String[] {
IntegrationMessageHeaderAccessor.CORRELATION_ID,
IntegrationMessageHeaderAccessor.SEQUENCE_SIZE,
IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER,
IntegrationMessageHeaderAccessor.CORRELATION_ID,
IntegrationMessageHeaderAccessor.SEQUENCE_SIZE,
IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER,
BINDER_REPLY_CHANNEL,
MessageHeaders.CONTENT_TYPE,
MessageHeaders.CONTENT_TYPE,
BINDER_ORIGINAL_CONTENT_TYPE,
REPLY_TO,
REPLY_TO,
BINDER_HISTORY
};

2
spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/BinderCleaner.java → spring-cloud-stream-binders/spring-cloud-stream-binder-spi/src/main/java/org/springframework/cloud/stream/binder/BindingCleaner.java
View File

@@ -25,7 +25,7 @@ import java.util.Map;
* @author Gary Russell
* @since 1.2
*/
public interface BinderCleaner {
public interface BindingCleaner {
/**
* Clean up all resources for the supplied stream/job.

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

@@ -828,7 +828,7 @@ public abstract class MessageChannelBinderSupport
}
/**
* Attempt to create a direct binding (avoiding the binder) if the consumer is local. Named channel producers are not
* Attempt to create a direct binding (avoiding the broker) if the consumer is local. Named channel producers are not
* bound directly.
* @param name The name.
* @param moduleOutputChannel The channel to bind.
@@ -884,8 +884,8 @@ public abstract class MessageChannelBinderSupport
}
/**
* Attempt to bind a producer directly (avoiding the binder) if there is already a local producer. PubSub producers
* cannot be bound directly. Create the direct binding, then unbind the existing binder producer.
* Attempt to bind a producer directly (avoiding the broker) if there is already a local producer. PubSub producers
* cannot be bound directly. Create the direct binding, then unbind the existing producer.
* @param name The name.
* @param consumerChannel The channel to bind the producer to.
*/

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

@@ -74,7 +74,7 @@ public abstract class BrokerBinderTests extends
assertEquals(2, count.get());
assertNull(spyOn("direct.0").receive(true));
// Remove direct binding and bind producer to the binder
// Remove direct binding and bind the producer
binder.unbindConsumers("direct.0");
binderBindUnbindLatency();

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

@@ -23,6 +23,7 @@ import java.util.Properties;
import org.junit.Before;
import org.junit.Test;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport.JavaClassMimeTypeConversion;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -31,7 +32,6 @@ import org.springframework.messaging.converter.ContentTypeResolver;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport.JavaClassMimeTypeConversion;
import org.springframework.xd.dirt.integration.bus.serializer.kryo.PojoCodec;
import org.springframework.xd.tuple.DefaultTuple;
import org.springframework.xd.tuple.Tuple;
@@ -62,8 +62,7 @@ public class MessageChannelBinderSupportTests {
public void testBytesPassThru() {
byte[] payload = "foo".getBytes();
Message<byte[]> message = MessageBuilder.withPayload(payload).build();
MessageValues converted = binder.serializePayloadIfNecessary(message
);
MessageValues converted = binder.serializePayloadIfNecessary(message);
assertSame(payload, converted.getPayload());
Message<?> convertedMessage = converted.toMessage();
assertSame(payload, convertedMessage.getPayload());
@@ -81,8 +80,7 @@ public class MessageChannelBinderSupportTests {
Message<byte[]> message = MessageBuilder.withPayload(payload)
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_OCTET_STREAM_VALUE)
.build();
MessageValues messageValues = binder.serializePayloadIfNecessary(message
);
MessageValues messageValues = binder.serializePayloadIfNecessary(message);
Message<?> converted = messageValues.toMessage();
assertSame(payload, converted.getPayload());
assertEquals(MimeTypeUtils.APPLICATION_OCTET_STREAM,
@@ -130,8 +128,7 @@ public class MessageChannelBinderSupportTests {
@Test
public void testPojoSerialization() {
MessageValues convertedValues = binder.serializePayloadIfNecessary(
new GenericMessage<Foo>(new Foo("bar"))
);
new GenericMessage<Foo>(new Foo("bar")));
Message<?> converted = convertedValues.toMessage();
MimeType mimeType = contentTypeResolver.resolve(converted.getHeaders());
assertEquals("application", mimeType.getType());
@@ -146,8 +143,7 @@ public class MessageChannelBinderSupportTests {
@Test
public void testPojoWithXJavaObjectMimeTypeNoType() {
MessageValues convertedValues = binder.serializePayloadIfNecessary(
new GenericMessage<Foo>(new Foo("bar"))
);
new GenericMessage<Foo>(new Foo("bar")));
Message<?> converted = convertedValues.toMessage();
MimeType mimeType = contentTypeResolver.resolve(converted.getHeaders());
assertEquals("application", mimeType.getType());
@@ -162,8 +158,7 @@ public class MessageChannelBinderSupportTests {
@Test
public void testPojoWithXJavaObjectMimeTypeExplicitType() {
MessageValues convertedValues = binder.serializePayloadIfNecessary(
new GenericMessage<Foo>(new Foo("bar"))
);
new GenericMessage<Foo>(new Foo("bar")));
Message<?> converted = convertedValues.toMessage();
MimeType mimeType = contentTypeResolver.resolve(converted.getHeaders());
assertEquals("application", mimeType.getType());
@@ -178,8 +173,7 @@ public class MessageChannelBinderSupportTests {
@Test
public void testTupleSerialization() {
Tuple payload = TupleBuilder.tuple().of("foo", "bar");
MessageValues convertedValues = binder.serializePayloadIfNecessary(new GenericMessage<Tuple>(payload)
);
MessageValues convertedValues = binder.serializePayloadIfNecessary(new GenericMessage<Tuple>(payload));
Message<?> converted = convertedValues.toMessage();
MimeType mimeType = contentTypeResolver.resolve(converted.getHeaders());
assertEquals("application", mimeType.getType());

2
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/ChannelBindingAdapter.java
View File

@@ -53,7 +53,7 @@ import org.springframework.util.StringUtils;
import org.springframework.cloud.stream.binder.Binder;
/**
* Binds input/output channels to the binder.
* Binds input/output channels.
*
* @author Mark Fisher
* @author Dave Syer

4
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfiguration.java
View File

@@ -65,7 +65,7 @@ public class ChannelBindingAdapterConfiguration {
private Binder<MessageChannel> binder;
@Bean
public ChannelBindingAdapter binderAdapter() {
public ChannelBindingAdapter bindingAdapter() {
ChannelBindingAdapter adapter = new ChannelBindingAdapter(this.module, this.binder);
adapter.setOutputChannels(getOutputChannels());
adapter.setInputChannels(getInputChannels());
@@ -81,7 +81,7 @@ public class ChannelBindingAdapterConfiguration {
}
public void refresh() {
ChannelBindingAdapter adapter = binderAdapter();
ChannelBindingAdapter adapter = bindingAdapter();
adapter.setOutputChannels(getOutputChannels());
adapter.setInputChannels(getInputChannels());
}