diff --git a/spring-cloud-stream/deleted-tests/ArbitraryInterfaceWithDefaultsTests.java b/spring-cloud-stream/deleted-tests/ArbitraryInterfaceWithDefaultsTests.java deleted file mode 100644 index d0c514b30..000000000 --- a/spring-cloud-stream/deleted-tests/ArbitraryInterfaceWithDefaultsTests.java +++ /dev/null @@ -1,71 +0,0 @@ -/* - * 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 - * - * https://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.junit.Test; -import org.junit.runner.RunWith; -import org.mockito.Mockito; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.messaging.MessageChannel; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; - -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.ArgumentMatchers.isNull; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.verifyNoMoreInteractions; - -/** - * @author Marius Bogoevici - * @author Janne Valkealahti - */ -// @checkstyle:off -@RunWith(SpringJUnit4ClassRunner.class) -@SpringBootTest(classes = ArbitraryInterfaceWithDefaultsTests.TestFooChannels.class, properties = "spring.cloud.stream.default-binder=mock") -public class ArbitraryInterfaceWithDefaultsTests { - - // @checkstyle:on - - @Autowired - public FooChannels fooChannels; - - @Autowired - private BinderFactory binderFactory; - - @SuppressWarnings({ "rawtypes", "unchecked" }) - @Test - public void testArbitraryInterfaceChannelsBound() { - final Binder binder = this.binderFactory.getBinder(null, MessageChannel.class); - verify(binder).bindConsumer(eq("foo"), isNull(), eq(this.fooChannels.foo()), - Mockito.any()); - verify(binder).bindConsumer(eq("bar"), isNull(), eq(this.fooChannels.bar()), - Mockito.any()); - verify(binder).bindProducer(eq("baz"), eq(this.fooChannels.baz()), Mockito.any()); - verify(binder).bindProducer(eq("qux"), eq(this.fooChannels.qux()), Mockito.any()); - verifyNoMoreInteractions(binder); - } - - @EnableBinding(FooChannels.class) - @EnableAutoConfiguration - public static class TestFooChannels { - - } - -} diff --git a/spring-cloud-stream/deleted-tests/BinderFactoryAutoConfigurationTests.java b/spring-cloud-stream/deleted-tests/BinderFactoryAutoConfigurationTests.java deleted file mode 100644 index 3f74cfc02..000000000 --- a/spring-cloud-stream/deleted-tests/BinderFactoryAutoConfigurationTests.java +++ /dev/null @@ -1,339 +0,0 @@ -/* - * Copyright 2015-2019 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 - * - * https://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 java.io.IOException; -import java.net.URL; -import java.net.URLClassLoader; - -import org.junit.Test; - -import org.springframework.boot.WebApplicationType; -import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration; -import org.springframework.boot.autoconfigure.web.servlet.ServletWebServerFactoryAutoConfiguration; -import org.springframework.boot.builder.SpringApplicationBuilder; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.binder.stub1.StubBinder1; -import org.springframework.cloud.stream.binder.stub1.StubBinder1Configuration; -import org.springframework.cloud.stream.binder.stub2.StubBinder2; -import org.springframework.cloud.stream.binder.stub2.StubBinder2ConfigurationA; -import org.springframework.cloud.stream.binder.stub2.StubBinder2ConfigurationB; -import org.springframework.cloud.stream.config.BinderFactoryAutoConfiguration; -import org.springframework.cloud.stream.config.BindingServiceConfiguration; -import org.springframework.context.ApplicationContext; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.context.annotation.Import; -import org.springframework.core.io.ClassPathResource; -import org.springframework.core.io.DefaultResourceLoader; -import org.springframework.messaging.MessageChannel; -import org.springframework.util.ObjectUtils; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.junit.Assert.fail; - -/** - * @author Marius Bogoevici - * @author Ilayaperumal Gopinathan - * @author Soby Chacko - * @author Artem Bilan - * @author Anshul Mehra - */ -public class BinderFactoryAutoConfigurationTests { - - private static ClassLoader createClassLoader(String[] additionalClasspathDirectories, - String... properties) throws IOException { - URL[] urls = ObjectUtils.isEmpty(additionalClasspathDirectories) ? new URL[0] - : new URL[additionalClasspathDirectories.length]; - if (!ObjectUtils.isEmpty(additionalClasspathDirectories)) { - for (int i = 0; i < additionalClasspathDirectories.length; i++) { - urls[i] = new URL(new ClassPathResource(additionalClasspathDirectories[i]) - .getURL().toString() + "/"); - } - } - return new URLClassLoader(urls, - BinderFactoryAutoConfigurationTests.class.getClassLoader()); - } - - private static ConfigurableApplicationContext createBinderTestContext( - String[] additionalClasspathDirectories, String... properties) - throws IOException { - ClassLoader classLoader = createClassLoader(additionalClasspathDirectories, - properties); - return new SpringApplicationBuilder(SimpleApplication.class) - .resourceLoader(new DefaultResourceLoader(classLoader)) - .properties(properties).web(WebApplicationType.NONE).run(); - } - - private static ConfigurableApplicationContext createBinderTestContextWithSources( - Class[] sources, String[] additionalClasspathDirectories, - String... properties) throws IOException { - ClassLoader classLoader = createClassLoader(additionalClasspathDirectories, - properties); - return new SpringApplicationBuilder(sources) - .resourceLoader(new DefaultResourceLoader(classLoader)) - .properties(properties).web(WebApplicationType.NONE).run(); - } - - @Test - public void loadBinderTypeRegistryWithSelfContainedAggregatorApp() throws Exception { - createBinderTestContextWithSources(new Class[] { SimpleApplication.class }, - new String[] {}, "spring.cloud.stream.internal.selfContained=true"); - } - - @SuppressWarnings("rawtypes") - @Test - public void loadBinderTypeRegistryWithOneBinder() throws Exception { - ConfigurableApplicationContext context = createBinderTestContext( - new String[] { "binder1" }, "spring.cloud.stream.default-binder=binder1"); - - BinderTypeRegistry binderTypeRegistry = context.getBean(BinderTypeRegistry.class); - assertThat(binderTypeRegistry).isNotNull(); - assertThat(binderTypeRegistry.getAll()).hasSize(3); - assertThat(binderTypeRegistry.getAll()).containsKey("binder1"); - assertThat((Class[]) binderTypeRegistry.get("binder1").getConfigurationClasses()) - .containsExactlyInAnyOrder(StubBinder1Configuration.class); - - BinderFactory binderFactory = context.getBean(BinderFactory.class); - - Binder binder1 = binderFactory.getBinder("binder1", MessageChannel.class); - assertThat(binder1).isInstanceOf(StubBinder1.class); - - Binder defaultBinder = binderFactory.getBinder(null, MessageChannel.class); - assertThat(defaultBinder).isSameAs(binder1); - } - - @SuppressWarnings("rawtypes") - @Test - public void loadBinderTypeRegistryWithOneBinderAndSharedEnvironment() - throws Exception { - ConfigurableApplicationContext context = createBinderTestContext( - new String[] { "binder1" }, "binder1.name=foo"); - - BinderFactory binderFactory = context.getBean(BinderFactory.class); - - Binder binder1 = binderFactory.getBinder("binder1", MessageChannel.class); - assertThat(binder1).hasFieldOrPropertyWithValue("name", "foo"); - } - - /* - * See https://github.com/spring-cloud/spring-cloud-stream/issues/1708 - */ - @SuppressWarnings("rawtypes") - @Test - public void loadBinderTypeRegistryWithSharedEnvironmentAndServletWebApplicationType() - throws Exception { - String[] properties = new String[] {"binder1.name=foo", "spring.main.web-application-type=SERVLET"}; - ClassLoader classLoader = createClassLoader(new String[] { "binder1" }, - properties); - ConfigurableApplicationContext context = new SpringApplicationBuilder(SimpleApplication.class, ServletWebServerFactoryAutoConfiguration.class) - .resourceLoader(new DefaultResourceLoader(classLoader)) - .properties(properties).web(WebApplicationType.SERVLET).run(); - - BinderFactory binderFactory = context.getBean(BinderFactory.class); - - Binder binder1 = binderFactory.getBinder("binder1", MessageChannel.class); - assertThat(binder1).hasFieldOrPropertyWithValue("name", "foo"); - } - - @SuppressWarnings("rawtypes") - @Test - public void loadBinderTypeRegistryWithOneCustomBinderAndSharedEnvironment() - throws Exception { - ConfigurableApplicationContext context = createBinderTestContext( - new String[] { "binder1" }, "binder1.name=foo", - "spring.cloud.stream.binders.custom.environment.foo=bar", - "spring.cloud.stream.binders.custom.environment.spring.main.sources=" + AdditionalBinderConfiguration.class.getName(), - "spring.cloud.stream.binders.custom.type=binder1"); - - BinderFactory binderFactory = context.getBean(BinderFactory.class); - - Binder binder1 = binderFactory.getBinder("custom", MessageChannel.class); - assertThat(binder1).hasFieldOrPropertyWithValue("name", "foo"); - - assertThat(binderFactory.getBinder(null, MessageChannel.class)).isSameAs(binder1); - - SimpleApplication simpleApplication = context.getBean(SimpleApplication.class); - - assertThat(simpleApplication.binderContext).isNotNull(); - - assertThat(simpleApplication.binderContext.containsBean("fooBean")).isTrue(); - } - - @SuppressWarnings("rawtypes") - @Test - public void testCustomEnvironmentHasAccessToOuterContext() throws Exception { - ConfigurableApplicationContext context = createBinderTestContext( - new String[] { "binder1" }, "binder1.name=foo", - "spring.cloud.stream.binders.custom.environment.foo=bar", - "spring.cloud.stream.binders.custom.type=binder1"); - - BinderFactory binderFactory = context.getBean(BinderFactory.class); - - Binder binder1 = binderFactory.getBinder("custom", MessageChannel.class); - - assertThat(binder1).hasFieldOrPropertyWithValue("name", "foo"); - assertThat(binder1).hasFieldOrPropertyWithValue("outerContext", context); - - assertThat(binderFactory.getBinder(null, MessageChannel.class)).isSameAs(binder1); - } - - @SuppressWarnings("rawtypes") - @Test - public void testStandardBinderDoesNotHaveTheOuterContextBean() throws Exception { - ConfigurableApplicationContext context = createBinderTestContext( - new String[] { "binder1" }, "binder1.name=foo"); - - BinderFactory binderFactory = context.getBean(BinderFactory.class); - Binder binder1 = binderFactory.getBinder("binder1", MessageChannel.class); - assertThat(binder1).hasFieldOrPropertyWithValue("name", "foo"); - - assertThat(((StubBinder1) binder1).getOuterContext()).isNull(); - } - - @SuppressWarnings("rawtypes") - @Test - public void loadBinderTypeRegistryWithTwoBinders() throws Exception { - ConfigurableApplicationContext context = createBinderTestContext( - new String[] { "binder1", "binder2" }); - BinderTypeRegistry binderTypeRegistry = context.getBean(BinderTypeRegistry.class); - assertThat(binderTypeRegistry).isNotNull(); - assertThat(binderTypeRegistry.getAll()).hasSize(4); - assertThat(binderTypeRegistry.getAll()).containsOnlyKeys("binder1", "binder2", - "mock", "integration"); - assertThat((Class[]) binderTypeRegistry.get("binder1").getConfigurationClasses()) - .containsExactly(StubBinder1Configuration.class); - assertThat((Class[]) binderTypeRegistry.get("binder2").getConfigurationClasses()) - .containsExactlyInAnyOrder(StubBinder2ConfigurationA.class, - StubBinder2ConfigurationB.class); - - BinderFactory binderFactory = context.getBean(BinderFactory.class); - - try { - binderFactory.getBinder(null, MessageChannel.class); - fail("Should throw an exception"); - } - catch (Exception e) { - assertThat(e).isInstanceOf(IllegalStateException.class); - assertThat(e.getMessage()).contains( - "A default binder has been requested, but there is more than one binder available"); - } - - Binder binder1 = binderFactory.getBinder("binder1", MessageChannel.class); - assertThat(binder1).isInstanceOf(StubBinder1.class); - Binder binder2 = binderFactory.getBinder("binder2", MessageChannel.class); - assertThat(binder2).isInstanceOf(StubBinder2.class); - } - - @SuppressWarnings("rawtypes") - @Test - public void loadBinderTypeRegistryWithCustomNonDefaultCandidate() throws Exception { - ConfigurableApplicationContext context = createBinderTestContext( - new String[] { "binder1" }, - "spring.cloud.stream.binders.custom.type=binder1", - "spring.cloud.stream.binders.custom.defaultCandidate=false", - "spring.cloud.stream.binders.custom.inheritEnvironment=false", - "spring.cloud.stream.default-binder=binder1"); - BinderTypeRegistry binderTypeRegistry = context.getBean(BinderTypeRegistry.class); - assertThat(binderTypeRegistry).isNotNull(); - assertThat(binderTypeRegistry.getAll().size()).isEqualTo(3); - assertThat(binderTypeRegistry.getAll().keySet()).contains("binder1"); - assertThat((Class[]) binderTypeRegistry.get("binder1").getConfigurationClasses()) - .contains(StubBinder1Configuration.class); - - BinderFactory binderFactory = context.getBean(BinderFactory.class); - - Binder defaultBinder = binderFactory.getBinder(null, MessageChannel.class); - assertThat(defaultBinder).isInstanceOf(StubBinder1.class); - assertThat(((StubBinder1) defaultBinder).getName()).isNullOrEmpty(); - - Binder binder1 = binderFactory.getBinder("binder1", MessageChannel.class); - assertThat(binder1).isInstanceOf(StubBinder1.class); - assertThat(binder1).isSameAs(defaultBinder); - } - - @SuppressWarnings("rawtypes") - @Test - public void loadDefaultBinderWithTwoBinders() throws Exception { - - ConfigurableApplicationContext context = createBinderTestContext( - new String[] { "binder1", "binder2" }, - "spring.cloud.stream.defaultBinder:binder2"); - BinderTypeRegistry binderTypeRegistry = context.getBean(BinderTypeRegistry.class); - assertThat(binderTypeRegistry).isNotNull(); - assertThat(binderTypeRegistry.getAll()).hasSize(4); - assertThat(binderTypeRegistry.getAll()).containsOnlyKeys("binder1", "binder2", - "mock", "integration"); - assertThat((Class[]) binderTypeRegistry.get("binder1").getConfigurationClasses()) - .containsExactlyInAnyOrder(StubBinder1Configuration.class); - assertThat((Class[]) binderTypeRegistry.get("binder2").getConfigurationClasses()) - .containsExactlyInAnyOrder(StubBinder2ConfigurationA.class, - StubBinder2ConfigurationB.class); - - BinderFactory binderFactory = context.getBean(BinderFactory.class); - - Binder binder1 = binderFactory.getBinder("binder1", MessageChannel.class); - assertThat(binder1).isInstanceOf(StubBinder1.class); - assertThat(((StubBinder1) binder1).getFromCustomization()).isEqualTo("customizer-applied-binder1"); - Binder binder2 = binderFactory.getBinder("binder2", MessageChannel.class); - assertThat(binder2).isInstanceOf(StubBinder2.class); - assertThat(((StubBinder2) binder2).getFromCustomization()).isEqualTo("customizer-applied-binder2"); - - Binder defaultBinder = binderFactory.getBinder(null, MessageChannel.class); - assertThat(defaultBinder).isSameAs(binder2); - } - - @Import({ BinderFactoryAutoConfiguration.class, - PropertyPlaceholderAutoConfiguration.class, - BindingServiceConfiguration.class }) - @EnableBinding - public static class SimpleApplication { - - private volatile ApplicationContext binderContext; - - @Bean - public DefaultBinderFactory.Listener testBinderListener() { - return (configurationName, binderContext) -> { - this.binderContext = binderContext; - }; - - } - - @Bean - public BinderCustomizer binderCustomizer() { - return (binder, binderName) -> { - if (binder instanceof StubBinder1) { - ((StubBinder1) binder).setFromCustomization("customizer-applied-binder1"); - } - else if (binder instanceof StubBinder2) { - ((StubBinder2) binder).setFromCustomization("customizer-applied-binder2"); - } - }; - } - } - - @Configuration - public static class AdditionalBinderConfiguration { - - @Bean - public String fooBean() { - return "foo"; - } - } - -} diff --git a/spring-cloud-stream/deleted-tests/BindingServiceTests.java b/spring-cloud-stream/deleted-tests/BindingServiceTests.java deleted file mode 100644 index 4f4003495..000000000 --- a/spring-cloud-stream/deleted-tests/BindingServiceTests.java +++ /dev/null @@ -1,780 +0,0 @@ -/* - * Copyright 2015-2018 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 - * - * https://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.binding; - -import java.lang.reflect.Field; -import java.util.Arrays; -import java.util.Collection; -import java.util.Collections; -import java.util.HashMap; -import java.util.Iterator; -import java.util.List; -import java.util.Map; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; - -import com.fasterxml.jackson.databind.ObjectMapper; -import org.junit.Ignore; -import org.junit.Test; -import org.mockito.ArgumentCaptor; -import org.mockito.Mockito; - -import org.springframework.beans.factory.ObjectProvider; -import org.springframework.boot.SpringApplication; -import org.springframework.boot.WebApplicationType; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.builder.SpringApplicationBuilder; -import org.springframework.boot.context.properties.source.MapConfigurationPropertySource; -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.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.Binding; -import org.springframework.cloud.stream.binder.ConsumerProperties; -import org.springframework.cloud.stream.binder.DefaultBinderFactory; -import org.springframework.cloud.stream.binder.DefaultBinderTypeRegistry; -import org.springframework.cloud.stream.binder.ProducerProperties; -import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; -import org.springframework.cloud.stream.config.BindingProperties; -import org.springframework.cloud.stream.config.BindingServiceConfiguration; -import org.springframework.cloud.stream.config.BindingServiceProperties; -import org.springframework.cloud.stream.messaging.Processor; -import org.springframework.cloud.stream.messaging.Sink; -import org.springframework.cloud.stream.reflection.GenericsUtils; -import org.springframework.cloud.stream.utils.MockBinderConfiguration; -import org.springframework.context.ApplicationContext; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Import; -import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.test.util.TestUtils; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.SubscribableChannel; -import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; -import org.springframework.util.ReflectionUtils; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.fail; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.ArgumentMatchers.isNull; -import static org.mockito.ArgumentMatchers.same; -import static org.mockito.Mockito.doAnswer; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; - -/** - * @author Gary Russell - * @author Mark Fisher - * @author Marius Bogoevici - * @author Ilayaperumal Gopinathan - * @author Janne Valkealahti - * @author Soby Chacko - * @author Michael Michailidis - */ -public class BindingServiceTests { - - @SuppressWarnings({ "unchecked", "rawtypes" }) - @Test - public void testDefaultGroup() throws Exception { - BindingServiceProperties properties = new BindingServiceProperties(); - Map bindingProperties = new HashMap<>(); - BindingProperties props = new BindingProperties(); - props.setDestination("foo"); - final String inputChannelName = "input"; - bindingProperties.put(inputChannelName, props); - properties.setBindings(bindingProperties); - DefaultBinderFactory binderFactory = createMockBinderFactory(); - Binder binder = binderFactory.getBinder("mock", MessageChannel.class); - BindingService service = new BindingService(properties, binderFactory, new ObjectMapper()); - MessageChannel inputChannel = new DirectChannel(); - Binding mockBinding = Mockito.mock(Binding.class); - when(binder.bindConsumer(eq("foo"), isNull(), same(inputChannel), - any(ConsumerProperties.class))).thenReturn(mockBinding); - Collection> bindings = service.bindConsumer(inputChannel, - inputChannelName); - assertThat(bindings).hasSize(1); - Binding binding = bindings.iterator().next(); - assertThat(binding).isSameAs(mockBinding); - service.unbindConsumers(inputChannelName); - verify(binder).bindConsumer(eq("foo"), isNull(), same(inputChannel), - any(ConsumerProperties.class)); - verify(binding).unbind(); - binderFactory.destroy(); - } - - @SuppressWarnings({ "unchecked", "rawtypes" }) - @Test - public void testMultipleConsumerBindings() throws Exception { - BindingServiceProperties properties = new BindingServiceProperties(); - Map bindingProperties = new HashMap<>(); - BindingProperties props = new BindingProperties(); - props.setDestination("foo,bar"); - final String inputChannelName = "input"; - bindingProperties.put(inputChannelName, props); - - properties.setBindings(bindingProperties); - - DefaultBinderFactory binderFactory = createMockBinderFactory(); - - Binder binder = binderFactory.getBinder("mock", MessageChannel.class); - BindingService service = new BindingService(properties, binderFactory, new ObjectMapper()); - MessageChannel inputChannel = new DirectChannel(); - - Binding mockBinding1 = Mockito.mock(Binding.class); - Binding mockBinding2 = Mockito.mock(Binding.class); - - when(binder.bindConsumer(eq("foo"), isNull(), same(inputChannel), - any(ConsumerProperties.class))).thenReturn(mockBinding1); - when(binder.bindConsumer(eq("bar"), isNull(), same(inputChannel), - any(ConsumerProperties.class))).thenReturn(mockBinding2); - - Collection> bindings = service.bindConsumer(inputChannel, - "input"); - assertThat(bindings).hasSize(2); - - Iterator> iterator = bindings.iterator(); - Binding binding1 = iterator.next(); - Binding binding2 = iterator.next(); - - assertThat(binding1).isSameAs(mockBinding1); - assertThat(binding2).isSameAs(mockBinding2); - - service.unbindConsumers("input"); - - verify(binder).bindConsumer(eq("foo"), isNull(), same(inputChannel), - any(ConsumerProperties.class)); - verify(binder).bindConsumer(eq("bar"), isNull(), same(inputChannel), - any(ConsumerProperties.class)); - verify(binding1).unbind(); - verify(binding2).unbind(); - - binderFactory.destroy(); - } - - @SuppressWarnings({ "unchecked", "rawtypes" }) - @Test - public void testMultipleConsumerBindingsFromIndexList() throws Exception { - BindingServiceProperties properties = new BindingServiceProperties(); - Map bindingProperties = new HashMap<>(); - BindingProperties props = new BindingProperties(); - props.setDestination("foo"); - - ConsumerProperties consumer = properties.getConsumerProperties("input"); - consumer.setInstanceIndexList(Arrays.asList(0, 1)); - consumer.setInstanceCount(2); - consumer.setPartitioned(true); - props.setConsumer(consumer); - - final String inputChannelName = "input"; - bindingProperties.put(inputChannelName, props); - - properties.setBindings(bindingProperties); - - DefaultBinderFactory binderFactory = createMockBinderFactory(); - - Binder binder = binderFactory.getBinder("mock", MessageChannel.class); - BindingService service = new BindingService(properties, binderFactory, new ObjectMapper()); - MessageChannel inputChannel = new DirectChannel(); - - Binding mockBinding1 = Mockito.mock(Binding.class, "FirstBinding"); - Binding mockBinding2 = Mockito.mock(Binding.class, "SecondBinding"); - - ArgumentCaptor captor = ArgumentCaptor.forClass(ConsumerProperties.class); - - when(binder.bindConsumer(eq("foo"), isNull(), same(inputChannel), - any(ConsumerProperties.class))).thenReturn(mockBinding1).thenReturn(mockBinding2); - - Collection> bindings = service.bindConsumer(inputChannel, - "input"); - assertThat(bindings).hasSize(2); - - Iterator> iterator = bindings.iterator(); - Binding binding1 = iterator.next(); - Binding binding2 = iterator.next(); - - assertThat(binding1).isSameAs(mockBinding1); - assertThat(binding2).isSameAs(mockBinding2); - - service.unbindConsumers("input"); - - verify(binder, times(2)).bindConsumer(eq("foo"), isNull(), same(inputChannel), - captor.capture()); - verify(binding1).unbind(); - verify(binding2).unbind(); - - List allValues = captor.getAllValues(); - - assertThat(allValues.size()).isEqualTo(2); - - assertThat(allValues.get(0).getInstanceIndex()).isEqualTo(0); - assertThat(allValues.get(1).getInstanceIndex()).isEqualTo(1); - - binderFactory.destroy(); - } - - @SuppressWarnings({ "unchecked", "rawtypes" }) - @Test - public void testConsumerBindingWhenMultiplexingIsEnabled() throws Exception { - BindingServiceProperties properties = new BindingServiceProperties(); - Map bindingProperties = new HashMap<>(); - BindingProperties props = new BindingProperties(); - props.setDestination("foo,bar"); - - ConsumerProperties consumer = properties.getConsumerProperties("input"); - consumer.setMultiplex(true); - props.setConsumer(consumer); - - final String inputChannelName = "input"; - bindingProperties.put(inputChannelName, props); - - properties.setBindings(bindingProperties); - - DefaultBinderFactory binderFactory = createMockBinderFactory(); - - Binder binder = binderFactory.getBinder("mock", MessageChannel.class); - BindingService service = new BindingService(properties, binderFactory, new ObjectMapper()); - MessageChannel inputChannel = new DirectChannel(); - - Binding mockBinding1 = Mockito.mock(Binding.class); - - when(binder.bindConsumer(eq("foo,bar"), isNull(), same(inputChannel), - any(ConsumerProperties.class))).thenReturn(mockBinding1); - - Collection> bindings = service.bindConsumer(inputChannel, - "input"); - assertThat(bindings).hasSize(1); - - Iterator> iterator = bindings.iterator(); - Binding binding1 = iterator.next(); - - assertThat(binding1).isSameAs(mockBinding1); - - service.unbindConsumers("input"); - - verify(binder).bindConsumer(eq("foo,bar"), isNull(), same(inputChannel), - any(ConsumerProperties.class)); - verify(binding1).unbind(); - - binderFactory.destroy(); - } - - @SuppressWarnings({ "unchecked", "rawtypes" }) - @Test - public void testExplicitGroup() throws Exception { - BindingServiceProperties properties = new BindingServiceProperties(); - Map bindingProperties = new HashMap<>(); - BindingProperties props = new BindingProperties(); - props.setDestination("foo"); - props.setGroup("fooGroup"); - final String inputChannelName = "input"; - bindingProperties.put(inputChannelName, props); - properties.setBindings(bindingProperties); - DefaultBinderFactory binderFactory = createMockBinderFactory(); - Binder binder = binderFactory.getBinder("mock", MessageChannel.class); - BindingService service = new BindingService(properties, binderFactory, new ObjectMapper()); - MessageChannel inputChannel = new DirectChannel(); - Binding mockBinding = Mockito.mock(Binding.class); - when(binder.bindConsumer(eq("foo"), eq("fooGroup"), same(inputChannel), - any(ConsumerProperties.class))).thenReturn(mockBinding); - Collection> bindings = service.bindConsumer(inputChannel, - inputChannelName); - assertThat(bindings).hasSize(1); - Binding binding = bindings.iterator().next(); - assertThat(binding).isSameAs(mockBinding); - - service.unbindConsumers(inputChannelName); - verify(binder).bindConsumer(eq("foo"), eq(props.getGroup()), same(inputChannel), - any(ConsumerProperties.class)); - verify(binding).unbind(); - binderFactory.destroy(); - } - - //TODO: Need to re-write the following test. - //@SuppressWarnings({ "rawtypes", "unchecked" }) -// @Test -// public void checkDynamicBinding() { -// BindingServiceProperties properties = new BindingServiceProperties(); -// BindingProperties bindingProperties = new BindingProperties(); -// bindingProperties.setProducer(new ProducerProperties()); -// properties.setBindings(Collections.singletonMap("foo", bindingProperties)); -// DefaultBinderFactory binderFactory = createMockBinderFactory(); -// final ExtendedPropertiesBinder binder = mock(ExtendedPropertiesBinder.class); -// Properties extendedProps = new Properties(); -// when(binder.getExtendedProducerProperties(anyString())).thenReturn(extendedProps); -// Binding mockBinding = Mockito.mock(Binding.class); -// final AtomicReference dynamic = new AtomicReference<>(); -// when(binder.bindProducer(matches("foo"), any(DirectChannel.class), -// any(ProducerProperties.class))).thenReturn(mockBinding); -// BindingService bindingService = new BindingService(properties, binderFactory, new ObjectMapper()) { -// -// @Override -// protected Binder getBinder(String channelName, -// Class bindableType) { -// return binder; -// } -// -// }; -// SubscribableChannelBindingTargetFactory bindableSubscribableChannelFactory; -// bindableSubscribableChannelFactory = new SubscribableChannelBindingTargetFactory( -// new MessageConverterConfigurer(properties, -// new CompositeMessageConverterFactory().getMessageConverterForAllRegistered())); -// final AtomicBoolean callbackInvoked = new AtomicBoolean(); -// BinderAwareChannelResolver resolver = new BinderAwareChannelResolver( -// bindingService, bindableSubscribableChannelFactory, -// new DynamicDestinationsBindable(), (name, channel, props, extended) -> { -// callbackInvoked.set(true); -// assertThat(name).isEqualTo("foo"); -// assertThat(channel).isNotNull(); -// assertThat(props).isNotNull(); -// assertThat(extended).isSameAs(extendedProps); -// props.setUseNativeEncoding(true); -// extendedProps.setProperty("bar", "baz"); -// }); -// ConfigurableListableBeanFactory beanFactory = mock( -// ConfigurableListableBeanFactory.class); -// when(beanFactory.getBean("foo", MessageChannel.class)) -// .thenThrow(new NoSuchBeanDefinitionException(MessageChannel.class)); -// when(beanFactory.getBean("bar", MessageChannel.class)) -// .thenThrow(new NoSuchBeanDefinitionException(MessageChannel.class)); -// doAnswer(new Answer() { -// -// @Override -// public Void answer(InvocationOnMock invocation) throws Throwable { -// dynamic.set(invocation.getArgument(1)); -// return null; -// } -// -// }).when(beanFactory).registerSingleton(eq("foo"), any(MessageChannel.class)); -// doAnswer(new Answer() { -// -// @Override -// public Object answer(InvocationOnMock invocation) throws Throwable { -// return dynamic.get(); -// } -// -// }).when(beanFactory).initializeBean(any(MessageChannel.class), eq("foo")); -// resolver.setBeanFactory(beanFactory); -// MessageChannel resolved = resolver.resolveDestination("foo"); -// assertThat(resolved).isSameAs(dynamic.get()); -// ArgumentCaptor captor = ArgumentCaptor -// .forClass(ProducerProperties.class); -// verify(binder).bindProducer(eq("foo"), eq(dynamic.get()), captor.capture()); -// assertThat(captor.getValue().isUseNativeEncoding()).isTrue(); -// assertThat(captor.getValue()).isInstanceOf(ExtendedProducerProperties.class); -// assertThat(((ExtendedProducerProperties) captor.getValue()).getExtension()) -// .isSameAs(extendedProps); -// doReturn(dynamic.get()).when(beanFactory).getBean("foo", MessageChannel.class); -// properties.setDynamicDestinations(new String[] { "foo" }); -// resolved = resolver.resolveDestination("foo"); -// assertThat(resolved).isSameAs(dynamic.get()); -// properties.setDynamicDestinations(new String[] { "test" }); -// try { -// resolver.resolveDestination("bar"); -// fail("Should throw an exception"); -// } -// catch (DestinationResolutionException e) { -// assertThat(e).hasMessageContaining( -// "Failed to find MessageChannel bean with name 'bar'"); -// } -// } - - @Test - public void testProducerPropertiesValidation() { - BindingServiceProperties serviceProperties = new BindingServiceProperties(); - Map bindingProperties = new HashMap<>(); - BindingProperties props = new BindingProperties(); - ProducerProperties producerProperties = new ProducerProperties(); - producerProperties.setPartitionCount(0); - props.setDestination("foo"); - props.setProducer(producerProperties); - final String outputChannelName = "output"; - bindingProperties.put(outputChannelName, props); - serviceProperties.setBindings(bindingProperties); - DefaultBinderFactory binderFactory = createMockBinderFactory(); - BindingService service = new BindingService(serviceProperties, binderFactory, new ObjectMapper()); - MessageChannel outputChannel = new DirectChannel(); - try { - service.bindProducer(outputChannel, outputChannelName); - fail("Producer properties should be validated."); - } - catch (IllegalStateException e) { - assertThat(e) - .hasMessageContaining("Partition count should be greater than zero."); - } - } - - @Test - public void testDefaultPropertyBehavior() { - ConfigurableApplicationContext run = SpringApplication.run( - DefaultConsumerPropertiesTestSink.class, - "--server.port=0", - "--spring.cloud.stream.default.contentType=text/plain", - "--spring.cloud.stream.bindings.input1.contentType=application/json", - "--spring.cloud.stream.default.group=foo", - "--spring.cloud.stream.bindings.input2.group=bar", - "--spring.cloud.stream.default.consumer.concurrency=5", - "--spring.cloud.stream.bindings.input2.consumer.concurrency=1", - "--spring.cloud.stream.bindings.input1.consumer.partitioned=true", - "--spring.cloud.stream.default.producer.partitionCount=10", - "--spring.cloud.stream.bindings.output2.producer.partitionCount=1", - "--spring.cloud.stream.bindings.inputXyz.contentType=application/json", - "--spring.cloud.stream.bindings.inputFooBar.contentType=application/avro", - "--spring.cloud.stream.bindings.input_snake_case.contentType=application/avro"); - - BindingServiceProperties bindingServiceProperties = run.getBeanFactory() - .getBean(BindingServiceProperties.class); - Map bindings = bindingServiceProperties.getBindings(); - - assertThat(bindings.get("input1").getContentType()).isEqualTo("application/json"); - assertThat(bindings.get("input2").getContentType()).isEqualTo("text/plain"); - assertThat(bindings.get("input1").getGroup()).isEqualTo("foo"); - assertThat(bindings.get("input2").getGroup()).isEqualTo("bar"); - assertThat(bindings.get("input1").getConsumer().getConcurrency()).isEqualTo(5); - assertThat(bindings.get("input2").getConsumer().getConcurrency()).isEqualTo(1); - assertThat(bindings.get("input1").getConsumer().isPartitioned()).isEqualTo(true); - assertThat(bindings.get("input2").getConsumer().isPartitioned()).isEqualTo(false); - assertThat(bindings.get("output1").getProducer().getPartitionCount()) - .isEqualTo(10); - assertThat(bindings.get("output2").getProducer().getPartitionCount()) - .isEqualTo(1); - - assertThat(bindings.get("inputXyz").getContentType()) - .isEqualTo("application/json"); - assertThat(bindings.get("inputFooBar").getContentType()) - .isEqualTo("application/avro"); - assertThat(bindings.get("inputFooBarBuzz").getContentType()) - .isEqualTo("text/plain"); - assertThat(bindings.get("input_snake_case").getContentType()) - .isEqualTo("application/avro"); - - run.close(); - } - - @Test - public void testConsumerPropertiesValidation() { - BindingServiceProperties serviceProperties = new BindingServiceProperties(); - Map bindingProperties = new HashMap<>(); - BindingProperties props = new BindingProperties(); - ConsumerProperties consumerProperties = new ConsumerProperties(); - consumerProperties.setConcurrency(0); - props.setDestination("foo"); - props.setConsumer(consumerProperties); - final String inputChannelName = "input"; - bindingProperties.put(inputChannelName, props); - serviceProperties.setBindings(bindingProperties); - DefaultBinderFactory binderFactory = createMockBinderFactory(); - BindingService service = new BindingService(serviceProperties, binderFactory, new ObjectMapper()); - MessageChannel inputChannel = new DirectChannel(); - try { - service.bindConsumer(inputChannel, inputChannelName); - fail("Consumer properties should be validated."); - } - catch (IllegalStateException e) { - assertThat(e) - .hasMessageContaining("Concurrency should be greater than zero."); - } - } - - @Test - public void testUnknownBinderOnBindingFailure() { - HashMap properties = new HashMap<>(); - properties.put("spring.cloud.stream.bindings.input.destination", "fooInput"); - properties.put("spring.cloud.stream.bindings.input.binder", "mock"); - properties.put("spring.cloud.stream.bindings.output.destination", "fooOutput"); - properties.put("spring.cloud.stream.bindings.output.binder", "mockError"); - BindingServiceProperties bindingServiceProperties = createBindingServiceProperties( - properties); - BindingService bindingService = new BindingService(bindingServiceProperties, - createMockBinderFactory(), new ObjectMapper()); - bindingService.bindConsumer(new DirectChannel(), "input"); - try { - bindingService.bindProducer(new DirectChannel(), "output"); - fail("Expected 'Unknown binder configuration'"); - } - catch (IllegalStateException e) { - assertThat(e).hasMessageContaining("Unknown binder configuration: mockError"); - } - } - - @SuppressWarnings("unchecked") - @Test - public void testUnrecognizedBinderAllowedIfNotUsed() { - HashMap properties = new HashMap<>(); - properties.put("spring.cloud.stream.bindings.input.destination", "fooInput"); - properties.put("spring.cloud.stream.bindings.output.destination", "fooOutput"); - properties.put("spring.cloud.stream.defaultBinder", "mock1"); - properties.put("spring.cloud.stream.binders.mock1.type", "mock"); - properties.put("spring.cloud.stream.binders.kafka1.type", "kafka"); - BindingServiceProperties bindingServiceProperties = createBindingServiceProperties( - properties); - BinderFactory binderFactory = new BindingServiceConfiguration() - .binderFactory(createMockBinderTypeRegistry(), bindingServiceProperties, Mockito.mock(ObjectProvider.class)); - BindingService bindingService = new BindingService(bindingServiceProperties, - binderFactory, new ObjectMapper()); - bindingService.bindConsumer(new DirectChannel(), "input"); - bindingService.bindProducer(new DirectChannel(), "output"); - } - - @SuppressWarnings("unchecked") - @Test - public void testUnrecognizedBinderDisallowedIfUsed() { - HashMap properties = new HashMap<>(); - properties.put("spring.cloud.stream.bindings.input.destination", "fooInput"); - properties.put("spring.cloud.stream.bindings.input.binder", "mock1"); - properties.put("spring.cloud.stream.bindings.output.destination", "fooOutput"); - properties.put("spring.cloud.stream.bindings.output.type", "kafka1"); - properties.put("spring.cloud.stream.binders.mock1.type", "mock"); - properties.put("spring.cloud.stream.binders.kafka1.type", "kafka"); - BindingServiceProperties bindingServiceProperties = createBindingServiceProperties( - properties); - BinderFactory binderFactory = new BindingServiceConfiguration() - .binderFactory(createMockBinderTypeRegistry(), bindingServiceProperties, Mockito.mock(ObjectProvider.class)); - BindingService bindingService = new BindingService(bindingServiceProperties, - binderFactory, new ObjectMapper()); - bindingService.bindConsumer(new DirectChannel(), "input"); - try { - bindingService.bindProducer(new DirectChannel(), "output"); - fail("Expected 'Unknown binder configuration'"); - } - catch (IllegalArgumentException e) { - assertThat(e).hasMessageContaining("Binder type kafka is not defined"); - } - } - - @Test - public void testResolveBindableType() { - Class bindableType = GenericsUtils.getParameterType(FooBinder.class, - Binder.class, 0); - assertThat(bindableType).isSameAs(SomeBindableType.class); - } - - @SuppressWarnings({ "rawtypes", "unchecked" }) - @Test - @Ignore - public void testLateBindingConsumer() throws Exception { - BindingServiceProperties properties = new BindingServiceProperties(); - properties.setBindingRetryInterval(1); - Map bindingProperties = new HashMap<>(); - BindingProperties props = new BindingProperties(); - props.setDestination("foo"); - final String inputChannelName = "input"; - bindingProperties.put(inputChannelName, props); - properties.setBindings(bindingProperties); - DefaultBinderFactory binderFactory = createMockBinderFactory(); - Binder binder = binderFactory.getBinder("mock", MessageChannel.class); - ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); - scheduler.initialize(); - BindingService service = new BindingService(properties, binderFactory, scheduler, new ObjectMapper()); - MessageChannel inputChannel = new DirectChannel(); - final Binding mockBinding = Mockito.mock(Binding.class); - final CountDownLatch fail = new CountDownLatch(2); - doAnswer(i -> { - fail.countDown(); - if (fail.getCount() == 1) { - throw new RuntimeException("fail"); - } - return mockBinding; - }).when(binder).bindConsumer(eq("foo"), isNull(), same(inputChannel), - any(ConsumerProperties.class)); - Collection> bindings = service.bindConsumer(inputChannel, - inputChannelName); - assertThat(fail.await(10, TimeUnit.SECONDS)).isTrue(); - assertThat(bindings).hasSize(1); - Binding delegate = TestUtils - .getPropertyValue(bindings.iterator().next(), "delegate", Binding.class); - int n = 0; - while (n++ < 300 && delegate == null) { - Thread.sleep(400); - } - assertThat(delegate).isSameAs(mockBinding); - service.unbindConsumers(inputChannelName); - verify(binder, times(2)).bindConsumer(eq("foo"), isNull(), same(inputChannel), - any(ConsumerProperties.class)); - verify(delegate).unbind(); - binderFactory.destroy(); - } - - @SuppressWarnings({ "rawtypes", "unchecked" }) - @Test - public void testLateBindingProducer() throws Exception { - BindingServiceProperties properties = new BindingServiceProperties(); - properties.setBindingRetryInterval(1); - Map bindingProperties = new HashMap<>(); - BindingProperties props = new BindingProperties(); - props.setDestination("foo"); - final String outputChannelName = "output"; - bindingProperties.put(outputChannelName, props); - properties.setBindings(bindingProperties); - DefaultBinderFactory binderFactory = createMockBinderFactory(); - Binder binder = binderFactory.getBinder("mock", MessageChannel.class); - ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); - scheduler.initialize(); - BindingService service = new BindingService(properties, binderFactory, scheduler, new ObjectMapper()); - MessageChannel outputChannel = new DirectChannel(); - final Binding mockBinding = Mockito.mock(Binding.class); - final CountDownLatch fail = new CountDownLatch(2); - doAnswer(i -> { - fail.countDown(); - if (fail.getCount() == 1) { - throw new RuntimeException("fail"); - } - return mockBinding; - }).when(binder).bindProducer(eq("foo"), same(outputChannel), - any(ProducerProperties.class)); - Binding binding = service.bindProducer(outputChannel, - outputChannelName); - assertThat(fail.await(10, TimeUnit.SECONDS)).isTrue(); - assertThat(binding).isNotNull(); - Binding delegate = TestUtils.getPropertyValue(binding, "delegate", Binding.class); - int n = 0; - while (n++ < 300 && delegate == null) { - Thread.sleep(100); - delegate = TestUtils.getPropertyValue(binding, "delegate", Binding.class); - } - assertThat(delegate).isSameAs(mockBinding); - service.unbindProducers(outputChannelName); - verify(binder, times(2)).bindProducer(eq("foo"), same(outputChannel), - any(ProducerProperties.class)); - verify(delegate).unbind(); - binderFactory.destroy(); - scheduler.destroy(); - } - - @SuppressWarnings("unchecked") - @Test - public void testBindingAutostartup() throws Exception { - ApplicationContext context = new SpringApplicationBuilder(FooConfiguration.class) - .web(WebApplicationType.NONE).run("--spring.jmx.enabled=false", - "--spring.cloud.stream.bindings.input.consumer.auto-startup=false"); - BindingService bindingService = context.getBean(BindingService.class); - - Field cbField = ReflectionUtils.findField(BindingService.class, - "consumerBindings"); - cbField.setAccessible(true); - Map cbMap = (Map) cbField.get(bindingService); - Binding inputBinding = ((List>) cbMap.get("input")).get(0); - assertThat(inputBinding.isRunning()).isFalse(); - } - - private DefaultBinderFactory createMockBinderFactory() { - BinderTypeRegistry binderTypeRegistry = createMockBinderTypeRegistry(); - return new DefaultBinderFactory( - Collections.singletonMap("mock", - new BinderConfiguration("mock", new HashMap<>(), true, true)), - binderTypeRegistry, null); - } - - private DefaultBinderTypeRegistry createMockBinderTypeRegistry() { - return new DefaultBinderTypeRegistry(Collections.singletonMap("mock", - new BinderType("mock", new Class[] { MockBinderConfiguration.class }))); - } - - private BindingServiceProperties createBindingServiceProperties( - HashMap properties) { - BindingServiceProperties bindingServiceProperties = new BindingServiceProperties(); - org.springframework.boot.context.properties.bind.Binder propertiesBinder; - propertiesBinder = new org.springframework.boot.context.properties.bind.Binder( - new MapConfigurationPropertySource(properties)); - propertiesBinder.bind("spring.cloud.stream", - org.springframework.boot.context.properties.bind.Bindable - .ofInstance(bindingServiceProperties)); - return bindingServiceProperties; - } - - public interface FooBinding { - - @Input("input1") - SubscribableChannel in1(); - - @Input("input2") - SubscribableChannel in2(); - - @Output("output1") - MessageChannel out1(); - - @Output("output2") - MessageChannel out2(); - - @Input("inputXyz") - SubscribableChannel inXyz(); - - @Input("inputFooBar") - SubscribableChannel inFooBar(); - - @Input("inputFooBarBuzz") - SubscribableChannel inFooBarBuzz(); - - @Input("input_snake_case") - SubscribableChannel inWithSnakeCase(); - - } - - @EnableBinding(FooBinding.class) - @EnableAutoConfiguration - public static class DefaultConsumerPropertiesTestSink { - - @Bean - public Binder binder() { - return Mockito.mock(Binder.class, - Mockito.withSettings().defaultAnswer(Mockito.RETURNS_MOCKS)); - } - - } - - @EnableBinding(Sink.class) - @Import(TestChannelBinderConfiguration.class) - @EnableAutoConfiguration - public static class FooConfiguration { - - @ServiceActivator(inputChannel = Processor.INPUT) - public void echo(Message value) throws Exception { - } - - } - - public static class FooBinder - implements Binder { - - @Override - public Binding bindConsumer(String name, String group, - SomeBindableType inboundBindTarget, - ConsumerProperties consumerProperties) { - throw new UnsupportedOperationException(); - } - - @Override - public Binding bindProducer(String name, - SomeBindableType outboundBindTarget, - ProducerProperties producerProperties) { - throw new UnsupportedOperationException(); - } - - } - - public static class SomeBindableType { - - } - -} diff --git a/spring-cloud-stream/deleted-tests/FooChannels.java b/spring-cloud-stream/deleted-tests/FooChannels.java deleted file mode 100644 index 1f5273520..000000000 --- a/spring-cloud-stream/deleted-tests/FooChannels.java +++ /dev/null @@ -1,40 +0,0 @@ -/* - * Copyright 2015-2019 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 - * - * https://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.cloud.stream.annotation.Input; -import org.springframework.cloud.stream.annotation.Output; -import org.springframework.messaging.MessageChannel; - -/** - * @author Marius Bogoevici - */ -public interface FooChannels { - - @Input - MessageChannel foo(); - - @Input - MessageChannel bar(); - - @Output - MessageChannel baz(); - - @Output - MessageChannel qux(); - -} diff --git a/spring-cloud-stream/deleted-tests/InvalidBindingConfigurationTests.java b/spring-cloud-stream/deleted-tests/InvalidBindingConfigurationTests.java deleted file mode 100644 index 7af63231f..000000000 --- a/spring-cloud-stream/deleted-tests/InvalidBindingConfigurationTests.java +++ /dev/null @@ -1,66 +0,0 @@ -/* - * Copyright 2017-2019 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 - * - * https://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.binding; - -import org.junit.Ignore; -import org.junit.Test; - -import org.springframework.beans.factory.BeanDefinitionStoreException; -import org.springframework.boot.SpringApplication; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.annotation.Input; -import org.springframework.cloud.stream.annotation.Output; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.SubscribableChannel; - -import static org.assertj.core.api.Assertions.assertThatThrownBy; - -/** - * @author Artem Bilan - * @since 1.3 - */ -public class InvalidBindingConfigurationTests { - - @Test - @Ignore - public void testDuplicateBeanByBindingConfig() { - assertThatThrownBy(() -> SpringApplication.run(TestBindingConfig.class)) - .isInstanceOf(BeanDefinitionStoreException.class) - .hasMessageContaining("bean definition with this name already exists") - .hasMessageContaining(TestInvalidBinding.NAME).hasNoCause(); - } - - public interface TestInvalidBinding { - - String NAME = "testName"; - - @Input(NAME) - SubscribableChannel in(); - - @Output(NAME) - MessageChannel out(); - - } - - @EnableBinding(TestInvalidBinding.class) - @EnableAutoConfiguration - public static class TestBindingConfig { - - } - -} diff --git a/spring-cloud-stream/deleted-tests/ProcessorBindingWithBindingTargetsTests.java b/spring-cloud-stream/deleted-tests/ProcessorBindingWithBindingTargetsTests.java deleted file mode 100644 index b43760c38..000000000 --- a/spring-cloud-stream/deleted-tests/ProcessorBindingWithBindingTargetsTests.java +++ /dev/null @@ -1,69 +0,0 @@ -/* - * 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 - * - * https://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.junit.Test; -import org.junit.runner.RunWith; -import org.mockito.Mockito; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.messaging.Processor; -import org.springframework.context.annotation.PropertySource; -import org.springframework.messaging.MessageChannel; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; - -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.ArgumentMatchers.isNull; -import static org.mockito.Mockito.verify; - -/** - * @author Marius Bogoevici - * @author Janne Valkealahti - */ -@RunWith(SpringJUnit4ClassRunner.class) -// @checkstyle:off -@SpringBootTest(classes = ProcessorBindingWithBindingTargetsTests.TestProcessor.class, properties = "spring.cloud.stream.defaultBinder=mock") -// @checkstyle:on -public class ProcessorBindingWithBindingTargetsTests { - - @Autowired - private BinderFactory binderFactory; - - @Autowired - private Processor testProcessor; - - @SuppressWarnings({ "rawtypes", "unchecked" }) - @Test - public void testSourceOutputChannelBound() { - final Binder binder = this.binderFactory.getBinder(null, MessageChannel.class); - verify(binder).bindConsumer(eq("testtock.0"), isNull(), - eq(this.testProcessor.input()), Mockito.any()); - verify(binder).bindProducer(eq("testtock.1"), eq(this.testProcessor.output()), - Mockito.any()); - } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - @PropertySource("classpath:/org/springframework/cloud/stream/binder/processor-binding-test.properties") - public static class TestProcessor { - - } - -} diff --git a/spring-cloud-stream/deleted-tests/ProcessorBindingsWithDefaultsTests.java b/spring-cloud-stream/deleted-tests/ProcessorBindingsWithDefaultsTests.java deleted file mode 100644 index 759e087a9..000000000 --- a/spring-cloud-stream/deleted-tests/ProcessorBindingsWithDefaultsTests.java +++ /dev/null @@ -1,69 +0,0 @@ -/* - * 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 - * - * https://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.junit.Test; -import org.junit.runner.RunWith; -import org.mockito.Mockito; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.messaging.Processor; -import org.springframework.messaging.MessageChannel; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; - -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.ArgumentMatchers.isNull; -import static org.mockito.Mockito.verifyNoMoreInteractions; - -/** - * @author Marius Bogoevici - * @author Janne Valkealahti - */ -@RunWith(SpringJUnit4ClassRunner.class) -// @checkstyle:off -@SpringBootTest(classes = ProcessorBindingsWithDefaultsTests.TestProcessor.class, properties = "spring.cloud.stream.defaultBinder=mock") -public class ProcessorBindingsWithDefaultsTests { - - // @checkstyle:on - - @Autowired - private BinderFactory binderFactory; - - @Autowired - private Processor processor; - - @SuppressWarnings({ "rawtypes", "unchecked" }) - @Test - public void testSourceOutputChannelBound() { - Binder binder = this.binderFactory.getBinder(null, MessageChannel.class); - Mockito.verify(binder).bindConsumer(eq("input"), isNull(), - eq(this.processor.input()), Mockito.any()); - Mockito.verify(binder).bindProducer(eq("output"), eq(this.processor.output()), - Mockito.any()); - verifyNoMoreInteractions(binder); - } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class TestProcessor { - - } - -} diff --git a/spring-cloud-stream/deleted-tests/SinkBindingWithDefaultTargetsTests.java b/spring-cloud-stream/deleted-tests/SinkBindingWithDefaultTargetsTests.java deleted file mode 100644 index 89218c870..000000000 --- a/spring-cloud-stream/deleted-tests/SinkBindingWithDefaultTargetsTests.java +++ /dev/null @@ -1,70 +0,0 @@ -/* - * 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 - * - * https://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.junit.Test; -import org.junit.runner.RunWith; -import org.mockito.Mockito; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.messaging.Sink; -import org.springframework.context.annotation.PropertySource; -import org.springframework.messaging.MessageChannel; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; - -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.ArgumentMatchers.isNull; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.verifyNoMoreInteractions; - -/** - * @author Marius Bogoevici - * @author Janne Valkealahti - * @author Janne Valkealahti - */ -@RunWith(SpringJUnit4ClassRunner.class) -// @checkstyle:off -@SpringBootTest(classes = SinkBindingWithDefaultTargetsTests.TestSink.class, properties = "spring.cloud.stream.defaultBinder=mock") -// @checkstyle:on -public class SinkBindingWithDefaultTargetsTests { - - @Autowired - private BinderFactory binderFactory; - - @Autowired - private Sink testSink; - - @Test - @SuppressWarnings({ "rawtypes", "unchecked" }) - public void testSourceOutputChannelBound() { - Binder binder = this.binderFactory.getBinder(null, MessageChannel.class); - verify(binder).bindConsumer(eq("testtock"), isNull(), eq(this.testSink.input()), - Mockito.any()); - verifyNoMoreInteractions(binder); - } - - @EnableBinding(Sink.class) - @EnableAutoConfiguration - @PropertySource("classpath:/org/springframework/cloud/stream/binder/sink-binding-test.properties") - public static class TestSink { - - } - -} diff --git a/spring-cloud-stream/deleted-tests/SourceBindingWithBindingTargetsTests.java b/spring-cloud-stream/deleted-tests/SourceBindingWithBindingTargetsTests.java deleted file mode 100644 index ea4567742..000000000 --- a/spring-cloud-stream/deleted-tests/SourceBindingWithBindingTargetsTests.java +++ /dev/null @@ -1,75 +0,0 @@ -/* - * 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 - * - * https://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.junit.Test; -import org.junit.runner.RunWith; -import org.mockito.Mockito; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.messaging.Source; -import org.springframework.context.annotation.PropertySource; -import org.springframework.integration.channel.PublishSubscribeChannel; -import org.springframework.integration.context.IntegrationContextUtils; -import org.springframework.messaging.MessageChannel; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; - -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.verifyNoMoreInteractions; - -/** - * @author Marius Bogoevici - * @author Ilayaperumal Gopinathan - */ -@RunWith(SpringJUnit4ClassRunner.class) -// @checkstyle:off -@SpringBootTest(classes = SourceBindingWithBindingTargetsTests.TestSource.class, properties = "spring.cloud.stream.defaultBinder=mock") -// @checkstyle:on -public class SourceBindingWithBindingTargetsTests { - - @Autowired - private BinderFactory binderFactory; - - @Autowired - private Source testSource; - - @Autowired - @Qualifier(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME) - private PublishSubscribeChannel errorChannel; - - @Test - @SuppressWarnings({ "rawtypes", "unchecked" }) - public void testSourceOutputChannelBound() { - Binder binder = this.binderFactory.getBinder(null, MessageChannel.class); - verify(binder).bindProducer(eq("testtock"), eq(this.testSource.output()), - Mockito.any()); - verifyNoMoreInteractions(binder); - } - - @EnableBinding(Source.class) - @EnableAutoConfiguration - @PropertySource("classpath:/org/springframework/cloud/stream/binder/source-binding-test.properties") - public static class TestSource { - - } - -} diff --git a/spring-cloud-stream/deleted-tests/SourceBindingWithDefaultsTests.java b/spring-cloud-stream/deleted-tests/SourceBindingWithDefaultsTests.java deleted file mode 100644 index d6cedd37e..000000000 --- a/spring-cloud-stream/deleted-tests/SourceBindingWithDefaultsTests.java +++ /dev/null @@ -1,65 +0,0 @@ -/* - * Copyright 2015-2019 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 - * - * https://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.junit.Test; -import org.junit.runner.RunWith; -import org.mockito.Mockito; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.messaging.Source; -import org.springframework.messaging.MessageChannel; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; - -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.verifyNoMoreInteractions; - -/** - * @author Marius Bogoevici - */ -@RunWith(SpringJUnit4ClassRunner.class) -// @checkstyle:off -@SpringBootTest(classes = SourceBindingWithDefaultsTests.TestSource.class, properties = "spring.cloud.stream.defaultBinder=mock") -// @checkstyle:on -public class SourceBindingWithDefaultsTests { - - @Autowired - private BinderFactory binderFactory; - - @Autowired - private Source testSource; - - @SuppressWarnings({ "rawtypes", "unchecked" }) - @Test - public void testSourceOutputChannelBound() { - Binder binder = this.binderFactory.getBinder(null, MessageChannel.class); - verify(binder).bindProducer(eq("output"), eq(this.testSource.output()), - Mockito.any()); - verifyNoMoreInteractions(binder); - } - - @EnableBinding(Source.class) - @EnableAutoConfiguration - public static class TestSource { - - } - -} diff --git a/spring-cloud-stream/deleted-tests/SourceBindingWithGlobalPropertiesOnlyTest.java b/spring-cloud-stream/deleted-tests/SourceBindingWithGlobalPropertiesOnlyTest.java deleted file mode 100644 index 6631ab0ce..000000000 --- a/spring-cloud-stream/deleted-tests/SourceBindingWithGlobalPropertiesOnlyTest.java +++ /dev/null @@ -1,66 +0,0 @@ -/* - * 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 - * - * https://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.assertj.core.api.Assertions; -import org.junit.Test; -import org.junit.runner.RunWith; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; -import org.springframework.cloud.stream.config.BindingProperties; -import org.springframework.cloud.stream.config.BindingServiceProperties; -import org.springframework.cloud.stream.config.SpelExpressionConverterConfiguration; -import org.springframework.cloud.stream.messaging.Source; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; - -/** - * @author Marius Bogoevici - * @author Ilayaperumal Gopinathan - */ -@RunWith(SpringJUnit4ClassRunner.class) -@SpringBootTest(classes = { TestChannelBinderConfiguration.class, - SourceBindingWithGlobalPropertiesOnlyTest.TestSource.class, - SpelExpressionConverterConfiguration.class }, properties = { - "spring.cloud.stream.default.contentType=application/json", - "spring.cloud.stream.default.producer.partitionKeyExpression=key" }) -public class SourceBindingWithGlobalPropertiesOnlyTest { - - @Autowired - private BindingServiceProperties bindingServiceProperties; - - @Test - public void testGlobalPropertiesSet() { - BindingProperties bindingProperties = this.bindingServiceProperties - .getBindingProperties(Source.OUTPUT); - Assertions.assertThat(bindingProperties.getContentType()) - .isEqualTo("application/json"); - Assertions.assertThat(bindingProperties.getProducer()).isNotNull(); - Assertions.assertThat(bindingProperties.getProducer().getPartitionKeyExpression() - .getExpressionString()).isEqualTo("key"); - } - - @EnableBinding(Source.class) - @EnableAutoConfiguration - public static class TestSource { - - } - -} diff --git a/spring-cloud-stream/deleted-tests/SourceBindingWithGlobalPropertiesTest.java b/spring-cloud-stream/deleted-tests/SourceBindingWithGlobalPropertiesTest.java deleted file mode 100644 index bd4fccd84..000000000 --- a/spring-cloud-stream/deleted-tests/SourceBindingWithGlobalPropertiesTest.java +++ /dev/null @@ -1,73 +0,0 @@ -/* - * 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 - * - * https://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.assertj.core.api.Assertions; -import org.junit.Test; -import org.junit.runner.RunWith; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.config.BindingProperties; -import org.springframework.cloud.stream.config.BindingServiceProperties; -import org.springframework.cloud.stream.messaging.Source; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; - -/** - * @author Marius Bogoevici - * @author Ilayaperumal Gopinathan - * @author Gary Russell - * @author Oleg Zhurakousky - */ -@RunWith(SpringJUnit4ClassRunner.class) -@SpringBootTest(classes = SourceBindingWithGlobalPropertiesTest.TestSource.class, properties = { - "spring.cloud.stream.default.contentType=application/json", - "spring.cloud.stream.bindings.output.destination=ticktock", - "spring.cloud.stream.default.producer.requiredGroups=someGroup", - "spring.cloud.stream.default.producer.partitionCount=1", - "spring.cloud.stream.bindings.output.producer.headerMode=none", - "spring.cloud.stream.bindings.output.producer.partitionCount=4", - "spring.cloud.stream.defaultBinder=mock" }) -public class SourceBindingWithGlobalPropertiesTest { - - @Autowired - private BindingServiceProperties serviceProperties; - - @Test - public void testGlobalPropertiesSet() { - BindingProperties bindingProperties = this.serviceProperties - .getBindingProperties(Source.OUTPUT); - Assertions.assertThat(bindingProperties.getContentType()) - .isEqualTo("application/json"); - Assertions.assertThat(bindingProperties.getDestination()).isEqualTo("ticktock"); - Assertions.assertThat(bindingProperties.getProducer().getRequiredGroups()) - .containsExactly("someGroup"); // default propagates to producer - Assertions.assertThat(bindingProperties.getProducer().getPartitionCount()) - .isEqualTo(4); // validates binding property takes precedence over default - Assertions.assertThat(bindingProperties.getProducer().getHeaderMode()) - .isEqualTo(HeaderMode.none); - } - - @EnableBinding(Source.class) - @EnableAutoConfiguration - public static class TestSource { - - } - -} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/CustomPartitionedProducerTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/CustomPartitionedProducerTest.java new file mode 100644 index 000000000..271b04fa1 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/CustomPartitionedProducerTest.java @@ -0,0 +1,253 @@ +/* + * Copyright 2017-2018 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 + * + * https://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.binding; + +import java.lang.reflect.Field; + +import org.junit.Test; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.cloud.stream.binder.PartitionHandler; +import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy; +import org.springframework.cloud.stream.binder.PartitionSelectorStrategy; +import org.springframework.cloud.stream.partitioning.CustomPartitionKeyExtractorClass; +import org.springframework.cloud.stream.partitioning.CustomPartitionSelectorClass; +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.PropertySource; +import org.springframework.integration.annotation.InboundChannelAdapter; +import org.springframework.integration.annotation.Poller; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.core.MessageSource; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.support.ChannelInterceptor; +import org.springframework.util.ReflectionUtils; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Ilayaperumal Gopinathan + * @author Oleg Zhurakousky + */ +public class CustomPartitionedProducerTest { + + @Test + public void testCustomPartitionedProducer() { + ApplicationContext context = SpringApplication.run( + CustomPartitionedProducerTest.TestSource.class, + "--spring.cloud.stream.output-bindings=output", + "--spring.jmx.enabled=false", "--spring.main.web-application-type=none", + "--spring.cloud.stream.bindings.output.producer.partitionKeyExtractorClass=" + + "org.springframework.cloud.stream.partitioning.CustomPartitionKeyExtractorClass", + "--spring.cloud.stream.bindings.output.producer.partitionSelectorClass=" + + "org.springframework.cloud.stream.partitioning.CustomPartitionSelectorClass", + "--spring.cloud.stream.default-binder=mock"); + DirectChannel messageChannel = context.getBean("output-out-0", DirectChannel.class); + for (ChannelInterceptor channelInterceptor : messageChannel + .getInterceptors()) { + if (channelInterceptor instanceof MessageConverterConfigurer.PartitioningInterceptor) { + Field partitionHandlerField = ReflectionUtils.findField( + MessageConverterConfigurer.PartitioningInterceptor.class, + "partitionHandler"); + ReflectionUtils.makeAccessible(partitionHandlerField); + PartitionHandler partitionHandler = (PartitionHandler) ReflectionUtils + .getField(partitionHandlerField, channelInterceptor); + Field partitonKeyExtractorField = ReflectionUtils.findField( + PartitionHandler.class, "partitionKeyExtractorStrategy"); + ReflectionUtils.makeAccessible(partitonKeyExtractorField); + Field partitonSelectorField = ReflectionUtils + .findField(PartitionHandler.class, "partitionSelectorStrategy"); + ReflectionUtils.makeAccessible(partitonSelectorField); + assertThat(((PartitionKeyExtractorStrategy) ReflectionUtils + .getField(partitonKeyExtractorField, partitionHandler)).getClass() + .equals(CustomPartitionKeyExtractorClass.class)).isTrue(); + assertThat(((PartitionSelectorStrategy) ReflectionUtils + .getField(partitonSelectorField, partitionHandler)).getClass() + .equals(CustomPartitionSelectorClass.class)).isTrue(); + } + } + } + + @Test + public void testCustomPartitionedProducerByName() { + ApplicationContext context = SpringApplication.run( + CustomPartitionedProducerTest.TestSource.class, + "--spring.cloud.stream.output-bindings=output", + "--spring.jmx.enabled=false", "--spring.main.web-application-type=none", + "--spring.cloud.stream.bindings.output.producer.partitionKeyExtractorName=customPartitionKeyExtractor", + "--spring.cloud.stream.bindings.output.producer.partitionSelectorName=customPartitionSelector", + "--spring.cloud.stream.default-binder=mock"); + DirectChannel messageChannel = context.getBean("output-out-0", DirectChannel.class); + for (ChannelInterceptor channelInterceptor : messageChannel + .getInterceptors()) { + if (channelInterceptor instanceof MessageConverterConfigurer.PartitioningInterceptor) { + Field partitionHandlerField = ReflectionUtils.findField( + MessageConverterConfigurer.PartitioningInterceptor.class, + "partitionHandler"); + ReflectionUtils.makeAccessible(partitionHandlerField); + PartitionHandler partitionHandler = (PartitionHandler) ReflectionUtils + .getField(partitionHandlerField, channelInterceptor); + Field partitonKeyExtractorField = ReflectionUtils.findField( + PartitionHandler.class, "partitionKeyExtractorStrategy"); + ReflectionUtils.makeAccessible(partitonKeyExtractorField); + Field partitonSelectorField = ReflectionUtils + .findField(PartitionHandler.class, "partitionSelectorStrategy"); + ReflectionUtils.makeAccessible(partitonSelectorField); + assertThat(((PartitionKeyExtractorStrategy) ReflectionUtils + .getField(partitonKeyExtractorField, partitionHandler)).getClass() + .equals(CustomPartitionKeyExtractorClass.class)).isTrue(); + assertThat(((PartitionSelectorStrategy) ReflectionUtils + .getField(partitonSelectorField, partitionHandler)).getClass() + .equals(CustomPartitionSelectorClass.class)).isTrue(); + } + } + } + + @Test + public void testCustomPartitionedProducerAsSingletons() { + ApplicationContext context = SpringApplication.run( + CustomPartitionedProducerTest.TestSource.class, + "--spring.cloud.stream.output-bindings=output", + "--spring.jmx.enabled=false", "--spring.main.web-application-type=none", + "--spring.cloud.stream.default-binder=mock"); + DirectChannel messageChannel = context.getBean("output-out-0", DirectChannel.class); + for (ChannelInterceptor channelInterceptor : messageChannel + .getInterceptors()) { + if (channelInterceptor instanceof MessageConverterConfigurer.PartitioningInterceptor) { + Field partitionHandlerField = ReflectionUtils.findField( + MessageConverterConfigurer.PartitioningInterceptor.class, + "partitionHandler"); + ReflectionUtils.makeAccessible(partitionHandlerField); + PartitionHandler partitionHandler = (PartitionHandler) ReflectionUtils + .getField(partitionHandlerField, channelInterceptor); + Field partitonKeyExtractorField = ReflectionUtils.findField( + PartitionHandler.class, "partitionKeyExtractorStrategy"); + ReflectionUtils.makeAccessible(partitonKeyExtractorField); + Field partitonSelectorField = ReflectionUtils + .findField(PartitionHandler.class, "partitionSelectorStrategy"); + ReflectionUtils.makeAccessible(partitonSelectorField); + assertThat(((PartitionKeyExtractorStrategy) ReflectionUtils + .getField(partitonKeyExtractorField, partitionHandler)).getClass() + .equals(CustomPartitionKeyExtractorClass.class)).isTrue(); + assertThat(((PartitionSelectorStrategy) ReflectionUtils + .getField(partitonSelectorField, partitionHandler)).getClass() + .equals(CustomPartitionSelectorClass.class)).isTrue(); + } + } + } + + public void testCustomPartitionedProducerMultipleInstances() { + ApplicationContext context = SpringApplication.run( + CustomPartitionedProducerTest.TestSourceMultipleStrategies.class, + "--spring.jmx.enabled=false", "--spring.main.web-application-type=none", + "--spring.cloud.stream.bindings.output.producer.partitionKeyExtractorName=customPartitionKeyExtractorOne", + "--spring.cloud.stream.bindings.output.producer.partitionSelectorName=customPartitionSelectorTwo", + "--spring.cloud.stream.default-binder=mock"); + DirectChannel messageChannel = context.getBean("output-out-0", DirectChannel.class); + for (ChannelInterceptor channelInterceptor : messageChannel + .getInterceptors()) { + if (channelInterceptor instanceof MessageConverterConfigurer.PartitioningInterceptor) { + Field partitionHandlerField = ReflectionUtils.findField( + MessageConverterConfigurer.PartitioningInterceptor.class, + "partitionHandler"); + ReflectionUtils.makeAccessible(partitionHandlerField); + PartitionHandler partitionHandler = (PartitionHandler) ReflectionUtils + .getField(partitionHandlerField, channelInterceptor); + Field partitonKeyExtractorField = ReflectionUtils.findField( + PartitionHandler.class, "partitionKeyExtractorStrategy"); + ReflectionUtils.makeAccessible(partitonKeyExtractorField); + Field partitonSelectorField = ReflectionUtils + .findField(PartitionHandler.class, "partitionSelectorStrategy"); + ReflectionUtils.makeAccessible(partitonSelectorField); + assertThat(((PartitionKeyExtractorStrategy) ReflectionUtils + .getField(partitonKeyExtractorField, partitionHandler)).getClass() + .equals(CustomPartitionKeyExtractorClass.class)).isTrue(); + assertThat(((PartitionSelectorStrategy) ReflectionUtils + .getField(partitonSelectorField, partitionHandler)).getClass() + .equals(CustomPartitionSelectorClass.class)).isTrue(); + } + } + } + + @EnableAutoConfiguration + @PropertySource("classpath:/org/springframework/cloud/stream/binder/custom-partitioned-producer-test.properties") + public static class TestSource { + + @Bean + public CustomPartitionSelectorClass customPartitionSelector() { + return new CustomPartitionSelectorClass(); + } + + @Bean + public CustomPartitionKeyExtractorClass customPartitionKeyExtractor() { + return new CustomPartitionKeyExtractorClass(); + } + + @Bean + @InboundChannelAdapter(value = "output-out-0", poller = @Poller(fixedDelay = "5000", maxMessagesPerPoll = "1")) + public MessageSource timerMessageSource() { + return new MessageSource() { + @Override + public Message receive() { + throw new MessagingException("test"); + } + }; + } + + } + + + @EnableAutoConfiguration + @PropertySource("classpath:/org/springframework/cloud/stream/binder/custom-partitioned-producer-test.properties") + public static class TestSourceMultipleStrategies { + + @Bean + public CustomPartitionSelectorClass customPartitionSelectorOne() { + return new CustomPartitionSelectorClass(); + } + + @Bean + public CustomPartitionSelectorClass customPartitionSelectorTwo() { + return new CustomPartitionSelectorClass(); + } + + @Bean + public CustomPartitionKeyExtractorClass customPartitionKeyExtractorOne() { + return new CustomPartitionKeyExtractorClass(); + } + + @Bean + public CustomPartitionKeyExtractorClass customPartitionKeyExtractorTwo() { + return new CustomPartitionKeyExtractorClass(); + } + + @Bean + @InboundChannelAdapter(value = "output-out-0", poller = @Poller(fixedDelay = "5000", maxMessagesPerPoll = "1")) + public MessageSource timerMessageSource() { + return new MessageSource() { + @Override + public Message receive() { + throw new MessagingException("test"); + } + }; + } + + } + +}