Re-enable CustomPartitionedProducerTest
This commit is contained in:
Oleg Zhurakousky
@@ -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 {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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";
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<String, BindingProperties> 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<MessageChannel> mockBinding = Mockito.mock(Binding.class);
|
||||
when(binder.bindConsumer(eq("foo"), isNull(), same(inputChannel),
|
||||
any(ConsumerProperties.class))).thenReturn(mockBinding);
|
||||
Collection<Binding<MessageChannel>> bindings = service.bindConsumer(inputChannel,
|
||||
inputChannelName);
|
||||
assertThat(bindings).hasSize(1);
|
||||
Binding<MessageChannel> 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<String, BindingProperties> 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<MessageChannel> mockBinding1 = Mockito.mock(Binding.class);
|
||||
Binding<MessageChannel> 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<Binding<MessageChannel>> bindings = service.bindConsumer(inputChannel,
|
||||
"input");
|
||||
assertThat(bindings).hasSize(2);
|
||||
|
||||
Iterator<Binding<MessageChannel>> iterator = bindings.iterator();
|
||||
Binding<MessageChannel> binding1 = iterator.next();
|
||||
Binding<MessageChannel> 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<String, BindingProperties> 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<MessageChannel> mockBinding1 = Mockito.mock(Binding.class, "FirstBinding");
|
||||
Binding<MessageChannel> mockBinding2 = Mockito.mock(Binding.class, "SecondBinding");
|
||||
|
||||
ArgumentCaptor<ConsumerProperties> captor = ArgumentCaptor.forClass(ConsumerProperties.class);
|
||||
|
||||
when(binder.bindConsumer(eq("foo"), isNull(), same(inputChannel),
|
||||
any(ConsumerProperties.class))).thenReturn(mockBinding1).thenReturn(mockBinding2);
|
||||
|
||||
Collection<Binding<MessageChannel>> bindings = service.bindConsumer(inputChannel,
|
||||
"input");
|
||||
assertThat(bindings).hasSize(2);
|
||||
|
||||
Iterator<Binding<MessageChannel>> iterator = bindings.iterator();
|
||||
Binding<MessageChannel> binding1 = iterator.next();
|
||||
Binding<MessageChannel> 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<ConsumerProperties> 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<String, BindingProperties> 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<MessageChannel> mockBinding1 = Mockito.mock(Binding.class);
|
||||
|
||||
when(binder.bindConsumer(eq("foo,bar"), isNull(), same(inputChannel),
|
||||
any(ConsumerProperties.class))).thenReturn(mockBinding1);
|
||||
|
||||
Collection<Binding<MessageChannel>> bindings = service.bindConsumer(inputChannel,
|
||||
"input");
|
||||
assertThat(bindings).hasSize(1);
|
||||
|
||||
Iterator<Binding<MessageChannel>> iterator = bindings.iterator();
|
||||
Binding<MessageChannel> 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<String, BindingProperties> 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<MessageChannel> mockBinding = Mockito.mock(Binding.class);
|
||||
when(binder.bindConsumer(eq("foo"), eq("fooGroup"), same(inputChannel),
|
||||
any(ConsumerProperties.class))).thenReturn(mockBinding);
|
||||
Collection<Binding<MessageChannel>> bindings = service.bindConsumer(inputChannel,
|
||||
inputChannelName);
|
||||
assertThat(bindings).hasSize(1);
|
||||
Binding<MessageChannel> 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<MessageChannel> mockBinding = Mockito.mock(Binding.class);
|
||||
// final AtomicReference<MessageChannel> 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 <T> Binder<T, ?, ?> getBinder(String channelName,
|
||||
// Class<T> 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<Void>() {
|
||||
//
|
||||
// @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<Object>() {
|
||||
//
|
||||
// @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<ProducerProperties> 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<String, BindingProperties> 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<String, BindingProperties> 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<String, BindingProperties> 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<String, String> 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<String, String> 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<String, String> 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<String, BindingProperties> 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<MessageChannel> 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<Binding<MessageChannel>> bindings = service.bindConsumer(inputChannel,
|
||||
inputChannelName);
|
||||
assertThat(fail.await(10, TimeUnit.SECONDS)).isTrue();
|
||||
assertThat(bindings).hasSize(1);
|
||||
Binding<MessageChannel> 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<String, BindingProperties> 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<MessageChannel> 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<MessageChannel> 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<String, Object> cbMap = (Map<String, Object>) cbField.get(bindingService);
|
||||
Binding<?> inputBinding = ((List<Binding<?>>) 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<String, String> 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<SomeBindableType, ConsumerProperties, ProducerProperties> {
|
||||
|
||||
@Override
|
||||
public Binding<SomeBindableType> bindConsumer(String name, String group,
|
||||
SomeBindableType inboundBindTarget,
|
||||
ConsumerProperties consumerProperties) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Binding<SomeBindableType> bindProducer(String name,
|
||||
SomeBindableType outboundBindTarget,
|
||||
ProducerProperties producerProperties) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public static class SomeBindableType {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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();
|
||||
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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.<ConsumerProperties>any());
|
||||
verifyNoMoreInteractions(binder);
|
||||
}
|
||||
|
||||
@EnableBinding(Sink.class)
|
||||
@EnableAutoConfiguration
|
||||
@PropertySource("classpath:/org/springframework/cloud/stream/binder/sink-binding-test.properties")
|
||||
public static class TestSink {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<String> timerMessageSource() {
|
||||
return new MessageSource<String>() {
|
||||
@Override
|
||||
public Message<String> 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<String> timerMessageSource() {
|
||||
return new MessageSource<String>() {
|
||||
@Override
|
||||
public Message<String> receive() {
|
||||
throw new MessagingException("test");
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user