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 477fd44f4..99e4f308c 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 @@ -1,5 +1,5 @@ /* - * Copyright 2013-2016 the original author or authors. + * Copyright 2013-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -20,7 +20,6 @@ import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -28,25 +27,19 @@ import org.junit.Before; import org.junit.Test; import org.mockito.Mockito; -import org.springframework.beans.factory.BeanFactory; -import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; -import org.springframework.beans.factory.support.DefaultListableBeanFactory; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.stream.binder.integration.SpringIntegrationBinderConfiguration; import org.springframework.cloud.stream.binding.AbstractBindingTargetFactory; +import org.springframework.cloud.stream.binding.Bindable; import org.springframework.cloud.stream.binding.BinderAwareChannelResolver; import org.springframework.cloud.stream.binding.BindingService; import org.springframework.cloud.stream.binding.DynamicDestinationsBindable; -import org.springframework.cloud.stream.binding.InputBindingLifecycle; -import org.springframework.cloud.stream.binding.MessageConverterConfigurer; -import org.springframework.cloud.stream.binding.OutputBindingLifecycle; -import org.springframework.cloud.stream.binding.SubscribableChannelBindingTargetFactory; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; -import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; -import org.springframework.context.support.StaticApplicationContext; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilder; -import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -54,10 +47,11 @@ import org.springframework.messaging.MessagingException; import org.springframework.messaging.SubscribableChannel; import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.fail; -import static org.mockito.Matchers.any; -import static org.mockito.Matchers.eq; -import static org.mockito.Matchers.matches; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.matches; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -66,10 +60,11 @@ import static org.mockito.Mockito.when; * @author Mark Fisher * @author Gary Russell * @author Ilayaperumal Gopinathan + * @author Oleg Zhurakousky */ public class BinderAwareChannelResolverTests { - protected final StaticApplicationContext context = new StaticApplicationContext(); + protected ConfigurableApplicationContext context; protected volatile BinderAwareChannelResolver resolver; @@ -81,68 +76,43 @@ public class BinderAwareChannelResolverTests { protected volatile DynamicDestinationsBindable dynamicDestinationsBindable; - private volatile List producerBindings; - @Before public void setupContext() throws Exception { - producerBindings = new ArrayList<>(); - this.binder = new TestBinder(); - BinderFactory binderFactory = new BinderFactory() { - - @Override - public Binder getBinder( - String configurationName, Class bindableType) { - return (Binder) binder; - } - }; - this.bindingServiceProperties = new BindingServiceProperties(); - Map bindings = new HashMap<>(); - BindingProperties bindingProperties = new BindingProperties(); - bindingProperties.setContentType("text/plain"); - bindings.put("foo", bindingProperties); - this.bindingServiceProperties.setBindings(bindings); - BindingService bindingService = new BindingService(bindingServiceProperties, - binderFactory); - MessageConverterConfigurer messageConverterConfigurer = new MessageConverterConfigurer( - this.bindingServiceProperties, - new CompositeMessageConverterFactory()); - messageConverterConfigurer.setBeanFactory(Mockito.mock(ConfigurableListableBeanFactory.class)); - messageConverterConfigurer.afterPropertiesSet(); - this.bindingTargetFactory = new SubscribableChannelBindingTargetFactory(messageConverterConfigurer); - dynamicDestinationsBindable = new DynamicDestinationsBindable(); - this.resolver = new BinderAwareChannelResolver(bindingService, this.bindingTargetFactory, - dynamicDestinationsBindable); - this.resolver.setBeanFactory(context.getBeanFactory()); - context.getBeanFactory().registerSingleton("channelResolver", this.resolver); - context.getBeanFactory().registerSingleton("dynamicDestinationBindable", this.dynamicDestinationsBindable); - context.registerSingleton("other", DirectChannel.class); - context.registerSingleton(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME, - DefaultMessageBuilderFactory.class); - context.getBeanFactory().registerSingleton("bindingService", bindingService); - context.registerSingleton("inputBindingLifecycle", InputBindingLifecycle.class); - context.registerSingleton("outputBindingLifecycle", OutputBindingLifecycle.class); - context.refresh(); + this.context = new SpringApplicationBuilder(SpringIntegrationBinderConfiguration.getCompleteConfiguration()).web(WebApplicationType.NONE).run(); + this.resolver = context.getBean(BinderAwareChannelResolver.class); + this.binder = context.getBean(Binder.class); + this.bindingServiceProperties = context.getBean(BindingServiceProperties.class); + this.bindingTargetFactory = context.getBean(AbstractBindingTargetFactory.class); } @Test public void resolveChannel() { - assertThat(producerBindings).hasSize(0); + Map bindables = context.getBeansOfType(Bindable.class); + assertThat(bindables).hasSize(1); + for (Bindable bindable : bindables.values()) { + assertEquals(0, bindable.getInputs().size()); // producer + assertEquals(0, bindable.getOutputs().size());// consumer + } MessageChannel registered = resolver.resolveDestination("foo"); - assertThat(producerBindings).hasSize(1); - TestBinder.TestBinding binding = producerBindings.get(0); - assertThat(binding.isBound()).describedAs("Must be bound"); + + bindables = context.getBeansOfType(Bindable.class); + assertThat(bindables).hasSize(1); + for (Bindable bindable : bindables.values()) { + assertEquals(0, bindable.getInputs().size()); // producer + assertEquals(1, bindable.getOutputs().size());// consumer + } DirectChannel testChannel = new DirectChannel(); + testChannel.setComponentName("INPUT"); final CountDownLatch latch = new CountDownLatch(1); final List> received = new ArrayList<>(); testChannel.subscribe(new MessageHandler() { - @Override public void handleMessage(Message message) throws MessagingException { received.add(message); latch.countDown(); } }); - binder.bindConsumer("foo", null, testChannel, new ConsumerProperties()); + this.binder.bindConsumer("foo", null, testChannel, new ConsumerProperties()); assertThat(received).hasSize(0); registered.send(MessageBuilder.withPayload("hello").build()); try { @@ -154,15 +124,18 @@ public class BinderAwareChannelResolverTests { } assertThat(received).hasSize(1); assertThat(new String((byte[])received.get(0).getPayload())).isEqualTo("hello"); - context.close(); - assertThat(producerBindings).hasSize(1); - assertThat(binding.isBound()).isFalse().describedAs("Must not be bound"); + this.context.close(); + for (Bindable bindable : bindables.values()) { + assertEquals(0, bindable.getInputs().size()); + assertEquals(0, bindable.getOutputs().size());//Must not be bound" + } } @Test public void resolveNonRegisteredChannel() { MessageChannel other = resolver.resolveDestination("other"); assertThat(context.getBean("other")).isSameAs(other); + this.context.close(); } @Test @@ -173,7 +146,6 @@ public class BinderAwareChannelResolverTests { genericProperties.setContentType("text/plain"); bindings.put("foo", genericProperties); this.bindingServiceProperties.setBindings(bindings); - @SuppressWarnings("unchecked") Binder binder = mock(Binder.class); Binder binder2 = mock(Binder.class); BinderFactory mockBinderFactory = Mockito.mock(BinderFactory.class); @@ -187,74 +159,12 @@ public class BinderAwareChannelResolverTests { when(mockBinderFactory.getBinder("someTransport", DirectChannel.class)).thenReturn(binder2); BindingService bindingService = new BindingService(bindingServiceProperties, mockBinderFactory); - @SuppressWarnings("unchecked") BinderAwareChannelResolver resolver = new BinderAwareChannelResolver(bindingService, this.bindingTargetFactory, new DynamicDestinationsBindable()); - BeanFactory beanFactory = new DefaultListableBeanFactory(); - resolver.setBeanFactory(beanFactory); + resolver.setBeanFactory(context.getBeanFactory()); SubscribableChannel resolved = (SubscribableChannel) resolver.resolveDestination("foo"); verify(binder).bindProducer(eq("foo"), any(MessageChannel.class), any(ProducerProperties.class)); - assertThat(resolved).isSameAs(beanFactory.getBean("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, ConsumerProperties 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(name, directHandler); - } - - @Override - public Binding bindProducer(String name, - MessageChannel outboundBindTarget, ProducerProperties 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); - TestBinding binding = new TestBinding(name, directHandler); - producerBindings.add(binding); - return binding; - } - - private final class TestBinding implements Binding { - - private final String name; - - private final DirectHandler directHandler; - - private boolean bound = true; - - private TestBinding(String name, DirectHandler directHandler) { - this.name = name; - this.directHandler = directHandler; - } - - @Override - public void unbind() { - bound = false; - destinations.get(name).unsubscribe(directHandler); - } - - public boolean isBound() { - return bound; - } - } + assertThat(resolved).isSameAs(context.getBean("foo")); + this.context.close(); } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ExtendedPropertiesBinderAwareChannelResolverTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ExtendedPropertiesBinderAwareChannelResolverTests.java index f04e17987..fad06ca66 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ExtendedPropertiesBinderAwareChannelResolverTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ExtendedPropertiesBinderAwareChannelResolverTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2016 the original author or authors. + * Copyright 2013-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,102 +17,49 @@ package org.springframework.cloud.stream.binder; import java.util.ArrayList; -import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import org.junit.Before; import org.junit.Test; -import org.mockito.Mockito; -import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; -import org.springframework.cloud.stream.binding.BinderAwareChannelResolver; -import org.springframework.cloud.stream.binding.BindingService; -import org.springframework.cloud.stream.binding.DynamicDestinationsBindable; -import org.springframework.cloud.stream.binding.InputBindingLifecycle; -import org.springframework.cloud.stream.binding.MessageConverterConfigurer; -import org.springframework.cloud.stream.binding.OutputBindingLifecycle; -import org.springframework.cloud.stream.binding.SubscribableChannelBindingTargetFactory; -import org.springframework.cloud.stream.config.BindingProperties; -import org.springframework.cloud.stream.config.BindingServiceProperties; -import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; +import org.springframework.cloud.stream.binding.Bindable; import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilder; -import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; -import org.springframework.messaging.SubscribableChannel; import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.fail; /** * @author Mark Fisher * @author Gary Russell * @author Ilayaperumal Gopinathan + * @author Oleg Zhurakousky */ public class ExtendedPropertiesBinderAwareChannelResolverTests extends BinderAwareChannelResolverTests { - private volatile ExtendedPropertiesBinder binder; - - private volatile List producerBindings; - - @Before - @Override - public void setupContext() throws Exception { - producerBindings = new ArrayList<>(); - this.binder = new TestBinder(); - BinderFactory binderFactory = new BinderFactory() { - - @Override - public Binder getBinder( - String configurationName, Class bindableType) { - return (Binder) binder; - } - }; - this.bindingServiceProperties = new BindingServiceProperties(); - Map bindings = new HashMap(); - BindingProperties bindingProperties = new BindingProperties(); - bindingProperties.setContentType("text/plain"); - bindings.put("foo", bindingProperties); - this.bindingServiceProperties.setBindings(bindings); - MessageConverterConfigurer messageConverterConfigurer = new MessageConverterConfigurer( - this.bindingServiceProperties, - new CompositeMessageConverterFactory()); - messageConverterConfigurer.setBeanFactory(Mockito.mock(ConfigurableListableBeanFactory.class)); - messageConverterConfigurer.afterPropertiesSet(); - this.bindingTargetFactory = new SubscribableChannelBindingTargetFactory(messageConverterConfigurer); - dynamicDestinationsBindable = new DynamicDestinationsBindable(); - BindingService bindingService = new BindingService(bindingServiceProperties, - binderFactory); - this.resolver = new BinderAwareChannelResolver(bindingService, this.bindingTargetFactory, - dynamicDestinationsBindable); - this.resolver.setBeanFactory(context.getBeanFactory()); - context.getBeanFactory().registerSingleton("channelResolver", this.resolver); - context.getBeanFactory().registerSingleton("dynamicDestinationBindable", this.dynamicDestinationsBindable); - context.registerSingleton("other", DirectChannel.class); - context.registerSingleton(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME, - DefaultMessageBuilderFactory.class); - context.getBeanFactory().registerSingleton("bindingService", bindingService); - context.registerSingleton("inputBindingLifecycle", InputBindingLifecycle.class); - context.registerSingleton("outputBindingLifecycle", OutputBindingLifecycle.class); - context.refresh(); - } - @Test @Override public void resolveChannel() { - assertThat(producerBindings).hasSize(0); + Map bindables = context.getBeansOfType(Bindable.class); + assertThat(bindables).hasSize(1); + for (Bindable bindable : bindables.values()) { + assertEquals(0, bindable.getInputs().size()); // producer + assertEquals(0, bindable.getOutputs().size());// consumer + } MessageChannel registered = resolver.resolveDestination("foo"); - assertThat(producerBindings).hasSize(1); - TestBinder.TestBinding binding = producerBindings.get(0); - assertThat(binding.isBound()).describedAs("Must be bound"); + bindables = context.getBeansOfType(Bindable.class); + assertThat(bindables).hasSize(1); + for (Bindable bindable : bindables.values()) { + assertEquals(0, bindable.getInputs().size()); // producer + assertEquals(1, bindable.getOutputs().size());// consumer + } DirectChannel testChannel = new DirectChannel(); final CountDownLatch latch = new CountDownLatch(1); final List> received = new ArrayList<>(); @@ -124,7 +71,7 @@ public class ExtendedPropertiesBinderAwareChannelResolverTests extends BinderAwa latch.countDown(); } }); - binder.bindConsumer("foo", null, testChannel, new ExtendedConsumerProperties(new ConsumerProperties())); + binder.bindConsumer("foo", null, testChannel, new ExtendedConsumerProperties(new ConsumerProperties())); assertThat(received).hasSize(0); registered.send(MessageBuilder.withPayload("hello").build()); try { @@ -137,79 +84,9 @@ public class ExtendedPropertiesBinderAwareChannelResolverTests extends BinderAwa assertThat(received).hasSize(1); assertThat(new String((byte[])received.get(0).getPayload())).isEqualTo("hello"); context.close(); - assertThat(producerBindings).hasSize(1); - assertThat(binding.isBound()).isFalse().describedAs("Must not be bound"); - } - - /** - * A simple test binder that creates queues for the destinations. Ignores groups. - */ - class TestBinder implements - ExtendedPropertiesBinder { - - private final Map destinations = new ConcurrentHashMap<>(); - - @Override - public Binding bindConsumer(String name, String group, - MessageChannel inboundBindTarget, ExtendedConsumerProperties 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(name, directHandler); - } - - @Override - public Binding bindProducer(String name, - MessageChannel outboundBindTarget, ExtendedProducerProperties 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); - TestBinding binding = new TestBinding(name, directHandler); - producerBindings.add(binding); - return binding; - } - - @Override - public ExtendedConsumerProperties getExtendedConsumerProperties(String channelName) { - return new ExtendedConsumerProperties(new ConsumerProperties()); - } - - @Override - public ExtendedProducerProperties getExtendedProducerProperties(String channelName) { - return new ExtendedProducerProperties(new ProducerProperties()); - } - - private final class TestBinding implements Binding { - - private final String name; - - private final DirectHandler directHandler; - - private boolean bound = true; - - private TestBinding(String name, DirectHandler directHandler) { - this.name = name; - this.directHandler = directHandler; - } - - @Override - public void unbind() { - bound = false; - destinations.get(name).unsubscribe(directHandler); - } - - public boolean isBound() { - return bound; - } + for (Bindable bindable : bindables.values()) { + assertEquals(0, bindable.getInputs().size()); + assertEquals(0, bindable.getOutputs().size());//Must not be bound" } } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/AbstractDestination.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/AbstractDestination.java new file mode 100644 index 000000000..bbd5ab28d --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/AbstractDestination.java @@ -0,0 +1,41 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * 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.integration; + +import org.springframework.messaging.SubscribableChannel; + +/** + * @author Oleg Zhurakousky + * + */ +abstract class AbstractDestination { + + private SubscribableChannel channel; + + SubscribableChannel getChannel() { + return channel; + } + + void setChannel(SubscribableChannel channel) { + this.channel = channel; + this.afterChannelIsSet(); + } + + void afterChannelIsSet() { + // noop + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SampleStreamApp.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SampleStreamApp.java new file mode 100644 index 000000000..23bed3c72 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SampleStreamApp.java @@ -0,0 +1,71 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * 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.integration; + +import java.nio.charset.StandardCharsets; + +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.messaging.Processor; +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.Import; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.messaging.Message; +import org.springframework.messaging.handler.annotation.SendTo; +import org.springframework.messaging.support.GenericMessage; + +import static org.junit.Assert.assertEquals; + +/** + * Sample spring cloud stream application that demonstrates the usage of {@link SpringIntegrationChannelBinder}. + * + * @author Oleg Zhurakousky + * + */ +@SpringBootApplication +@EnableBinding(Processor.class) +@Import(SpringIntegrationBinderConfiguration.class) +public class SampleStreamApp { + + public static void main(String[] args) { + ApplicationContext context = new SpringApplicationBuilder(SampleStreamApp.class).web(WebApplicationType.NONE) + .run("--server.port=0"); + SourceDestination source = context.getBean(SourceDestination.class); + TargetDestination target = context.getBean(TargetDestination.class); + source.send(new GenericMessage("Hello".getBytes())); + + Message message = target.receive(); + assertEquals("Hello", new String((byte[])message.getPayload(), StandardCharsets.UTF_8)); + } + + @StreamListener(Processor.INPUT) + @SendTo(Processor.OUTPUT) + public String receive(String value) { + System.out.println("Handling payload: " + value); + return value; + } + + @ServiceActivator(inputChannel="input.anonymous.errors") + public void error(String value) { + System.out.println("Handling ERROR payload: " + value); + } +} + + diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SourceDestination.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SourceDestination.java new file mode 100644 index 000000000..ee03161c4 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SourceDestination.java @@ -0,0 +1,40 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * 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.integration; + +import org.springframework.messaging.Message; + +/** + * Implementation of binder endpoint that represents the source destination + * (e.g., destination from which messages will be received by Processor.INPUT). + *
+ * You can interact with it by calling {@link #send(Message)} operation. + * + * @author Oleg Zhurakousky + * + */ +public class SourceDestination extends AbstractDestination { + + /** + * Allows the {@link Message} to be sent to a Binder to be delegated + * to binder's input destination (e.g., Processor.INPUT). + * + */ + public void send(Message message) { + this.getChannel().send(message); + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationBinderConfiguration.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationBinderConfiguration.java new file mode 100644 index 000000000..56cb544ec --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationBinderConfiguration.java @@ -0,0 +1,99 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * 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.integration; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binder.BinderType; +import org.springframework.cloud.stream.binder.BinderTypeRegistry; +import org.springframework.cloud.stream.binder.ConsumerProperties; +import org.springframework.cloud.stream.binder.DefaultBinderTypeRegistry; +import org.springframework.cloud.stream.binder.ProducerProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.config.EnableIntegration; + +/** + * {@link Binder} configuration backed by Spring Integration. + * + * Please see {@link SpringIntegrationChannelBinder} for more details. + * + * @author Oleg Zhurakousky + * + * @see SpringIntegrationChannelBinder + */ +@Configuration +@ConditionalOnMissingBean(Binder.class) +@EnableIntegration +public class SpringIntegrationBinderConfiguration { + + public static final String NAME = "integration"; + + /** + * Utility operation to return an array of configuration classes + * defined in {@link EnableBinding} annotation. + * Typically used for tests that do not rely on creating an SCSt boot + * application annotated with {@link EnableBinding}, yet require + * full {@link Binder} configuration. + */ + public static Class[] getCompleteConfiguration() { + List> configClasses = new ArrayList<>(); + configClasses.add(SpringIntegrationBinderConfiguration.class); + Import annotation = AnnotationUtils.getAnnotation(EnableBinding.class, Import.class); + Map annotationAttributes = AnnotationUtils.getAnnotationAttributes(annotation); + configClasses.addAll(Arrays.asList((Class[])annotationAttributes.get("value"))); + return configClasses.toArray(new Class[] {}); + } + + @Bean + public BinderTypeRegistry binderTypeRegistry() { + BinderType binderType = new BinderType(NAME, new Class[] {SpringIntegrationBinderConfiguration.class}); + BinderTypeRegistry btr = new DefaultBinderTypeRegistry(Collections.singletonMap(NAME, binderType)); + return btr; + } + + @Bean + public SourceDestination sourceDestination() { + return new SourceDestination(); + } + + @Bean + public TargetDestination targetDestination() { + return new TargetDestination(); + } + + @SuppressWarnings("unchecked") + @Bean + public Binder springIntegrationChannelBinder(SpringIntegrationProvisioner provisioner) { + return (Binder) new SpringIntegrationChannelBinder(provisioner); + } + + @Bean + public SpringIntegrationProvisioner springIntegrationProvisioner() { + return new SpringIntegrationProvisioner(); + } + +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationChannelBinder.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationChannelBinder.java new file mode 100644 index 000000000..aa1d8921b --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationChannelBinder.java @@ -0,0 +1,257 @@ +/* + * Copyright 2015-2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * 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.integration; + +import java.util.function.Consumer; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder; +import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binder.ConsumerProperties; +import org.springframework.cloud.stream.binder.ProducerProperties; +import org.springframework.cloud.stream.binder.integration.SpringIntegrationProvisioner.SpringIntegrationConsumerDestination; +import org.springframework.cloud.stream.binder.integration.SpringIntegrationProvisioner.SpringIntegrationProducerDestination; +import org.springframework.cloud.stream.provisioning.ConsumerDestination; +import org.springframework.cloud.stream.provisioning.ProducerDestination; +import org.springframework.core.AttributeAccessor; +import org.springframework.integration.core.MessageProducer; +import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.handler.BridgeHandler; +import org.springframework.integration.support.DefaultErrorMessageStrategy; +import org.springframework.integration.support.ErrorMessageStrategy; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.SubscribableChannel; +import org.springframework.retry.RecoveryCallback; +import org.springframework.retry.RetryCallback; +import org.springframework.retry.RetryContext; +import org.springframework.retry.RetryListener; +import org.springframework.retry.support.RetryTemplate; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +/** + * Implementation of {@link Binder} backed by Spring Integration framework. + * It is useful for localized demos and testing. + *

+ * This binder extends from the same base class ({@link AbstractMessageChannelBinder}) as + * other binders (i.e., Rabbit, Kafka etc). Interaction with this binder is done via source + * and target destination which emulate real binder's destinations (i.e., Kafka topic) + *
+ * The destination classes are + *

    + *
  • {@link SourceDestination}
    • + *
  • {@link TargetDestination}
    • + *
+ * Simply autowire them in your your application and send/receive messages. + *

+ * You must also add {@link SpringIntegrationBinderConfiguration} to your configuration. + * Below is the example using Spring Boot test. + * 
+ *
+ * @RunWith(SpringJUnit4ClassRunner.class)
+ * @SpringBootTest(classes = {SpringIntegrationBinderConfiguration.class, TestWithSIBinder.MyProcessor.class})
+ * public class TestWithSIBinder {
+ *     @Autowired
+ *     private SourceDestination sourceDestination;
+ *
+ *     @Autowired
+ *     private TargetDestination targetDestination;
+ *
+ *     @Test
+ *     public void testWiring() {
+ *         sourceDestination.send(new GenericMessage("Hello"));
+ *         assertEquals("Hello world", new String((byte[])targetDestination.receive().getPayload(), StandardCharsets.UTF_8));
+ *     }
+ *
+ *     @SpringBootApplication
+ *     @EnableBinding(Processor.class)
+ *     public static class MyProcessor {
+ *         @StreamListener(Processor.INPUT)
+ *         @SendTo(Processor.OUTPUT)
+ *         public String transform(String in) {
+ *             return in + " world";
+ *         }
+ *     }
+ * }
+ *
+ * + * @author Oleg Zhurakousky + */ +class SpringIntegrationChannelBinder extends AbstractMessageChannelBinder { + + @Autowired + private BeanFactory beanFactory; + + SpringIntegrationChannelBinder(SpringIntegrationProvisioner provisioningProvider) { + super(new String[] {}, provisioningProvider); + } + + @Override + protected MessageHandler createProducerMessageHandler(ProducerDestination destination, + ProducerProperties producerProperties, MessageChannel errorChannel) throws Exception { + BridgeHandler handler = new BridgeHandler(); + handler.setBeanFactory(this.beanFactory); + handler.setOutputChannel(((SpringIntegrationProducerDestination)destination).getChannel()); + return handler; + } + + @Override + protected MessageProducer createConsumerEndpoint(ConsumerDestination destination, String group, ConsumerProperties properties) + throws Exception { + ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy(); + SubscribableChannel siBinderInputChannel = ((SpringIntegrationConsumerDestination)destination).getChannel(); + + IntegrationMessageListeningContainer messageListenerContainer = new IntegrationMessageListeningContainer(); + IntegrationBinderInboundChannelAdapter adapter = new IntegrationBinderInboundChannelAdapter(messageListenerContainer); + + String groupName = StringUtils.hasText(group) ? group : "anonymous"; + ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(destination, groupName, properties); + if (properties.getMaxAttempts() > 1) { + adapter.setRetryTemplate(buildRetryTemplate(properties)); + } + else { + adapter.setErrorMessageStrategy(errorMessageStrategy); + adapter.setErrorChannel(errorInfrastructure.getErrorChannel()); + } + + siBinderInputChannel.subscribe(messageListenerContainer); + + return adapter; + } + + /** + * Implementation of simple message listener container modeled after AMQP SimpleMessageListenerContainer + */ + private static class IntegrationMessageListeningContainer implements MessageHandler { + + private Consumer> listener; + + @Override + public void handleMessage(Message message) throws MessagingException { + this.listener.accept(message); + } + + public void setMessageListener(Consumer> listener) { + this.listener = listener; + } + } + + /** + * Implementation of inbound channel adapter modeled after AmqpInboundChannelAdapter + */ + private static class IntegrationBinderInboundChannelAdapter extends MessageProducerSupport { + + private static final ThreadLocal attributesHolder = new ThreadLocal(); + + private final IntegrationMessageListeningContainer listenerContainer; + + private RetryTemplate retryTemplate; + + private RecoveryCallback recoveryCallback; + + IntegrationBinderInboundChannelAdapter(IntegrationMessageListeningContainer listenerContainer) { + this.listenerContainer = listenerContainer; + } + + @SuppressWarnings("unused") + // Temporarily unused until DLQ strategy for this binder becomes a requirement + public void setRecoveryCallback(RecoveryCallback recoveryCallback) { + this.recoveryCallback = recoveryCallback; + } + + public void setRetryTemplate(RetryTemplate retryTemplate) { + this.retryTemplate = retryTemplate; + } + + @Override + protected void onInit() { + if (this.retryTemplate != null) { + Assert.state(getErrorChannel() == null, "Cannot have an 'errorChannel' property when a 'RetryTemplate' is " + + "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to " + + "send an error message when retries are exhausted"); + } + Listener messageListener = new Listener(); + if (this.retryTemplate != null) { + this.retryTemplate.registerListener(messageListener); + } + this.listenerContainer.setMessageListener(messageListener); + } + + protected class Listener implements RetryListener, Consumer> { + + @Override + @SuppressWarnings("unchecked") + public void accept(Message message) { + try { + if (IntegrationBinderInboundChannelAdapter.this.retryTemplate == null) { + try { + processMessage(message); + } + finally { + attributesHolder.remove(); + } + } + else { + IntegrationBinderInboundChannelAdapter.this.retryTemplate.execute(context -> { + processMessage(message); + return null; + }, + (RecoveryCallback) IntegrationBinderInboundChannelAdapter.this.recoveryCallback); + } + } + catch (RuntimeException e) { + if (getErrorChannel() != null) { + getMessagingTemplate().send(getErrorChannel(), buildErrorMessage(null, + new IllegalStateException("Message conversion failed: " + message, e))); + } + else { + throw e; + } + } + } + + private void processMessage(Message message) { + sendMessage(message); + } + + @Override + public boolean open(RetryContext context, RetryCallback callback) { + if (IntegrationBinderInboundChannelAdapter.this.recoveryCallback != null) { + attributesHolder.set(context); + } + return true; + } + + @Override + public void close(RetryContext context, RetryCallback callback, + Throwable throwable) { + attributesHolder.remove(); + } + + @Override + public void onError(RetryContext context, RetryCallback callback, + Throwable throwable) { + // Empty + } + } + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationProvisioner.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationProvisioner.java new file mode 100644 index 000000000..ddd3a8f46 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationProvisioner.java @@ -0,0 +1,133 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * 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.integration; + +import java.util.HashMap; +import java.util.Map; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binder.ConsumerProperties; +import org.springframework.cloud.stream.binder.ProducerProperties; +import org.springframework.cloud.stream.provisioning.ConsumerDestination; +import org.springframework.cloud.stream.provisioning.ProducerDestination; +import org.springframework.cloud.stream.provisioning.ProvisioningException; +import org.springframework.cloud.stream.provisioning.ProvisioningProvider; +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.PublishSubscribeChannel; +import org.springframework.messaging.Message; +import org.springframework.messaging.SubscribableChannel; + +/** + * {@link ProvisioningProvider} to support {@link SpringIntegrationChannelBinder}. It + * exists primarily to support {@link AbstractMessageChannel} semantics for creating + * {@link ConsumerDestination} and {@link ProducerDestination}, to interact with this + * {@link Binder}. + * + * @author Oleg Zhurakousky + * + */ +class SpringIntegrationProvisioner implements ProvisioningProvider { + + private final Map provisionedDestinations = new HashMap<>(); + + @Autowired + private SourceDestination source; + + @Autowired + private TargetDestination target; + + /** + * Will provision producer destination as an SI {@link PublishSubscribeChannel}. + *
+ * This provides convenience of registering additional subscriber (handler in the test method) + * along side of being able to call {@link TargetDestination#receive()} to get a + * {@link Message} for additional assertions. + */ + @Override + public ProducerDestination provisionProducerDestination(String name, ProducerProperties properties) throws ProvisioningException { + SubscribableChannel destination = this.provisionDestination(name, true); + this.target.setChannel(destination); + return new SpringIntegrationProducerDestination(name, destination); + } + + /** + * Will provision consumer destination as SI {@link DirectChannel} + */ + @Override + public ConsumerDestination provisionConsumerDestination(String name, String group, ConsumerProperties properties) throws ProvisioningException { + SubscribableChannel destination = this.provisionDestination(name, false); + this.source.setChannel(destination); + return new SpringIntegrationConsumerDestination(name, destination); + } + + private SubscribableChannel provisionDestination(String name, boolean pubSub) { + String destinationName = name + ".destination"; + SubscribableChannel destination = this.provisionedDestinations.get(destinationName); + if (destination == null) { + destination = pubSub ? new PublishSubscribeChannel() : new DirectChannel(); + ((AbstractMessageChannel)destination).setBeanName(destinationName); + ((AbstractMessageChannel)destination).setComponentName(destinationName); + this.provisionedDestinations.put(destinationName, destination); + } + return destination; + } + + class SpringIntegrationConsumerDestination implements ConsumerDestination { + private final String name; + private final SubscribableChannel channel; + + SpringIntegrationConsumerDestination(String name, SubscribableChannel channel) { + this.name = name; + this.channel = channel; + } + + public SubscribableChannel getChannel() { + return this.channel; + } + + @Override + public String getName() { + return this.name; + } + } + + class SpringIntegrationProducerDestination implements ProducerDestination { + private final String name; + private final SubscribableChannel channel; + + SpringIntegrationProducerDestination(String name, SubscribableChannel channel) { + this.name = name; + this.channel = channel; + } + + @Override + public String getNameForPartition(int partition) { + return this.getName() + partition; + } + + public SubscribableChannel getChannel() { + return this.channel; + } + + @Override + public String getName() { + return this.name; + } + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/TargetDestination.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/TargetDestination.java new file mode 100644 index 000000000..04ed36fb8 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/TargetDestination.java @@ -0,0 +1,64 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * 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.integration; + +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedTransferQueue; +import java.util.concurrent.TimeUnit; + +import org.springframework.messaging.Message; + +/** + * Implementation of binder endpoint that represents the target destination + * (e.g., destination which receives messages sent to Processor.OUTPUT) + *
+ * You can interact with it by calling {@link #receive()} operation. + * + * @author Oleg Zhurakousky + * + */ +public class TargetDestination extends AbstractDestination { + + private BlockingQueue> messages; + + /** + * Allows to access {@link Message}s received by this {@link TargetDestination}. + * @param timeout how long to wait before giving up + */ + public Message receive(long timeout) { + try { + return this.messages.poll(timeout, TimeUnit.MILLISECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return null; + } + + /** + * Allows to access {@link Message}s received by this {@link TargetDestination}. + */ + public Message receive() { + return this.receive(0); + } + + @Override + void afterChannelIsSet() { + this.messages = new LinkedTransferQueue<>(); + this.getChannel().subscribe(message -> messages.offer(message)); + } +}