From 6c880726b8099a92a0f6d4a12c84d9b322375591 Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Tue, 9 Feb 2016 17:53:23 -0500 Subject: [PATCH] Binder API simplification Resolves #330 - removes `unbind` from the binder and moves it to the `Binding` instance itself; - `Binding` is now an interface with a default implementation provided by SCS - Removed all methods from Binder except for unbind() - Removed old and unused code - Removed circular reference between default Binder and Binding implementations - Removed Binder type --- .../kafka/KafkaMessageChannelBinder.java | 16 +- .../stream/binder/kafka/KafkaBinderTests.java | 36 +-- .../binder/kafka/RawModeKafkaBinderTests.java | 47 ++-- .../binder/rabbit/RabbitBindingCleaner.java | 6 +- .../rabbit/RabbitMessageChannelBinder.java | 47 ++-- .../rabbit/RabbitBinderCleanerTests.java | 14 +- .../binder/rabbit/RabbitBinderTests.java | 89 +++--- .../redis/RedisMessageChannelBinder.java | 45 ++- .../stream/binder/redis/RedisBinderTests.java | 45 ++- .../stream/binder/AbstractBinderTests.java | 52 ++-- .../stream/binder/AbstractTestBinder.java | 7 +- .../binder/PartitionCapableBinderTests.java | 40 ++- .../MessageChannelBinderSupportTests.java | 9 +- .../stream/test/binder/TestSupportBinder.java | 33 ++- ...BinderSupport.java => AbstractBinder.java} | 196 ++----------- .../cloud/stream/binder/Binder.java | 7 +- .../cloud/stream/binder/Binding.java | 150 +--------- .../cloud/stream/binder/DefaultBinding.java | 83 ++++++ .../cloud/stream/binder/DirectHandler.java | 40 +++ .../stream/binding/BindableProxyFactory.java | 14 +- .../stream/binding/ChannelBindingService.java | 6 +- .../BinderAwareChannelResolverTests.java | 71 ++++- .../local/LocalMessageChannelBinder.java | 259 ------------------ .../stream/binder/stub1/StubBinder1.java | 4 - .../stream/binder/stub2/StubBinder2.java | 4 - .../stub2/StubBinder2ConfigurationA.java | 1 - .../binding/ChannelBindingServiceTests.java | 31 +-- .../MockBinderRegistryConfiguration.java | 3 - 28 files changed, 455 insertions(+), 900 deletions(-) rename spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/{MessageChannelBinderSupport.java => AbstractBinder.java} (82%) create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DirectHandler.java delete mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/local/LocalMessageChannelBinder.java diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 031b943e6..32e675561 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -36,13 +36,14 @@ import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.serialization.ByteArraySerializer; import org.springframework.beans.factory.DisposableBean; +import org.springframework.cloud.stream.binder.AbstractBinder; import org.springframework.cloud.stream.binder.BinderException; import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.binder.BinderPropertyKeys; import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.binder.DefaultBinding; import org.springframework.cloud.stream.binder.DefaultBindingPropertiesAccessor; import org.springframework.cloud.stream.binder.EmbeddedHeadersMessageConverter; -import org.springframework.cloud.stream.binder.MessageChannelBinderSupport; import org.springframework.cloud.stream.binder.MessageValues; import org.springframework.http.MediaType; import org.springframework.integration.channel.FixedSubscriberChannel; @@ -127,7 +128,7 @@ import scala.collection.Seq; * @author Mark Fisher * @author Soby Chacko */ -public class KafkaMessageChannelBinder extends MessageChannelBinderSupport { +public class KafkaMessageChannelBinder extends AbstractBinder { public static final ByteArraySerializer BYTE_ARRAY_SERIALIZER = new ByteArraySerializer(); @@ -489,10 +490,8 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport { consumer.setBeanFactory(this.getBeanFactory()); consumer.setBeanName("outbound." + name); consumer.afterPropertiesSet(); - Binding producerBinding = Binding.forProducer(name, moduleOutputChannel, consumer, - producerPropertiesAccessor); - addBinding(producerBinding); - producerBinding.start(); + DefaultBinding producerBinding = new DefaultBinding<>(name, null, moduleOutputChannel, consumer, producerPropertiesAccessor); + consumer.start(); return producerBinding; } catch (Exception e) { @@ -644,9 +643,8 @@ public class KafkaMessageChannelBinder extends MessageChannelBinderSupport { String groupedName = groupedName(name, group); edc.setBeanName("inbound." + groupedName); - Binding consumerBinding = Binding.forConsumer(name, group, edc, moduleInputChannel, accessor); - addBinding(consumerBinding); - consumerBinding.start(); + DefaultBinding consumerBinding = new DefaultBinding<>(name, group, moduleInputChannel, edc, accessor); + edc.start(); return consumerBinding; } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 70b2a9483..383d5d8d3 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -174,8 +174,8 @@ public class KafkaBinderTests extends PartitionCapableBinderTests { Message inbound = receive(moduleInputChannel); assertNotNull(inbound); assertArrayEquals(ratherBigPayload, (byte[]) inbound.getPayload()); - binder.unbind(producerBinding); - binder.unbind(consumerBinding); + producerBinding.unbind(); + consumerBinding.unbind(); } } @@ -205,8 +205,8 @@ public class KafkaBinderTests extends PartitionCapableBinderTests { Collection partitions = binder.getCoreBinder().getConnectionFactory().getPartitions( "foo" + uniqueBindingId + ".0"); assertThat(partitions, hasSize(10)); - binder.unbind(producerBinding); - binder.unbind(consumerBinding); + producerBinding.unbind(); + consumerBinding.unbind(); } @Test @@ -238,8 +238,8 @@ public class KafkaBinderTests extends PartitionCapableBinderTests { Collection partitions = binder.getCoreBinder().getConnectionFactory().getPartitions( "foo" + uniqueBindingId + ".0"); assertThat(partitions, hasSize(6)); - binder.unbind(producerBinding); - binder.unbind(consumerBinding); + producerBinding.unbind(); + consumerBinding.unbind(); } @Test @@ -270,8 +270,8 @@ public class KafkaBinderTests extends PartitionCapableBinderTests { Collection partitions = binder.getCoreBinder().getConnectionFactory().getPartitions( "foo" + uniqueBindingId + ".0"); assertThat(partitions, hasSize(6)); - binder.unbind(producerBinding); - binder.unbind(consumerBinding); + producerBinding.unbind(); + consumerBinding.unbind(); } @Test @@ -302,8 +302,8 @@ public class KafkaBinderTests extends PartitionCapableBinderTests { Collection partitions = binder.getCoreBinder().getConnectionFactory().getPartitions( "foo" + uniqueBindingId + ".0"); assertThat(partitions, hasSize(5)); - binder.unbind(producerBinding); - binder.unbind(consumerBinding); + producerBinding.unbind(); + consumerBinding.unbind(); } @Test @@ -334,8 +334,8 @@ public class KafkaBinderTests extends PartitionCapableBinderTests { Collection partitions = binder.getCoreBinder().getConnectionFactory().getPartitions( "foo" + uniqueBindingId + ".0"); assertThat(partitions, hasSize(5)); - binder.unbind(producerBinding); - binder.unbind(consumerBinding); + producerBinding.unbind(); + consumerBinding.unbind(); } @Test @@ -422,7 +422,7 @@ public class KafkaBinderTests extends PartitionCapableBinderTests { Message receivedMessage2 = (Message) receive(input1); assertThat(receivedMessage2, not(nullValue())); assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload2)); - binder.unbind(consumerBinding); + consumerBinding.unbind(); String testPayload3 = "foo-" + UUID.randomUUID().toString(); output.send(new GenericMessage<>(testPayload3.getBytes())); @@ -438,8 +438,8 @@ public class KafkaBinderTests extends PartitionCapableBinderTests { Message receivedMessage6 = (Message) receive(input1); assertThat(receivedMessage6, not(nullValue())); assertThat(new String(receivedMessage6.getPayload()), equalTo(testPayload3)); - binder.unbind(consumerBinding); - binder.unbind(producerBinding); + consumerBinding.unbind(); + producerBinding.unbind(); } @Test @@ -468,7 +468,7 @@ public class KafkaBinderTests extends PartitionCapableBinderTests { Message receivedMessage2 = (Message) receive(input1); assertThat(receivedMessage2, not(nullValue())); assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload2)); - binder.unbind(consumerBinding); + consumerBinding.unbind(); String testPayload3 = "foo-" + UUID.randomUUID().toString(); output.send(new GenericMessage<>(testPayload3.getBytes())); @@ -478,7 +478,7 @@ public class KafkaBinderTests extends PartitionCapableBinderTests { Message receivedMessage3 = (Message) receive(input1); assertThat(receivedMessage3, not(nullValue())); assertThat(new String(receivedMessage3.getPayload()), equalTo(testPayload3)); - binder.unbind(consumerBinding); - binder.unbind(producerBinding); + consumerBinding.unbind(); + producerBinding.unbind(); } } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java index 962af27e9..8dbc440e7 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java @@ -23,10 +23,8 @@ 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 java.util.Arrays; -import java.util.List; import java.util.Properties; import org.junit.Ignore; @@ -36,7 +34,6 @@ import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.binder.BinderPropertyKeys; import org.springframework.cloud.stream.binder.Binding; -import org.springframework.cloud.stream.binder.TestUtils; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; @@ -72,10 +69,6 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { DirectChannel output = new DirectChannel(); output.setBeanName("test.output"); Binding outputBinding = binder.bindProducer("partJ.0", output, properties); - @SuppressWarnings("unchecked") - List> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class); - assertEquals(1, bindings.size()); - properties.clear(); properties.put("concurrency", "2"); properties.put("count","3"); @@ -109,10 +102,10 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { ((byte[]) receive2.getPayload())[0]), containsInAnyOrder((byte)0, (byte)1, (byte)2)); - binder.unbind(input0Binding); - binder.unbind(input1Binding); - binder.unbind(input2Binding); - binder.unbind(outputBinding); + input0Binding.unbind(); + input1Binding.unbind(); + input2Binding.unbind(); + outputBinding.unbind(); } @Test @@ -128,11 +121,8 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { DirectChannel output = new DirectChannel(); output.setBeanName("test.output"); Binding outputBinding = binder.bindProducer("part.0", output, properties); - @SuppressWarnings("unchecked") - List> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class); - assertEquals(1, bindings.size()); try { - AbstractEndpoint endpoint = bindings.get(0).getEndpoint(); + AbstractEndpoint endpoint = extractEndpoint(outputBinding); assertThat(getEndpointRouting(endpoint), containsString("part.0-' + headers['partition']")); } catch (UnsupportedOperationException ignored) { @@ -179,10 +169,10 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { ((byte[]) receive2.getPayload())[0]), containsInAnyOrder((byte)0, (byte)1, (byte)2)); - binder.unbind(input0Binding); - binder.unbind(input1Binding); - binder.unbind(input2Binding); - binder.unbind(outputBinding); + input0Binding.unbind(); + input1Binding.unbind(); + input2Binding.unbind(); + outputBinding.unbind(); } @Test @@ -200,8 +190,8 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { Message inbound = receive(moduleInputChannel); assertNotNull(inbound); assertEquals("foo", new String((byte[])inbound.getPayload())); - binder.unbind(producerBinding); - binder.unbind(consumerBinding); + producerBinding.unbind(); + consumerBinding.unbind(); } // Ignored, since raw mode does not support headers @@ -251,7 +241,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { assertEquals("foo", new String((byte[]) tapped2.getPayload())); } // delete one tap stream is deleted - binder.unbind(input3Binding); + input3Binding.unbind(); Message message2 = MessageBuilder.withPayload("bar".getBytes()).build(); moduleOutputChannel.send(message2); @@ -267,11 +257,14 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { assertNotNull(receive(module3InputChannel)); // clean up - binder.unbind(input1Binding); - binder.unbind(input2Binding); - binder.unbind(input3Binding); - binder.unbind(producerBinding); - assertTrue(getBindings(binder).isEmpty()); + input1Binding.unbind(); + input2Binding.unbind(); + input3Binding.unbind(); + producerBinding.unbind(); + assertFalse(extractEndpoint(input1Binding).isRunning()); + assertFalse(extractEndpoint(input2Binding).isRunning()); + assertFalse(extractEndpoint(input3Binding).isRunning()); + assertFalse(extractEndpoint(producerBinding).isRunning()); } private void assertMessageReceive(QueueChannel moduleInputChannel, String payload) { diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitBindingCleaner.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitBindingCleaner.java index d1ded2561..28a820659 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitBindingCleaner.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitBindingCleaner.java @@ -26,7 +26,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.cloud.stream.binder.BindingCleaner; -import org.springframework.cloud.stream.binder.MessageChannelBinderSupport; +import org.springframework.cloud.stream.binder.AbstractBinder; import org.springframework.web.client.RestTemplate; import org.springframework.web.util.UriComponentsBuilder; @@ -102,7 +102,7 @@ public class RabbitBindingCleaner implements BindingCleaner { private List findStreamQueues(String adminUri, String vhost, String binderPrefix, String stream, RestTemplate restTemplate) { - String queueNamePrefix = adjustPrefix(MessageChannelBinderSupport.applyPrefix(binderPrefix, stream)); + String queueNamePrefix = adjustPrefix(AbstractBinder.applyPrefix(binderPrefix, stream)); List> queues = listAllQueues(adminUri, vhost, restTemplate); List removedQueues = new ArrayList<>(); for (Map queue : queues) { @@ -147,7 +147,7 @@ public class RabbitBindingCleaner implements BindingCleaner { .pathSegment("exchanges", "{vhost}") .buildAndExpand(vhost).encode().toUri(); List> exchanges = restTemplate.getForObject(uri, List.class); - String exchangeNamePrefix = adjustPrefix(MessageChannelBinderSupport.applyPrefix(binderPrefix, entity)); + String exchangeNamePrefix = adjustPrefix(AbstractBinder.applyPrefix(binderPrefix, entity)); for (Map exchange : exchanges) { String exchangeName = (String) exchange.get("name"); if (exchangeName.startsWith(exchangeNamePrefix)) { diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index 8596fe07e..e655ce755 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -59,13 +59,13 @@ import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; import org.springframework.amqp.support.AmqpHeaders; import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor; import org.springframework.amqp.support.postprocessor.GZipPostProcessor; -import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.beans.factory.support.DefaultListableBeanFactory; +import org.springframework.cloud.stream.binder.AbstractBinder; import org.springframework.cloud.stream.binder.BinderPropertyKeys; import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.binder.DefaultBinding; import org.springframework.cloud.stream.binder.DefaultBindingPropertiesAccessor; -import org.springframework.cloud.stream.binder.MessageChannelBinderSupport; import org.springframework.cloud.stream.binder.MessageValues; import org.springframework.context.Lifecycle; import org.springframework.context.support.GenericApplicationContext; @@ -108,7 +108,7 @@ import com.rabbitmq.client.Envelope; * @author David Turanski * @author Marius Bogoevici */ -public class RabbitMessageChannelBinder extends MessageChannelBinderSupport implements DisposableBean { +public class RabbitMessageChannelBinder extends AbstractBinder { public static final AnonymousQueue.Base64UrlNamingStrategy ANONYMOUS_GROUP_NAME_GENERATOR = new AnonymousQueue.Base64UrlNamingStrategy("anonymous."); @@ -455,9 +455,9 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl return args; } - private Binding doRegisterConsumer(String name, String group, MessageChannel moduleInputChannel, Queue queue, - RabbitPropertiesAccessor properties) { - Binding consumerBinding = null; + private Binding doRegisterConsumer(final String name, String group, MessageChannel moduleInputChannel, Queue queue, + final RabbitPropertiesAccessor properties) { + DefaultBinding consumerBinding = null; // Fix for XD-2503 // Temporarily overrides the thread context classloader with the one where the SimpleMessageListenerContainer // is defined @@ -511,14 +511,19 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl mapper.setReplyHeaderNames(properties.getReplyHeaderPattens(this.defaultReplyHeaderPatterns)); adapter.setHeaderMapper(mapper); adapter.afterPropertiesSet(); - consumerBinding = Binding.forConsumer(name, group, adapter, moduleInputChannel, properties); - addBinding(consumerBinding); + consumerBinding = new DefaultBinding(name, group, moduleInputChannel, adapter, properties) { + + @Override + protected void afterUnbind() { + cleanAutoDeclareContext(properties.getPrefix(defaultPrefix), name); + } + }; ReceivingHandler convertingBridge = new ReceivingHandler(); convertingBridge.setOutputChannel(moduleInputChannel); convertingBridge.setBeanName(name + ".convert.bridge"); convertingBridge.afterPropertiesSet(); bridgeToModuleChannel.subscribe(convertingBridge); - consumerBinding.start(); + adapter.start(); } finally { Thread.currentThread().setContextClassLoader(originalClassloader); @@ -635,9 +640,9 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl consumer.setBeanFactory(getBeanFactory()); consumer.setBeanName("outbound." + name); consumer.afterPropertiesSet(); - Binding producerBinding = Binding.forProducer(name, moduleOutputChannel, consumer, properties); - addBinding(producerBinding); - producerBinding.start(); + DefaultBinding producerBinding = new DefaultBinding<>(name, null, moduleOutputChannel, consumer, properties); + + consumer.start(); return producerBinding; } @@ -714,13 +719,6 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl return prefix + DEAD_LETTER_EXCHANGE; } - @Override - protected void afterUnbind(Binding binding) { - if (Binding.Type.consumer.equals(binding.getType())) { - cleanAutoDeclareContext(binding); - } - } - private void addToAutoDeclareContext(String name, Object bean) { synchronized (this.autoDeclareContext) { if (!this.autoDeclareContext.containsBean(name)) { @@ -729,11 +727,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl } } - private void cleanAutoDeclareContext(Binding binding) { - Assert.isTrue(binding.getPropertiesAccessor() instanceof RabbitPropertiesAccessor, - "Binding was not created by this binder"); - String prefix = ((RabbitPropertiesAccessor) binding.getPropertiesAccessor()).getPrefix(this.defaultPrefix); - String name = binding.getName(); + public void cleanAutoDeclareContext(String prefix, String name) { synchronized(this.autoDeclareContext) { removeSingleton(applyPrefix(prefix,name) + ".binding"); removeSingleton(applyPrefix(prefix,name)); @@ -752,11 +746,6 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl } } - @Override - public void destroy() { - stopBindings(); - } - @Override public void doManualAck(LinkedList messageHeadersList) { Iterator iterator = messageHeadersList.iterator(); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java index 68058fca9..bc82b4775 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java @@ -38,7 +38,7 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.core.ChannelCallback; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; -import org.springframework.cloud.stream.binder.MessageChannelBinderSupport; +import org.springframework.cloud.stream.binder.AbstractBinder; import org.springframework.cloud.stream.test.junit.rabbit.RabbitTestSupport; import org.springframework.web.client.RestTemplate; import org.springframework.web.util.UriComponentsBuilder; @@ -69,8 +69,8 @@ public class RabbitBinderCleanerTests { CachingConnectionFactory connectionFactory = rabbitWithMgmtEnabled.getResource(); RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory); for (int i = 0; i < 5; i++) { - String queue1Name = MessageChannelBinderSupport.applyPrefix(BINDER_PREFIX, stream1 + ".default." + i); - String queue2Name = MessageChannelBinderSupport.applyPrefix(BINDER_PREFIX, stream2 + ".default." + i); + String queue1Name = AbstractBinder.applyPrefix(BINDER_PREFIX, stream1 + ".default." + i); + String queue2Name = AbstractBinder.applyPrefix(BINDER_PREFIX, stream2 + ".default." + i); if (firstQueue == null) { firstQueue = queue1Name; } @@ -86,7 +86,7 @@ public class RabbitBinderCleanerTests { template.put(uri, new AmqpQueue(false, true)); uri = UriComponentsBuilder.fromUriString("http://localhost:15672/api/queues") .pathSegment("{vhost}", "{queue}") - .buildAndExpand("/", MessageChannelBinderSupport.constructDLQName(queue1Name)).encode().toUri(); + .buildAndExpand("/", AbstractBinder.constructDLQName(queue1Name)).encode().toUri(); template.put(uri, new AmqpQueue(false, true)); TopicExchange exchange = new TopicExchange(queue1Name); rabbitAdmin.declareExchange(exchange); @@ -96,21 +96,21 @@ public class RabbitBinderCleanerTests { rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(queue2Name)).to(exchange).with(queue2Name)); } final TopicExchange topic1 = new TopicExchange( - MessageChannelBinderSupport.applyPrefix(BINDER_PREFIX, stream1 + ".foo.bar")); + AbstractBinder.applyPrefix(BINDER_PREFIX, stream1 + ".foo.bar")); rabbitAdmin.declareExchange(topic1); rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(firstQueue)).to(topic1).with("#")); String foreignQueue = UUID.randomUUID().toString(); rabbitAdmin.declareQueue(new Queue(foreignQueue)); rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(foreignQueue)).to(topic1).with("#")); final TopicExchange topic2 = new TopicExchange( - MessageChannelBinderSupport.applyPrefix(BINDER_PREFIX, stream2 + ".foo.bar")); + AbstractBinder.applyPrefix(BINDER_PREFIX, stream2 + ".foo.bar")); rabbitAdmin.declareExchange(topic2); rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(firstQueue)).to(topic2).with("#")); new RabbitTemplate(connectionFactory).execute(new ChannelCallback() { @Override public Void doInRabbit(Channel channel) throws Exception { - String queueName = MessageChannelBinderSupport.applyPrefix(BINDER_PREFIX, stream1 + ".default." + 4); + String queueName = AbstractBinder.applyPrefix(BINDER_PREFIX, stream1 + ".default." + 4); String consumerTag = channel.basicConsume(queueName, new DefaultConsumer(channel)); try { waitForConsumerStateNot(queueName, 0); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index 8d95e7f5d..1cb06b8f2 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -122,8 +122,8 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { }); moduleOutputChannel.send(message); assertTrue(latch.await(10, TimeUnit.SECONDS)); - binder.unbind(consumerBinding); - binder.unbind(producerBinding); + producerBinding.unbind(); + consumerBinding.unbind(); } @Test @@ -132,10 +132,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { Properties properties = new Properties(); properties.put("transacted", "true"); // test transacted with defaults; not allowed with ackmode NONE Binding consumerBinding = binder.bindConsumer("props.0", null, new DirectChannel(), properties); - @SuppressWarnings("unchecked") - List> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class); - assertEquals(1, bindings.size()); - AbstractEndpoint endpoint = bindings.get(0).getEndpoint(); + AbstractEndpoint endpoint = extractEndpoint(consumerBinding); SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint, "messageListenerContainer", SimpleMessageListenerContainer.class); assertEquals(AcknowledgeMode.AUTO, container.getAcknowledgeMode()); @@ -152,8 +149,8 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { 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")); - binder.unbind(consumerBinding); - assertEquals(0, bindings.size()); + consumerBinding.unbind(); + assertFalse(endpoint.isRunning()); properties = new Properties(); properties.put("ackMode", "NONE"); @@ -171,16 +168,13 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { properties.put("partitionIndex", 0); consumerBinding = binder.bindConsumer("props.0", "test", new DirectChannel(), properties); - @SuppressWarnings("unchecked") - List> bindingsNow = TestUtils.getPropertyValue(binder, "binder.bindings", List.class); - assertEquals(1, bindingsNow.size()); - endpoint = bindingsNow.get(0).getEndpoint(); + endpoint = extractEndpoint(consumerBinding); container = verifyContainer(endpoint); assertEquals("foo.props.0.test", container.getQueueNames()[0]); - binder.unbind(consumerBinding); - assertEquals(0, bindingsNow.size()); + consumerBinding.unbind(); + assertFalse(endpoint.isRunning()); } @Test @@ -188,17 +182,15 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { Binder binder = getBinder(); Binding producerBinding = binder.bindProducer("props.0", new DirectChannel(), null); @SuppressWarnings("unchecked") - List> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class); - assertEquals(1, bindings.size()); - AbstractEndpoint endpoint = bindings.get(0).getEndpoint(); + AbstractEndpoint endpoint = extractEndpoint(producerBinding); MessageDeliveryMode mode = TestUtils.getPropertyValue(endpoint, "handler.delegate.defaultDeliveryMode", MessageDeliveryMode.class); assertEquals(MessageDeliveryMode.PERSISTENT, mode); List requestHeaders = TestUtils.getPropertyValue(endpoint, "handler.delegate.headerMapper.requestHeaderMatcher.strategies", List.class); assertEquals(2, requestHeaders.size()); - binder.unbind(producerBinding); - assertEquals(0, bindings.size()); + producerBinding.unbind(); + assertFalse(endpoint.isRunning()); Properties properties = new Properties(); properties.put("prefix", "foo."); @@ -211,8 +203,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { properties.put(BinderPropertyKeys.NEXT_MODULE_COUNT, "1"); producerBinding = binder.bindProducer("props.0", new DirectChannel(), properties); - assertEquals(1, bindings.size()); - endpoint = bindings.get(0).getEndpoint(); + endpoint = extractEndpoint(producerBinding); assertEquals( "'props.0-' + headers['partition']", TestUtils.getPropertyValue(endpoint, "handler.delegate.routingKeyExpression", @@ -222,8 +213,8 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { assertEquals(MessageDeliveryMode.NON_PERSISTENT, mode); verifyFooRequestProducer(endpoint); - binder.unbind(producerBinding); - assertEquals(0, bindings.size()); + producerBinding.unbind(); + assertFalse(endpoint.isRunning()); } @Test @@ -264,7 +255,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { } assertTrue(n < 100); - binder.unbind(consumerBinding); + consumerBinding.unbind(); assertNotNull(admin.getQueueProperties(TEST_PREFIX + "durabletest.0.tgroup.dlq")); } @@ -291,7 +282,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { }); Binding consumerBinding = binder.bindConsumer("nondurabletest.0", "tgroup", moduleInputChannel, properties); - binder.unbind(consumerBinding); + consumerBinding.unbind(); assertNull(admin.getQueueProperties(TEST_PREFIX + "nondurabletest.0.dlq")); } @@ -330,7 +321,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { } assertTrue(n < 100); - binder.unbind(consumerBinding); + consumerBinding.unbind(); ApplicationContext context = TestUtils.getPropertyValue(binder, "binder.autoDeclareContext", ApplicationContext.class); @@ -419,11 +410,11 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { assertNotNull(received); assertEquals(0, received.getMessageProperties().getHeaders().get("partition")); - binder.unbind(input0Binding); - binder.unbind(input1Binding); - binder.unbind(defaultConsumerBinding1); - binder.unbind(defaultConsumerBinding2); - binder.unbind(outputBinding); + input0Binding.unbind(); + input1Binding.unbind(); + defaultConsumerBinding1.unbind(); + defaultConsumerBinding2.unbind(); + outputBinding.unbind(); } @Test @@ -505,11 +496,11 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { assertNotNull(received); assertEquals(0, received.getMessageProperties().getHeaders().get("partition")); - binder.unbind(input0Binding); - binder.unbind(input1Binding); - binder.unbind(defaultConsumerBinding1); - binder.unbind(defaultConsumerBinding2); - binder.unbind(outputBinding); + input0Binding.unbind(); + input1Binding.unbind(); + defaultConsumerBinding1.unbind(); + defaultConsumerBinding2.unbind(); + outputBinding.unbind(); } @Test @@ -557,7 +548,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { } assertTrue(n < 100); - binder.unbind(consumerBinding); + consumerBinding.unbind(); } @SuppressWarnings("unchecked") @@ -613,8 +604,8 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { assertEquals("bar", new String(in.getPayload())); assertNull(in.getHeaders().get(AmqpHeaders.DELIVERY_MODE)); - binder.unbind(producerBinding); - binder.unbind(consumerBinding); + producerBinding.unbind(); + consumerBinding.unbind(); } /* @@ -700,16 +691,16 @@ public class RabbitBinderTests extends PartitionCapableBinderTests { assertNotNull(message); assertEquals("1", message.getPayload()); - binder.unbind(late0ProducerBinding); - binder.unbind(late0ConsumerBinding); - binder.unbind(partlate0ProducerBinding); - binder.unbind(partlate0Consumer0Binding); - binder.unbind(partlate0Consumer1Binding); - binder.unbind(noDlqProducerBinding); - binder.unbind(noDlqConsumerBinding); - binder.unbind(pubSubProducerBinding); - binder.unbind(nonDurableConsumerBinding); - binder.unbind(durableConsumerBinding); + late0ProducerBinding.unbind(); + late0ConsumerBinding.unbind(); + partlate0ProducerBinding.unbind(); + partlate0Consumer0Binding.unbind(); + partlate0Consumer1Binding.unbind(); + noDlqProducerBinding.unbind(); + noDlqConsumerBinding.unbind(); + pubSubProducerBinding.unbind(); + nonDurableConsumerBinding.unbind(); + durableConsumerBinding.unbind(); binder.cleanup(); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java index 1fbb70020..141c1644b 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java @@ -25,13 +25,13 @@ import java.util.Properties; import java.util.Set; import java.util.UUID; -import org.springframework.beans.factory.DisposableBean; -import org.springframework.cloud.stream.binder.DefaultBindingPropertiesAccessor; +import org.springframework.cloud.stream.binder.AbstractBinder; import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.binder.BinderPropertyKeys; import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.binder.DefaultBinding; +import org.springframework.cloud.stream.binder.DefaultBindingPropertiesAccessor; import org.springframework.cloud.stream.binder.EmbeddedHeadersMessageConverter; -import org.springframework.cloud.stream.binder.MessageChannelBinderSupport; import org.springframework.cloud.stream.binder.MessageValues; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.RedisOperations; @@ -65,11 +65,11 @@ import org.springframework.util.StringUtils; * @author David Turanski * @author Jennifer Hickey */ -public class RedisMessageChannelBinder extends MessageChannelBinderSupport implements DisposableBean { +public class RedisMessageChannelBinder extends AbstractBinder { private static final String ERROR_HEADER = "errorKey"; - private static final String CONSUMER_GROUPS_KEY_PREFIX = "groups."; + static final String CONSUMER_GROUPS_KEY_PREFIX = "groups."; private static final SpelExpressionParser parser = new SpelExpressionParser(); @@ -170,7 +170,7 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple } private Binding doRegisterConsumer(String bindingName, String group, String channelName, MessageChannel moduleInputChannel, - MessageProducerSupport adapter, RedisPropertiesAccessor properties) { + MessageProducerSupport adapter, final RedisPropertiesAccessor properties) { DirectChannel bridgeToModuleChannel = new DirectChannel(); bridgeToModuleChannel.setBeanFactory(this.getBeanFactory()); bridgeToModuleChannel.setBeanName(channelName + ".bridge"); @@ -178,15 +178,21 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple adapter.setOutputChannel(bridgeInputChannel); adapter.setBeanName("inbound." + channelName); adapter.afterPropertiesSet(); - Binding consumerBinding = Binding.forConsumer(channelName, group, adapter, moduleInputChannel, properties); - addBinding(consumerBinding); + DefaultBinding consumerBinding = new DefaultBinding(channelName, group, moduleInputChannel, adapter, properties) { + + @Override + protected void afterUnbind() { + String key = RedisMessageChannelBinder.CONSUMER_GROUPS_KEY_PREFIX + getName(); + RedisMessageChannelBinder.this.redisOperations.boundZSetOps(key).incrementScore(getGroup(), -1); + } + }; ReceivingHandler convertingBridge = new ReceivingHandler(); convertingBridge.setOutputChannel(moduleInputChannel); convertingBridge.setBeanName(channelName + ".bridge.handler"); convertingBridge.afterPropertiesSet(); bridgeToModuleChannel.subscribe(convertingBridge); this.redisOperations.boundZSetOps(CONSUMER_GROUPS_KEY_PREFIX + bindingName).incrementScore(group, 1); - consumerBinding.start(); + adapter.start(); return consumerBinding; } @@ -247,17 +253,6 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple } } - @Override - protected void afterUnbind(Binding binding) { - if (Binding.Type.consumer.equals(binding.getType())) { - String key = CONSUMER_GROUPS_KEY_PREFIX + binding.getName(); - boolean durable = binding.getPropertiesAccessor().isDurable(defaultDurableSubscription); - if (!durable) { - this.redisOperations.boundZSetOps(key).incrementScore(binding.getGroup(), -1); - } - } - } - @Override public Binding bindProducer(final String name, MessageChannel moduleOutputChannel, Properties properties) { Assert.isInstanceOf(SubscribableChannel.class, moduleOutputChannel); @@ -290,17 +285,11 @@ public class RedisMessageChannelBinder extends MessageChannelBinderSupport imple consumer.setBeanFactory(this.getBeanFactory()); consumer.setBeanName("outbound." + name); consumer.afterPropertiesSet(); - Binding producerBinding = Binding.forProducer(name, moduleOutputChannel, consumer, properties); - addBinding(producerBinding); - producerBinding.start(); + DefaultBinding producerBinding = new DefaultBinding<>(name, null, moduleOutputChannel, consumer, properties); + consumer.start(); return producerBinding; } - @Override - public void destroy() { - stopBindings(); - } - private class SendingHandler extends AbstractMessageHandler { private final String bindingName; diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisBinderTests.java index 367f548c6..8b55f12fb 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/test/java/org/springframework/cloud/stream/binder/redis/RedisBinderTests.java @@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.redis; import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.instanceOf; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; @@ -91,15 +92,11 @@ public class RedisBinderTests extends PartitionCapableBinderTests { Properties properties = new Properties(); properties.put("maxAttempts", "1"); // disable retry Binding binding = binder.bindConsumer("props.0", "test", new DirectChannel(), properties); - @SuppressWarnings("unchecked") - List> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class); - assertEquals(1, bindings.size()); - assertEquals(binding, bindings.get(0)); - AbstractEndpoint endpoint = binding.getEndpoint(); + AbstractEndpoint endpoint = extractEndpoint(binding); assertThat(endpoint, instanceOf(RedisQueueMessageDrivenEndpoint.class)); assertSame(DirectChannel.class, TestUtils.getPropertyValue(endpoint, "outputChannel").getClass()); - binder.unbind(binding); - assertEquals(0, bindings.size()); + binding.unbind(); + assertFalse(endpoint.isRunning()); properties.put("backOffInitialInterval", "2000"); properties.put("backOffMaxInterval", "20000"); @@ -109,13 +106,11 @@ public class RedisBinderTests extends PartitionCapableBinderTests { properties.put("partitionIndex", 0); binding = binder.bindConsumer("props.0", "test", new DirectChannel(), properties); - assertEquals(1, bindings.size()); - assertEquals(binding, bindings.get(0)); - endpoint = binding.getEndpoint(); + endpoint = extractEndpoint(binding); verifyConsumer(endpoint); - binder.unbind(binding); - assertEquals(0, bindings.size()); + binding.unbind(); + assertFalse(endpoint.isRunning()); } @Test @@ -123,19 +118,15 @@ public class RedisBinderTests extends PartitionCapableBinderTests { Binder binder = getBinder(); Binding consumerBinding = binder.bindConsumer("props.0", "test", new DirectChannel(), null); Binding producerBinding = binder.bindProducer("props.0", new DirectChannel(), null); + AbstractEndpoint producerEndpoint = extractEndpoint(producerBinding); @SuppressWarnings("unchecked") - List> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class); - assertEquals(2, bindings.size()); - assertEquals(producerBinding, bindings.get(1)); - AbstractEndpoint endpoint = producerBinding.getEndpoint(); - @SuppressWarnings("unchecked") - Map adapters = TestUtils.getPropertyValue(endpoint, "handler.adapters", Map.class); + Map adapters = TestUtils.getPropertyValue(producerEndpoint, "handler.adapters", Map.class); RedisQueueOutboundChannelAdapter adapter = adapters.get("test"); assertEquals( "props.0.test", TestUtils.getPropertyValue(adapter, "queueNameExpression", Expression.class).getExpressionString()); - binder.unbind(producerBinding); - assertEquals(1, bindings.size()); + producerBinding.unbind(); + assertFalse(producerEndpoint.isRunning()); Properties properties = new Properties(); properties.put("partitionKeyExpression", "'foo'"); @@ -145,16 +136,16 @@ public class RedisBinderTests extends PartitionCapableBinderTests { properties.put(BinderPropertyKeys.NEXT_MODULE_COUNT, "1"); producerBinding = binder.bindProducer("props.0", new DirectChannel(), properties); - assertEquals(2, bindings.size()); - endpoint = bindings.get(1).getEndpoint(); - adapter = (RedisQueueOutboundChannelAdapter) TestUtils.getPropertyValue(endpoint, "handler.adapters", Map.class).get("test"); + producerEndpoint = extractEndpoint(producerBinding); + adapter = (RedisQueueOutboundChannelAdapter) TestUtils.getPropertyValue(producerEndpoint, "handler.adapters", Map.class).get("test"); assertEquals( "'props.0.test-' + headers['partition']", TestUtils.getPropertyValue(adapter, "queueNameExpression", Expression.class).getExpressionString()); - binder.unbind(producerBinding); - binder.unbind(consumerBinding); - assertEquals(0, bindings.size()); + producerBinding.unbind(); + consumerBinding.unbind(); + assertFalse(producerEndpoint.isRunning()); + assertFalse(extractEndpoint(consumerBinding).isRunning()); } private void verifyConsumer(AbstractEndpoint endpoint) { @@ -190,7 +181,7 @@ public class RedisBinderTests extends PartitionCapableBinderTests { Object rightPop = template.boundListOps("ERRORS:retry.0.test").rightPop(5, TimeUnit.SECONDS); assertNotNull(rightPop); assertThat(new String((byte[]) rightPop), containsString("foo")); - binder.unbind(consumerBinding); + consumerBinding.unbind(); } @Test diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java index 499a67f52..ea113bc5f 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java @@ -20,23 +20,22 @@ import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.hasProperty; import static org.hamcrest.collection.IsArrayContainingInAnyOrder.arrayContainingInAnyOrder; 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 java.util.Collection; import java.util.Collections; -import java.util.List; import java.util.UUID; import org.junit.After; import org.junit.Test; -import org.springframework.beans.DirectFieldAccessor; import org.springframework.http.MediaType; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -78,16 +77,16 @@ public abstract class AbstractBinderTests { Binding foo1ProducerBinding = binder.bindProducer("foo.1", new DirectChannel(), null); Binding foo1ConsumerBinding = binder.bindConsumer("foo.1", "test", new DirectChannel(), null); Binding foo2ProducerBinding = binder.bindProducer("foo.2", new DirectChannel(), null); - Collection bindings = getBindings(binder); - assertEquals(5, bindings.size()); - binder.unbind(foo0ProducerBinding); - assertEquals(4, bindings.size()); - binder.unbind(foo0ConsumerBinding); - binder.unbind(foo1ProducerBinding); - assertEquals(2, bindings.size()); - binder.unbind(foo1ConsumerBinding); - binder.unbind(foo2ProducerBinding); - assertTrue(bindings.isEmpty()); + foo0ProducerBinding.unbind(); + assertFalse(TestUtils.getPropertyValue(foo0ProducerBinding, "endpoint", AbstractEndpoint.class).isRunning()); + foo0ConsumerBinding.unbind(); + foo1ProducerBinding.unbind(); + assertFalse(TestUtils.getPropertyValue(foo0ConsumerBinding, "endpoint", AbstractEndpoint.class).isRunning()); + assertFalse(TestUtils.getPropertyValue(foo1ProducerBinding, "endpoint", AbstractEndpoint.class).isRunning()); + foo1ConsumerBinding.unbind(); + foo2ProducerBinding.unbind(); + assertFalse(TestUtils.getPropertyValue(foo1ConsumerBinding, "endpoint", AbstractEndpoint.class).isRunning()); + assertFalse(TestUtils.getPropertyValue(foo2ProducerBinding, "endpoint", AbstractEndpoint.class).isRunning()); } @Test @@ -107,8 +106,8 @@ public abstract class AbstractBinderTests { assertEquals("foo", inbound.getPayload()); assertNull(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)); assertEquals("foo/bar", inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE)); - binder.unbind(producerBinding); - binder.unbind(consumerBinding); + producerBinding.unbind(); + consumerBinding.unbind(); } @Test @@ -148,11 +147,11 @@ public abstract class AbstractBinderTests { hasProperty("payload", equalTo(testPayload2.getBytes())))); - binder.unbind(producerBinding1); - binder.unbind(consumerBinding1); + producerBinding1.unbind(); + producerBinding2.unbind(); - binder.unbind(producerBinding2); - binder.unbind(consumerBinding2); + consumerBinding1.unbind(); + consumerBinding2.unbind(); } @Test @@ -171,21 +170,10 @@ public abstract class AbstractBinderTests { assertEquals("foo", inbound.getPayload()); assertNull(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)); assertNull(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE)); - binder.unbind(producerBinding); - binder.unbind(consumerBinding); + producerBinding.unbind(); + consumerBinding.unbind(); } - protected Collection getBindings(Binder testBinder) { - if (testBinder instanceof AbstractTestBinder) { - return getBindingsFromBinder(((AbstractTestBinder) testBinder).getCoreBinder()); - } - return Collections.EMPTY_LIST; - } - - protected Collection getBindingsFromBinder(Binder binder) { - DirectFieldAccessor accessor = new DirectFieldAccessor(binder); - return (List) accessor.getPropertyValue("bindings"); - } protected abstract Binder getBinder() throws Exception; diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractTestBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractTestBinder.java index a4cc37e69..b58b35a99 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractTestBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractTestBinder.java @@ -29,7 +29,7 @@ import org.springframework.messaging.MessageChannel; * @author Gary Russell * @author Mark Fisher */ -public abstract class AbstractTestBinder implements Binder { +public abstract class AbstractTestBinder implements Binder { protected Set queues = new HashSet(); @@ -63,11 +63,6 @@ public abstract class AbstractTestBinder public abstract void cleanup(); - @Override - public void unbind(Binding binding) { - binder.unbind(binding); - } - public C getBinder() { return this.binder; } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java index fcef3b5ca..d49c65255 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java @@ -28,7 +28,6 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; import java.util.Arrays; -import java.util.List; import java.util.Properties; import java.util.UUID; @@ -37,6 +36,7 @@ import org.hamcrest.Matcher; import org.hamcrest.Matchers; import org.junit.Test; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; @@ -80,7 +80,7 @@ abstract public class PartitionCapableBinderTests extends BrokerBinderTests { assertThat(receivedMessage2, not(nullValue())); assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload1)); - binder.unbind(binding2); + binding2.unbind(); String testPayload2 = "foo-" + UUID.randomUUID().toString(); output.send(new GenericMessage<>(testPayload2.getBytes())); @@ -100,9 +100,9 @@ abstract public class PartitionCapableBinderTests extends BrokerBinderTests { assertThat(receivedMessage2, not(nullValue())); assertThat(new String(receivedMessage2.getPayload()), equalTo(testPayload3)); - binder.unbind(producerBinding); - binder.unbind(binding1); - binder.unbind(binding2); + producerBinding.unbind(); + binding1.unbind(); + binding2.unbind(); } @Test @@ -163,11 +163,8 @@ abstract public class PartitionCapableBinderTests extends BrokerBinderTests { DirectChannel output = new DirectChannel(); output.setBeanName("test.output"); Binding outputBinding = binder.bindProducer("part.0", output, producerProperties); - @SuppressWarnings("unchecked") - List> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class); - assertEquals(4, bindings.size()); try { - AbstractEndpoint endpoint = bindings.get(3).getEndpoint(); + AbstractEndpoint endpoint = extractEndpoint(outputBinding); assertThat(getEndpointRouting(endpoint), containsString( getExpectedRoutingBaseDestination("part.0", "test") + "-' + headers['partition']")); } @@ -227,10 +224,10 @@ abstract public class PartitionCapableBinderTests extends BrokerBinderTests { containsOur3Messages); } - binder.unbind(input0Binding); - binder.unbind(input1Binding); - binder.unbind(input2Binding); - binder.unbind(outputBinding); + input0Binding.unbind(); + input1Binding.unbind(); + input2Binding.unbind(); + outputBinding.unbind(); } @Test @@ -261,11 +258,8 @@ abstract public class PartitionCapableBinderTests extends BrokerBinderTests { DirectChannel output = new DirectChannel(); output.setBeanName("test.output"); Binding outputBinding = binder.bindProducer("partJ.0", output, producerProperties); - @SuppressWarnings("unchecked") - List> bindings = TestUtils.getPropertyValue(binder, "binder.bindings", List.class); - assertEquals(4, bindings.size()); if (usesExplicitRouting()) { - AbstractEndpoint endpoint = bindings.get(3).getEndpoint(); + AbstractEndpoint endpoint = extractEndpoint(outputBinding); assertThat(getEndpointRouting(endpoint), containsString( getExpectedRoutingBaseDestination("partJ.0", "test") + "-' + headers['partition']")); } @@ -295,10 +289,10 @@ abstract public class PartitionCapableBinderTests extends BrokerBinderTests { containsInAnyOrder(0, 1, 2)); } - binder.unbind(input0Binding); - binder.unbind(input1Binding); - binder.unbind(input2Binding); - binder.unbind(outputBinding); + input0Binding.unbind(); + input1Binding.unbind(); + input2Binding.unbind(); + outputBinding.unbind(); } /** @@ -333,4 +327,8 @@ abstract public class PartitionCapableBinderTests extends BrokerBinderTests { protected abstract String getClassUnderTestName(); + protected AbstractEndpoint extractEndpoint(Binding binding) { + DirectFieldAccessor accessor = new DirectFieldAccessor(binding); + return (AbstractEndpoint) accessor.getPropertyValue("endpoint"); + } } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/test/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupportTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/test/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupportTests.java index a57075376..6967f7543 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/test/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupportTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/test/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupportTests.java @@ -25,12 +25,10 @@ import java.util.Collections; import java.util.List; import java.util.Properties; -import com.esotericsoftware.kryo.Kryo; -import com.esotericsoftware.kryo.Registration; import org.junit.Before; import org.junit.Test; -import org.springframework.cloud.stream.binder.MessageChannelBinderSupport.JavaClassMimeTypeConversion; +import org.springframework.cloud.stream.binder.AbstractBinder.JavaClassMimeTypeConversion; import org.springframework.cloud.stream.tuple.DefaultTuple; import org.springframework.cloud.stream.tuple.Tuple; import org.springframework.cloud.stream.tuple.TupleBuilder; @@ -46,6 +44,9 @@ import org.springframework.messaging.support.GenericMessage; import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; +import com.esotericsoftware.kryo.Kryo; +import com.esotericsoftware.kryo.Registration; + /** * @author Gary Russell * @author David Turanski @@ -265,7 +266,7 @@ public class MessageChannelBinderSupportTests { } - public class TestMessageChannelBinder extends MessageChannelBinderSupport { + public class TestMessageChannelBinder extends AbstractBinder { @Override protected Binding doBindConsumer(String name, String group, MessageChannel channel, Properties properties) { diff --git a/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/test/binder/TestSupportBinder.java b/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/test/binder/TestSupportBinder.java index ec0d38ff4..4d9b5aec8 100644 --- a/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/test/binder/TestSupportBinder.java +++ b/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/test/binder/TestSupportBinder.java @@ -53,7 +53,7 @@ public class TestSupportBinder implements Binder { @Override public Binding bindConsumer(String name, String group, MessageChannel inboundBindTarget, Properties properties) { - return null; + return new TestBinding(name, inboundBindTarget, messageCollector); } /** @@ -69,14 +69,7 @@ public class TestSupportBinder implements Binder { } }); this.messageChannels.put(name, outboundBindTarget); - return null; - } - - @Override - public void unbind(Binding binding) { - if (Binding.Type.producer.equals(binding.getType())) { - messageCollector.unregister(binding.getTarget()); - } + return new TestBinding(name, outboundBindTarget, messageCollector); } public MessageCollector messageCollector() { @@ -114,4 +107,26 @@ public class TestSupportBinder implements Binder { return queue; } } + + /** + * @author Marius Bogoevici + */ + public static class TestBinding implements Binding { + + private final MessageChannel target; + private final MessageCollectorImpl messageCollector; + + private String name; + + public TestBinding(String name, MessageChannel target, MessageCollectorImpl messageCollector) { + this.name = name; + this.target = target; + this.messageCollector = messageCollector; + } + + @Override + public void unbind() { + messageCollector.unregister(target); + } + } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupport.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java similarity index 82% rename from spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupport.java rename to spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java index 3eee0ac1d..350fcdb7a 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageChannelBinderSupport.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java @@ -16,7 +16,6 @@ package org.springframework.cloud.stream.binder; -import static org.springframework.util.MimeTypeUtils.ALL; import static org.springframework.util.MimeTypeUtils.APPLICATION_OCTET_STREAM; import static org.springframework.util.MimeTypeUtils.APPLICATION_OCTET_STREAM_VALUE; import static org.springframework.util.MimeTypeUtils.TEXT_PLAIN; @@ -25,12 +24,9 @@ import static org.springframework.util.MimeTypeUtils.TEXT_PLAIN_VALUE; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.UnsupportedEncodingException; -import java.util.ArrayList; import java.util.Arrays; -import java.util.Collections; import java.util.HashSet; import java.util.LinkedList; -import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.Properties; @@ -42,25 +38,18 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.BeansException; -import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; -import org.springframework.beans.factory.support.DefaultListableBeanFactory; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; -import org.springframework.context.Lifecycle; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.core.serializer.support.SerializationFailedException; import org.springframework.expression.EvaluationContext; import org.springframework.expression.Expression; -import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.codec.Codec; import org.springframework.integration.expression.ExpressionUtils; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.MessagingException; import org.springframework.retry.backoff.ExponentialBackOffPolicy; import org.springframework.retry.policy.SimpleRetryPolicy; import org.springframework.retry.support.RetryTemplate; @@ -76,9 +65,9 @@ import org.springframework.util.StringUtils; * @author Gary Russell * @author Ilayaperumal Gopinathan * @author Mark Fisher + * @author Marius Bogoevici */ -public abstract class MessageChannelBinderSupport - implements Binder, ApplicationContextAware, InitializingBean { +public abstract class AbstractBinder implements ApplicationContextAware, InitializingBean, Binder { protected static final String PARTITION_HEADER = "partition"; @@ -95,8 +84,6 @@ public abstract class MessageChannelBinderSupport private final StringConvertingContentTypeResolver contentTypeResolver = new StringConvertingContentTypeResolver(); - protected static final List MEDIATYPES_MEDIATYPE_ALL = Collections.singletonList(ALL); - private static final int DEFAULT_BACKOFF_INITIAL_INTERVAL = 1000; private static final int DEFAULT_BACKOFF_MAX_INTERVAL = 10000; @@ -156,27 +143,12 @@ public abstract class MessageChannelBinderSupport BinderPropertyKeys.BATCH_BUFFER_LIMIT, })); - private final List> bindings = Collections.synchronizedList(new ArrayList>()); - private final IdGenerator idGenerator = new AlternativeJdkIdGenerator(); protected volatile EvaluationContext evaluationContext; private volatile PartitionSelectorStrategy partitionSelector = new DefaultPartitionSelector(); - /** - * Used in the canonical case, when the binding does not involve an alias name. - */ - protected final SharedChannelProvider directChannelProvider = new - SharedChannelProvider( - DirectChannel.class) { - - @Override - protected DirectChannel createSharedChannel(String name) { - return new DirectChannel(); - } - }; - protected volatile long defaultBackOffInitialInterval = DEFAULT_BACKOFF_INITIAL_INTERVAL; protected volatile long defaultBackOffMaxInterval = DEFAULT_BACKOFF_MAX_INTERVAL; @@ -209,7 +181,7 @@ public abstract class MessageChannelBinderSupport /** * For binder implementations that support a prefix, apply the prefix to the name. * @param prefix the prefix. - * @param name the name. + * @param name the name. */ public static String applyPrefix(String prefix, String name) { return prefix + name; @@ -277,7 +249,8 @@ public abstract class MessageChannelBinderSupport } /** - * Set the default retry back off max interval for this binder; can be overridden with consumer 'backOffMaxInterval' + * Set the default retry back off max interval for this binder; can be overridden with consumer + * 'backOffMaxInterval' * property. * @param defaultBackOffMaxInterval */ @@ -362,7 +335,7 @@ public abstract class MessageChannelBinderSupport } @Override - public final Binding bindConsumer(String name, String group, MessageChannel inputChannel, Properties properties) { + public final Binding bindConsumer(String name, String group, T target, Properties properties) { DefaultBindingPropertiesAccessor accessor = new DefaultBindingPropertiesAccessor(properties); if (StringUtils.isEmpty(group)) { Assert.isTrue(!accessor.getProperty(BinderPropertyKeys.DURABLE, defaultDurableSubscription), @@ -370,75 +343,14 @@ public abstract class MessageChannelBinderSupport Assert.isTrue(accessor.getPartitionIndex() < 0, "A consumer group is required for a partitioned subscription"); } - return doBindConsumer(name, group, inputChannel, properties); + return doBindConsumer(name, group, target, properties); } - protected abstract Binding doBindConsumer(String name, String group, MessageChannel inputChannel, Properties properties); - - /** - * Create a producer for the named channel and bind it to the binder. Synchronized to avoid creating multiple - * instances. - * @param name The name. - * @param channelName The name of the channel to be created, and registered as bean. - * @param properties The properties. - * @return The channel. - */ - protected synchronized MessageChannel doBindDynamicProducer(String name, String channelName, - Properties properties) { - MessageChannel channel = this.directChannelProvider.lookupSharedChannel(channelName); - if (channel == null) { - try { - channel = this.directChannelProvider.createAndRegisterChannel(channelName); - bindProducer(name, channel, properties); - } - catch (RuntimeException e) { - destroyCreatedChannel(channelName, channel); - throw new BinderException( - "Failed to bind dynamic channel '" + name + "' with properties " + properties, e); - } - } - return channel; - } - - private void destroyCreatedChannel(String name, MessageChannel channel) { - BeanFactory beanFactory = this.applicationContext.getBeanFactory(); - if (beanFactory.containsBean(name)) { - if (beanFactory instanceof DefaultListableBeanFactory) { - ((DefaultListableBeanFactory) beanFactory).destroySingleton(name); - } - } - } - - @Override - public void unbind(Binding binding) { - binding.stop(); - this.bindings.remove(binding); - afterUnbind(binding); - } - - protected void afterUnbind(Binding binding) { - } - - protected void addBinding(Binding binding) { - this.bindings.add(binding); - } - - protected void stopBindings() { - for (Lifecycle bean : this.bindings) { - try { - bean.stop(); - } - catch (Exception e) { - if (this.logger.isWarnEnabled()) { - this.logger.warn("failed to stop adapter", e); - } - } - } - } + protected abstract Binding doBindConsumer(String name, String group, T inputTarget, Properties properties); /** * Construct a name comprised of the name and group. - * @param name the name. + * @param name the name. * @param group the group. * @return the constructed name. */ @@ -552,10 +464,11 @@ public abstract class MessageChannelBinderSupport * partition selector class is provided, it will be invoked to determine the partition. Otherwise, if the partition * expression is not null, it is evaluated against the key and is expected to return an integer to which the modulo * function will be applied, using the partitionCount as the divisor. If no partition expression is provided, the - * key will be passed to the binder partition strategy along with the partitionCount. The default partition strategy + * key will be passed to the binder partition strategy along with the partitionCount. The default partition + * strategy * uses {@code key.hashCode()}, and the result will be the mod of that value. * @param message the message. - * @param meta the partitioning metadata. + * @param meta the partitioning metadata. * @return the partition. */ protected int determinePartition(Message message, PartitioningMetadata meta) { @@ -638,12 +551,14 @@ public abstract class MessageChannelBinderSupport } /** - * Validate the provided deployment properties for the consumer against those supported by this binder implementation. - * The consumer is that part of the binder that consumes messages from the underlying infrastructure and sends them to + * Validate the provided deployment properties for the consumer against those supported by this binder + * implementation. + * The consumer is that part of the binder that consumes messages from the underlying infrastructure and sends them + * to * the next module. Consumer properties are used to configure the consumer. - * @param name The name. + * @param name The name. * @param properties The properties. - * @param supported The supported properties. + * @param supported The supported properties. */ protected void validateConsumerProperties(String name, Properties properties, Set supported) { if (properties != null) { @@ -652,12 +567,14 @@ public abstract class MessageChannelBinderSupport } /** - * Validate the provided deployment properties for the producer against those supported by this binder implementation. - * When a module sends a message to the binder, the producer uses these properties while sending it to the underlying + * Validate the provided deployment properties for the producer against those supported by this binder + * implementation. + * When a module sends a message to the binder, the producer uses these properties while sending it to the + * underlying * infrastructure. - * @param name The name. + * @param name The name. * @param properties The properties. - * @param supported The supported properties. + * @param supported The supported properties. */ protected void validateProducerProperties(String name, Properties properties, Set supported) { if (properties != null) { @@ -759,56 +676,6 @@ public abstract class MessageChannelBinderSupport } - /** - * Looks up or optionally creates a new channel to use. - * @author Eric Bottard - */ - protected abstract class SharedChannelProvider { - - private final Class requiredType; - - protected SharedChannelProvider(Class clazz) { - this.requiredType = clazz; - } - - public synchronized final T lookupOrCreateSharedChannel(String name) { - T channel = lookupSharedChannel(name); - if (channel == null) { - channel = createAndRegisterChannel(name); - } - return channel; - } - - @SuppressWarnings("unchecked") - public T createAndRegisterChannel(String name) { - T channel = createSharedChannel(name); - ConfigurableListableBeanFactory beanFactory = MessageChannelBinderSupport.this.applicationContext.getBeanFactory(); - beanFactory.registerSingleton(name, channel); - channel = (T) beanFactory.initializeBean(channel, name); - if (MessageChannelBinderSupport.this.logger.isDebugEnabled()) { - MessageChannelBinderSupport.this.logger.debug("Registered channel:" + name); - } - return channel; - } - - protected abstract T createSharedChannel(String name); - - public T lookupSharedChannel(String name) { - T channel = null; - if (MessageChannelBinderSupport.this.applicationContext.containsBean(name)) { - try { - channel = MessageChannelBinderSupport.this.applicationContext.getBean(name, this.requiredType); - } - catch (Exception e) { - throw new IllegalArgumentException("bean '" + name - + "' is already registered but does not match the required type"); - } - } - return channel; - } - - } - /** * Handles representing any java class as a {@link MimeType}. * @author David Turanski @@ -888,21 +755,6 @@ public abstract class MessageChannelBinderSupport } - public static class DirectHandler implements MessageHandler { - - private final MessageChannel outputChannel; - - public DirectHandler(MessageChannel outputChannel) { - this.outputChannel = outputChannel; - } - - @Override - public void handleMessage(Message message) throws MessagingException { - this.outputChannel.send(message); - } - - } - /** * Perform manual acknowledgement based on the metadata stored in the binder. */ diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binder.java index d49cf5848..7ded65508 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binder.java @@ -22,6 +22,7 @@ import java.util.Properties; * A strategy interface used to bind an app 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 @@ -50,10 +51,4 @@ public interface Binder { */ Binding bindProducer(String name, T outboundBindTarget, Properties properties); - /** - * Unbind the target component represented by the provided Binding and stop any active components. - * @param binding the Binding instance to unbind - */ - void unbind(Binding binding); - } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java index 90f469cb6..7b3faa4e6 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2016 the original author or authors. + * Copyright 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. @@ -16,10 +16,6 @@ package org.springframework.cloud.stream.binder; -import org.springframework.context.Lifecycle; -import org.springframework.integration.endpoint.AbstractEndpoint; -import org.springframework.util.Assert; - /** * Represents a binding between an input or output and an adapter endpoint that connects via a Binder. The binding * could be for a consumer or a producer. A consumer binding represents a connection from an adapter to an @@ -31,141 +27,11 @@ import org.springframework.util.Assert; * @author Marius Bogoevici * @see org.springframework.cloud.stream.annotation.EnableBinding */ -public class Binding implements Lifecycle { - - public enum Type { - producer, consumer - } - - private final String name; - - private final String group; - - private final T target; - - private final AbstractEndpoint endpoint; - - private final Type type; - - private final DefaultBindingPropertiesAccessor properties; - - private Binding(String name, String group, T target, AbstractEndpoint endpoint, Type type, - DefaultBindingPropertiesAccessor properties) { - Assert.notNull(target, "target must not be null"); - Assert.notNull(endpoint, "endpoint must not be null"); - this.name = name; - this.group = group; - this.target = target; - this.endpoint = endpoint; - this.type = type; - this.properties = properties; - } - - public static Binding forConsumer(String name, String group, AbstractEndpoint adapterFromBinder, T inputTarget, - DefaultBindingPropertiesAccessor properties) { - return new Binding<>(name, group, inputTarget, adapterFromBinder, Type.consumer, properties); - } - - public static Binding forProducer(String name, T outputTarget, AbstractEndpoint adapterToBinder, - DefaultBindingPropertiesAccessor properties) { - return new Binding<>(name, null, outputTarget, adapterToBinder, Type.producer, properties); - } - - public String getName() { - return name; - } - - public String getGroup() { - return group; - } - - public T getTarget() { - return target; - } - - public AbstractEndpoint getEndpoint() { - return endpoint; - } - - public Type getType() { - return type; - } - - public DefaultBindingPropertiesAccessor getPropertiesAccessor() { - return properties; - } - - @Override - public void start() { - endpoint.start(); - } - - @Override - public void stop() { - endpoint.stop(); - } - - @Override - public boolean isRunning() { - return endpoint.isRunning(); - } - - @Override - public String toString() { - return type + " Binding [name=" + name + ", target=" + target + ", endpoint=" + endpoint.getComponentName() - + "]"; - } - - @Override - public int hashCode() { - final int prime = 31; - int result = 1; - result = prime * result + ((endpoint == null) ? 0 : endpoint.hashCode()); - result = prime * result + ((group == null) ? 0 : group.hashCode()); - result = prime * result + ((name == null) ? 0 : name.hashCode()); - result = prime * result + ((properties == null) ? 0 : properties.hashCode()); - result = prime * result + ((target == null) ? 0 : target.hashCode()); - result = prime * result + ((type == null) ? 0 : type.hashCode()); - return result; - } - - @Override - public boolean equals(Object obj) { - if (this == obj) - return true; - if (obj == null) - return false; - if (getClass() != obj.getClass()) - return false; - Binding other = (Binding) obj; - if (endpoint == null) { - if (other.endpoint != null) - return false; - } else if (!endpoint.equals(other.endpoint)) - return false; - if (group == null) { - if (other.group != null) - return false; - } else if (!group.equals(other.group)) - return false; - if (name == null) { - if (other.name != null) - return false; - } else if (!name.equals(other.name)) - return false; - if (properties == null) { - if (other.properties != null) - return false; - } else if (!properties.equals(other.properties)) - return false; - if (target == null) { - if (other.target != null) - return false; - } else if (!target.equals(other.target)) - return false; - if (type != other.type) - return false; - return true; - } - +public interface Binding { + /** + * Unbinds the target component represented by this instance and stops any active components. Implementations must + * be idempotent. After this method is invoked, the target is not expected to receive any message, this instance + * should be discarded, and a new Binding should be created instead. + */ + void unbind(); } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java new file mode 100644 index 000000000..7ba7f359a --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java @@ -0,0 +1,83 @@ +/* + * Copyright 2013-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. + */ + +package org.springframework.cloud.stream.binder; + + +import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.util.Assert; + +/** + * Default implementation for a {@link Binding}. + * + * @author Jennifer Hickey + * @author Mark Fisher + * @author Gary Russell + * @author Marius Bogoevici + * @see org.springframework.cloud.stream.annotation.EnableBinding + */ +public class DefaultBinding implements Binding { + + private final String name; + + private final String group; + + private final T target; + + private final AbstractEndpoint endpoint; + + private final DefaultBindingPropertiesAccessor properties; + + public DefaultBinding(String name, String group, T target, AbstractEndpoint endpoint, + DefaultBindingPropertiesAccessor properties) { + Assert.notNull(target, "target must not be null"); + Assert.notNull(endpoint, "endpoint must not be null"); + this.name = name; + this.group = group; + this.target = target; + this.endpoint = endpoint; + this.properties = properties; + } + + + public String getName() { + return name; + } + + public String getGroup() { + return group; + } + + + @Override + public final void unbind() { + endpoint.stop(); + afterUnbind(); + } + + protected void afterUnbind() { + } + + public DefaultBindingPropertiesAccessor getPropertiesAccessor() { + return properties; + } + + @Override + public String toString() { + return " Binding [name=" + name + ", target=" + target + ", endpoint=" + endpoint.getComponentName() + + "]"; + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DirectHandler.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DirectHandler.java new file mode 100644 index 000000000..c45ade875 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DirectHandler.java @@ -0,0 +1,40 @@ +/* + * Copyright 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. + */ + +package org.springframework.cloud.stream.binder; + +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; + +/** + * @author Marius Bogoevici + */ +public class DirectHandler implements MessageHandler { + + private final MessageChannel outputChannel; + + public DirectHandler(MessageChannel outputChannel) { + this.outputChannel = outputChannel; + } + + @Override + public void handleMessage(Message message) throws MessagingException { + this.outputChannel.send(message); + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java index 5f80be48f..ba71d7c33 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java @@ -39,7 +39,7 @@ import org.springframework.cloud.stream.aggregate.SharedChannelRegistry; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.annotation.Input; import org.springframework.cloud.stream.annotation.Output; -import org.springframework.cloud.stream.binder.MessageChannelBinderSupport; +import org.springframework.cloud.stream.binder.DirectHandler; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; @@ -180,7 +180,7 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean binder = getBinderForChannel(inputChannelName); List> bindings = this.consumerBindings.remove(inputChannelName); if (bindings != null && !CollectionUtils.isEmpty(bindings)) { for (Binding binding : bindings) { - binder.unbind(binding); + binding.unbind(); } } else if (log.isWarnEnabled()) { @@ -104,10 +103,9 @@ public class ChannelBindingService { } public void unbindProducers(String outputChannelName) { - Binder binder = getBinderForChannel(outputChannelName); Binding binding = this.producerBindings.remove(outputChannelName); if (binding != null) { - binder.unbind(binding); + binding.unbind(); } else if (log.isWarnEnabled()) { log.warn("Trying to unbind channel '" + outputChannelName + "', but no binding found."); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java index 0d4d401f6..3ddee39eb 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java @@ -32,6 +32,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -41,13 +42,11 @@ import org.mockito.Mockito; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.support.DefaultListableBeanFactory; -import org.springframework.cloud.stream.binder.local.LocalMessageChannelBinder; import org.springframework.cloud.stream.binding.BinderAwareChannelResolver; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.ChannelBindingServiceProperties; import org.springframework.context.support.StaticApplicationContext; import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.utils.IntegrationUtils; @@ -55,8 +54,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; -import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; -import org.springframework.scheduling.support.PeriodicTrigger; +import org.springframework.messaging.SubscribableChannel; /** * @author Mark Fisher @@ -68,13 +66,11 @@ public class BinderAwareChannelResolverTests { private volatile BinderAwareChannelResolver resolver; - private volatile LocalMessageChannelBinder binder; + private volatile Binder binder; @Before public void setupContext() throws Exception { - this.binder = new LocalMessageChannelBinder(); - this.binder.setApplicationContext(context); - this.binder.afterPropertiesSet(); + this.binder = new TestBinder(); this.resolver = new BinderAwareChannelResolver(new BinderFactory() { @Override public Binder getBinder(String configurationName) { @@ -85,14 +81,9 @@ public class BinderAwareChannelResolverTests { context.getBeanFactory().registerSingleton("channelResolver", this.resolver); context.registerSingleton("other", DirectChannel.class); - context.registerSingleton("taskScheduler", ThreadPoolTaskScheduler.class); context.registerSingleton(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME, DefaultMessageBuilderFactory.class); context.refresh(); - - PollerMetadata poller = new PollerMetadata(); - poller.setTrigger(new PeriodicTrigger(1000)); - binder.setPoller(poller); } @Test @@ -156,4 +147,58 @@ public class BinderAwareChannelResolverTests { assertSame(resolved, beanFactory.getBean("someTransport:foo")); } + /** + * A simple test binder that creates queues for the destinations. Ignores groups. + */ + private class TestBinder implements Binder { + + private final Map destinations = new ConcurrentHashMap<>(); + + @Override + public Binding bindConsumer(String name, String group, MessageChannel inboundBindTarget, + Properties properties) { + synchronized (destinations) { + if (!destinations.containsKey(name)) { + destinations.put(name, new DirectChannel()); + } + } + DirectHandler directHandler = new DirectHandler(inboundBindTarget); + destinations.get(name).subscribe(directHandler); + return new TestBinding(inboundBindTarget, name, directHandler); + } + + + @Override + public Binding bindProducer(String name, MessageChannel outboundBindTarget, Properties properties) { + synchronized (destinations) { + if (!destinations.containsKey(name)) { + destinations.put(name, new DirectChannel()); + } + } + DirectHandler directHandler = new DirectHandler(destinations.get(name)); + // for test purposes we can assume it is a SubscribableChannel + ((SubscribableChannel) outboundBindTarget).subscribe(directHandler); + return new TestBinding(outboundBindTarget, name, directHandler); + } + + private class TestBinding implements Binding { + + private final MessageChannel target; + + private final String name; + + private final DirectHandler directHandler; + + public TestBinding(MessageChannel outboundChannel, String name, DirectHandler directHandler) { + this.target = outboundChannel; + this.name = name; + this.directHandler = directHandler; + } + + @Override + public void unbind() { + destinations.get(name).unsubscribe(directHandler); + } + } + } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/local/LocalMessageChannelBinder.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/local/LocalMessageChannelBinder.java deleted file mode 100644 index e75acdf25..000000000 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/local/LocalMessageChannelBinder.java +++ /dev/null @@ -1,259 +0,0 @@ -/* - * Copyright 2013-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. - */ - -package org.springframework.cloud.stream.binder.local; - -import java.util.Collection; -import java.util.Properties; - -import org.springframework.cloud.stream.binder.DefaultBindingPropertiesAccessor; -import org.springframework.cloud.stream.binder.Binding; -import org.springframework.cloud.stream.binder.MessageChannelBinderSupport; -import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.channel.PublishSubscribeChannel; -import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.config.ConsumerEndpointFactoryBean; -import org.springframework.integration.handler.BridgeHandler; -import org.springframework.integration.scheduling.PollerMetadata; -import org.springframework.integration.support.context.NamedComponent; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.PollableChannel; -import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; -import org.springframework.util.Assert; -import org.springframework.util.MimeType; - -/** - * A simple implementation of {@link org.springframework.cloud.stream.binder.Binder} for in-process use. For inbound and outbound, creates a - * {@link DirectChannel} or a {@link QueueChannel} depending on whether the binding is aliased or not then bridges the - * passed {@link MessageChannel} to the channel which is registered in the given application context. If that channel - * does not yet exist, it will be created. - * - * @author David Turanski - * @author Mark Fisher - * @author Gary Russell - * @author Jennifer Hickey - * @author Ilayaperumal Gopinathan - * @since 1.0 - */ -public class LocalMessageChannelBinder extends MessageChannelBinderSupport { - - public static final String THREAD_NAME_PREFIX = "binder.local-"; - - private static final int DEFAULT_EXECUTOR_CORE_POOL_SIZE = 0; - - private static final int DEFAULT_EXECUTOR_MAX_POOL_SIZE = 200; - - private static final int DEFAULT_EXECUTOR_QUEUE_SIZE = Integer.MAX_VALUE; - - private static final int DEFAULT_EXECUTOR_KEEPALIVE_SECONDS = 60; - - private volatile PollerMetadata poller; - - private final ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); - - private volatile int executorCorePoolSize = DEFAULT_EXECUTOR_CORE_POOL_SIZE; - - private volatile int executorMaxPoolSize = DEFAULT_EXECUTOR_MAX_POOL_SIZE; - - private volatile int executorQueueSize = DEFAULT_EXECUTOR_QUEUE_SIZE; - - private volatile int executorKeepAliveSeconds = DEFAULT_EXECUTOR_KEEPALIVE_SECONDS; - - private final SharedChannelProvider pubsubChannelProvider = new SharedChannelProvider( - PublishSubscribeChannel.class) { - - @Override - protected PublishSubscribeChannel createSharedChannel(String name) { - PublishSubscribeChannel publishSubscribeChannel = new PublishSubscribeChannel(executor); - publishSubscribeChannel.setIgnoreFailures(true); - return publishSubscribeChannel; - } - }; - - /** - * Set the poller to use when QueueChannels are used. - */ - public void setPoller(PollerMetadata poller) { - this.poller = poller; - } - - /** - * Set the {@link ThreadPoolTaskExecutor}} core pool size to limit the number of concurrent - * threads. The executor is used for PubSub operations. - * Default: 0 (threads created on demand until maxPoolSize). - * @param executorCorePoolSize the pool size. - */ - public void setExecutorCorePoolSize(int executorCorePoolSize) { - this.executorCorePoolSize = executorCorePoolSize; - } - - /** - * Set the {@link ThreadPoolTaskExecutor}} max pool size to limit the number of concurrent - * threads. The executor is used for PubSub operations. - * Default: 200. - * @param executorMaxPoolSize the pool size. - */ - public void setExecutorMaxPoolSize(int executorMaxPoolSize) { - this.executorMaxPoolSize = executorMaxPoolSize; - } - - /** - * Set the {@link ThreadPoolTaskExecutor}} queue size to limit the number of concurrent - * threads. The executor is used for PubSub operations. - * Default: {@link Integer#MAX_VALUE}. - * @param executorQueueSize the queue size. - */ - public void setExecutorQueueSize(int executorQueueSize) { - this.executorQueueSize = executorQueueSize; - } - - /** - * Set the {@link ThreadPoolTaskExecutor}} keep alive seconds. - * The executor is used for PubSub operations. - * @param executorKeepAliveSeconds the keep alive seconds. - */ - public void setExecutorKeepAliveSeconds(int executorKeepAliveSeconds) { - this.executorKeepAliveSeconds = executorKeepAliveSeconds; - } - - @Override - public void afterPropertiesSet() throws Exception { - super.afterPropertiesSet(); - this.executor.setCorePoolSize(this.executorCorePoolSize); - this.executor.setMaxPoolSize(this.executorMaxPoolSize); - this.executor.setQueueCapacity(this.executorQueueSize); - this.executor.setKeepAliveSeconds(this.executorKeepAliveSeconds); - this.executor.setThreadNamePrefix(THREAD_NAME_PREFIX); - this.executor.initialize(); - } - - @Override - protected Binding doBindConsumer(String name, String group, MessageChannel moduleInputChannel, - Properties properties) { - validateConsumerProperties(name, properties, CONSUMER_STANDARD_PROPERTIES); - return doRegisterConsumer(name, moduleInputChannel, this.pubsubChannelProvider, properties); - } - - private Binding doRegisterConsumer(String name, MessageChannel moduleInputChannel, - SharedChannelProvider channelProvider, Properties properties) { - Assert.hasText(name, "a valid name is required to register an inbound channel"); - Assert.notNull(moduleInputChannel, "channel must not be null"); - MessageChannel registeredChannel = channelProvider.lookupOrCreateSharedChannel("localbinder." + name); - bridge(name, registeredChannel, moduleInputChannel, - "inbound." + ((NamedComponent) registeredChannel).getComponentName(), - new LocalBindingPropertiesAccessor(properties)); - // TODO: ? - return null; - } - - /** - * Looks up or creates a DirectChannel with the given name and creates a bridge to that channel from the provided - * channel instance. - */ - @Override - public Binding bindProducer(String name, MessageChannel moduleOutputChannel, Properties properties) { - validateConsumerProperties(name, properties, PRODUCER_STANDARD_PROPERTIES); - return doRegisterProducer(name, moduleOutputChannel, this.pubsubChannelProvider, properties); - } - - private Binding doRegisterProducer(String name, MessageChannel moduleOutputChannel, - SharedChannelProvider channelProvider, Properties properties) { - Assert.hasText(name, "a valid name is required to register an outbound channel"); - Assert.notNull(moduleOutputChannel, "channel must not be null"); - MessageChannel registeredChannel = channelProvider.lookupOrCreateSharedChannel("localbinder." + name); - bridge(name, moduleOutputChannel, registeredChannel, - "outbound." + ((NamedComponent) registeredChannel).getComponentName(), - new LocalBindingPropertiesAccessor(properties)); - // TODO: ? - return null; - } - - @Override - public void unbind(Binding binding) { - } - - protected BridgeHandler bridge(String name, MessageChannel from, MessageChannel to, String bridgeName, - LocalBindingPropertiesAccessor properties) { - return bridge(name, from, to, bridgeName, null, properties); - } - - - protected BridgeHandler bridge(String name, MessageChannel from, MessageChannel to, String bridgeName, - final Collection acceptedMimeTypes, LocalBindingPropertiesAccessor properties) { - - final boolean isInbound = bridgeName.startsWith("inbound."); - - BridgeHandler handler = new BridgeHandler() { - - @Override - protected boolean shouldCopyRequestHeaders() { - return false; - } - - @Override - protected Object handleRequestMessage(Message requestMessage) { - return requestMessage; - } - - }; - - handler.setBeanFactory(getBeanFactory()); - handler.setOutputChannel(to); - handler.setBeanName(bridgeName); - handler.afterPropertiesSet(); - - // Usage of a CEFB allows to handle both Subscribable & Pollable channels the same way - ConsumerEndpointFactoryBean cefb = new ConsumerEndpointFactoryBean(); - cefb.setInputChannel(from); - cefb.setHandler(handler); - cefb.setBeanFactory(getBeanFactory()); - if (from instanceof PollableChannel) { - cefb.setPollerMetadata(poller); - } - try { - cefb.afterPropertiesSet(); - } - catch (Exception e) { - throw new IllegalStateException(e); - } - - try { - cefb.getObject().setComponentName(handler.getComponentName()); - Binding binding = isInbound ? Binding.forConsumer(name, null, cefb.getObject(), to, properties) - : Binding.forProducer(name, from, cefb.getObject(), properties); - addBinding(binding); - binding.start(); - } - catch (Exception e) { - throw new IllegalStateException(e); - } - return handler; - } - - protected T getBean(String name, Class requiredType) { - return getApplicationContext().getBean(name, requiredType); - } - - private static class LocalBindingPropertiesAccessor extends DefaultBindingPropertiesAccessor { - - public LocalBindingPropertiesAccessor(Properties properties) { - super(properties); - } - - } - -} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub1/StubBinder1.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub1/StubBinder1.java index 8089684c2..28cf61749 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub1/StubBinder1.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub1/StubBinder1.java @@ -47,8 +47,4 @@ public class StubBinder1 implements Binder { return null; } - @Override - public void unbind(Binding binding) { - } - } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2.java index a4cf2e307..13186a3c8 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2.java @@ -44,8 +44,4 @@ public class StubBinder2 implements Binder { return null; } - @Override - public void unbind(Binding binding) { - } - } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2ConfigurationA.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2ConfigurationA.java index 1043ee9d2..40cf4f7e0 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2ConfigurationA.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2ConfigurationA.java @@ -17,7 +17,6 @@ package org.springframework.cloud.stream.binder.stub2; import org.springframework.cloud.stream.binder.Binder; -import org.springframework.cloud.stream.binder.stub2.StubBinder2; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java index ea5896975..1c927fc38 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java @@ -54,7 +54,6 @@ import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.ChannelBindingServiceProperties; import org.springframework.cloud.stream.utils.MockBinderConfiguration; import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.core.DestinationResolutionException; @@ -81,8 +80,8 @@ public class ChannelBindingServiceTests { Binder binder = binderFactory.getBinder("mock"); ChannelBindingService service = new ChannelBindingService(properties, binderFactory); MessageChannel inputChannel = new DirectChannel(); - Binding mockBinding = Binding.forConsumer("foo", null, Mockito.mock(AbstractEndpoint.class), - inputChannel, null); + @SuppressWarnings("unchecked") + Binding mockBinding = Mockito.mock(Binding.class); when(binder.bindConsumer("foo", null, inputChannel, new Properties())) .thenReturn(mockBinding); Collection> bindings = service.bindConsumer(inputChannel, inputChannelName); @@ -91,7 +90,7 @@ public class ChannelBindingServiceTests { assertThat(binding, sameInstance(mockBinding)); service.unbindConsumers(inputChannelName); verify(binder).bindConsumer("foo", props.getGroup(), inputChannel, properties.getConsumerProperties(inputChannelName)); - verify(binder).unbind(binding); + verify(binding).unbind(); binderFactory.destroy(); } @@ -116,10 +115,10 @@ public class ChannelBindingServiceTests { ChannelBindingService service = new ChannelBindingService(properties, binderFactory); MessageChannel inputChannel = new DirectChannel(); - Binding mockBinding1 = Binding.forConsumer("foo", null, Mockito.mock(AbstractEndpoint.class), - inputChannel, null); - Binding mockBinding2 = Binding.forConsumer("bar", null, Mockito.mock(AbstractEndpoint.class), - inputChannel, null); + @SuppressWarnings("unchecked") + Binding mockBinding1 = Mockito.mock(Binding.class); + @SuppressWarnings("unchecked") + Binding mockBinding2 = Mockito.mock(Binding.class); when(binder.bindConsumer("foo", null, inputChannel, new Properties())) .thenReturn(mockBinding1); @@ -140,8 +139,8 @@ public class ChannelBindingServiceTests { verify(binder).bindConsumer("foo", props.getGroup(), inputChannel, properties.getConsumerProperties(inputChannelName)); verify(binder).bindConsumer("bar", props.getGroup(), inputChannel, properties.getConsumerProperties(inputChannelName)); - verify(binder).unbind(binding1); - verify(binder).unbind(binding2); + verify(binding1).unbind(); + verify(binding2).unbind(); binderFactory.destroy(); } @@ -163,8 +162,8 @@ public class ChannelBindingServiceTests { Binder binder = binderFactory.getBinder("mock"); ChannelBindingService service = new ChannelBindingService(properties, binderFactory); MessageChannel inputChannel = new DirectChannel(); - Binding mockBinding = Binding.forConsumer("foo", "fooGroup", Mockito.mock(AbstractEndpoint.class), - inputChannel, null); + @SuppressWarnings("unchecked") + Binding mockBinding = Mockito.mock(Binding.class); when(binder.bindConsumer("foo", "fooGroup", inputChannel, new Properties())) .thenReturn(mockBinding); Collection> bindings = service.bindConsumer(inputChannel, inputChannelName); @@ -174,7 +173,7 @@ public class ChannelBindingServiceTests { service.unbindConsumers(inputChannelName); verify(binder).bindConsumer("foo", props.getGroup(), inputChannel, properties.getConsumerProperties(inputChannelName)); - verify(binder).unbind(binding); + verify(binding).unbind(); binderFactory.destroy(); } @@ -189,9 +188,9 @@ public class ChannelBindingServiceTests { Binder binder = binderFactory.getBinder("mock"); MessageChannel inputChannel = new DirectChannel(); - Binding mockBinding = Binding.forConsumer("bar", null, Mockito.mock(AbstractEndpoint.class), - inputChannel, null); - + @SuppressWarnings("unchecked") + Binding mockBinding = Mockito.mock(Binding.class); + @SuppressWarnings("unchecked") final AtomicReference dynamic = new AtomicReference<>(); when(binder.bindProducer( matches("mock:bar"), any(DirectChannel.class), any(Properties.class))).thenReturn(mockBinding); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderRegistryConfiguration.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderRegistryConfiguration.java index 4b41fed85..33cdde48d 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderRegistryConfiguration.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderRegistryConfiguration.java @@ -17,14 +17,11 @@ package org.springframework.cloud.stream.utils; import java.util.Collections; -import java.util.Properties; import org.springframework.cloud.stream.binder.Binder; -import org.springframework.cloud.stream.binder.BinderConfiguration; import org.springframework.cloud.stream.binder.BinderFactory; import org.springframework.cloud.stream.binder.BinderType; import org.springframework.cloud.stream.binder.BinderTypeRegistry; -import org.springframework.cloud.stream.binder.DefaultBinderFactory; import org.springframework.cloud.stream.binder.DefaultBinderTypeRegistry; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration;