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

Refactoring and polishing

Browse Source 
- Refactored Input/OutputBindingLifecycle to ensure they properly use Spring's dependency injection mechanisms
    - Refactored additional tests to use new SI-backed binder

polishing
This commit is contained in:
Oleg Zhurakousky
2017-11-25 08:02:01 -05:00
parent 5a87211237
commit 2bac583d81
7 changed files with 162 additions and 319 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindingLifecycle.java Normal file
View File

@@ -0,0 +1,81 @@
/*
* Copyright 2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binding;
import java.util.Map;
import org.springframework.context.SmartLifecycle;
/**
* Base implementation of lifecycle operations for {@link BindingService}
* aware {@link Bindable}s.
*
* @see InputBindingLifecycle
* @see OutputBindingLifecycle
*
* @author Oleg Zhurakousky
*/
abstract class AbstractBindingLifecycle implements SmartLifecycle {
final BindingService bindingService;
private final Map<String, Bindable> bindables;
private volatile boolean running;
AbstractBindingLifecycle(BindingService bindingService, Map<String, Bindable> bindables) {
this.bindingService = bindingService;
this.bindables = bindables;
}
@Override
public void start() {
if (!running) {
bindables.values().stream().forEach(this::doStartWithBindable);
this.running = true;
}
}
@Override
public void stop() {
if (running) {
bindables.values().stream().forEach(this::doStopWithBindable);
this.running = false;
}
}
@Override
public boolean isRunning() {
return running;
}
@Override
public boolean isAutoStartup() {
return true;
}
@Override
public void stop(Runnable callback) {
stop();
if (callback != null) {
callback.run();
}
}
abstract void doStartWithBindable(Bindable bindable);
abstract void doStopWithBindable(Bindable bindable);
}

92
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/InputBindingLifecycle.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2016 the original author or authors.
* 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.
@@ -18,89 +18,17 @@ package org.springframework.cloud.stream.binding;
import java.util.Map;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.SmartLifecycle;
/**
* Coordinates binding/unbinding of input binding targets in accordance to the lifecycle
* of the host context.
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
* @author Oleg Zhurakousky
*/
public class InputBindingLifecycle implements SmartLifecycle, ApplicationContextAware {
public class InputBindingLifecycle extends AbstractBindingLifecycle {
private volatile boolean running;
private ConfigurableApplicationContext applicationContext;
@Override
public void setApplicationContext(ApplicationContext applicationContext)
throws BeansException {
this.applicationContext = (ConfigurableApplicationContext) applicationContext;
}
@Override
public void start() {
if (!running) {
// retrieve the BindingService lazily, avoiding early initialization
try {
BindingService bindingService = this.applicationContext
.getBean(BindingService.class);
Map<String, Bindable> bindables = this.applicationContext
.getBeansOfType(Bindable.class);
for (Bindable bindable : bindables.values()) {
bindable.bindInputs(bindingService);
}
}
catch (BeansException e) {
throw new IllegalStateException(
"Cannot perform binding, no proper implementation found", e);
}
this.running = true;
}
}
@Override
public void stop() {
if (running) {
try {
// retrieve the BindingService lazily, avoiding early
// initialization
BindingService bindingService = this.applicationContext
.getBean(BindingService.class);
Map<String, Bindable> bindables = this.applicationContext
.getBeansOfType(Bindable.class);
for (Bindable bindable : bindables.values()) {
bindable.unbindInputs(bindingService);
}
}
catch (BeansException e) {
throw new IllegalStateException(
"Cannot perform unbinding, no proper implementation found", e);
}
this.running = false;
}
}
@Override
public boolean isRunning() {
return running;
}
@Override
public boolean isAutoStartup() {
return true;
}
@Override
public void stop(Runnable callback) {
stop();
if (callback != null) {
callback.run();
}
public InputBindingLifecycle(BindingService bindingService, Map<String, Bindable> bindables) {
super(bindingService, bindables);
}
/**
@@ -111,4 +39,14 @@ public class InputBindingLifecycle implements SmartLifecycle, ApplicationContext
public int getPhase() {
return Integer.MAX_VALUE - 1000;
}
@Override
void doStartWithBindable(Bindable bindable) {
bindable.bindInputs(bindingService);
}
@Override
void doStopWithBindable(Bindable bindable) {
bindable.unbindInputs(bindingService);
}
}

93
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/OutputBindingLifecycle.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015 the original author or authors.
* 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.
@@ -18,91 +18,18 @@ package org.springframework.cloud.stream.binding;
import java.util.Map;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.SmartLifecycle;
/**
* Coordinates binding/unbinding of output binding targets in accordance to the lifecycle
* of the host context.
*
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
* @author Oleg Zhurakousky
*/
public class OutputBindingLifecycle implements SmartLifecycle, ApplicationContextAware {
public class OutputBindingLifecycle extends AbstractBindingLifecycle {
private volatile boolean running;
private ConfigurableApplicationContext applicationContext;
@Override
public void setApplicationContext(ApplicationContext applicationContext)
throws BeansException {
this.applicationContext = (ConfigurableApplicationContext) applicationContext;
}
@Override
public void start() {
if (!running) {
// retrieve the BindingService lazily, avoiding early initialization
try {
BindingService bindingService = this.applicationContext
.getBean(BindingService.class);
Map<String, Bindable> bindables = this.applicationContext
.getBeansOfType(Bindable.class);
for (Bindable bindable : bindables.values()) {
bindable.bindOutputs(bindingService);
}
}
catch (BeansException e) {
throw new IllegalStateException(
"Cannot perform binding, no proper implementation found", e);
}
this.running = true;
}
}
@Override
public void stop() {
if (running) {
try {
// retrieve the BindingService lazily, avoiding early
// initialization
BindingService bindingService = this.applicationContext
.getBean(BindingService.class);
Map<String, Bindable> bindables = this.applicationContext
.getBeansOfType(Bindable.class);
for (Bindable bindable : bindables.values()) {
bindable.unbindOutputs(bindingService);
}
}
catch (BeansException e) {
throw new IllegalStateException(
"Cannot perform unbinding, no proper implementation found", e);
}
this.running = false;
}
}
@Override
public boolean isRunning() {
return running;
}
@Override
public boolean isAutoStartup() {
return true;
}
@Override
public void stop(Runnable callback) {
stop();
if (callback != null) {
callback.run();
}
public OutputBindingLifecycle(BindingService bindingService, Map<String, Bindable> bindables) {
super(bindingService, bindables);
}
/**
@@ -113,4 +40,14 @@ public class OutputBindingLifecycle implements SmartLifecycle, ApplicationContex
public int getPhase() {
return Integer.MIN_VALUE + 1000;
}
@Override
void doStartWithBindable(Bindable bindable) {
bindable.bindOutputs(bindingService);
}
@Override
void doStopWithBindable(Bindable bindable) {
bindable.unbindOutputs(bindingService);
}
}

9
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java
View File

@@ -32,6 +32,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.binding.AbstractBindingTargetFactory;
import org.springframework.cloud.stream.binding.Bindable;
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
import org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor;
import org.springframework.cloud.stream.binding.BindingService;
@@ -136,14 +137,14 @@ public class BindingServiceConfiguration {
@Bean
@DependsOn("bindingService")
public OutputBindingLifecycle outputBindingLifecycle() {
return new OutputBindingLifecycle();
public OutputBindingLifecycle outputBindingLifecycle(BindingService bindingService, Map<String, Bindable> bindables) {
return new OutputBindingLifecycle(bindingService, bindables);
}
@Bean
@DependsOn("bindingService")
public InputBindingLifecycle inputBindingLifecycle() {
return new InputBindingLifecycle();
public InputBindingLifecycle inputBindingLifecycle(BindingService bindingService, Map<String, Bindable> bindables) {
return new InputBindingLifecycle(bindingService, bindables);
}
@Bean

199
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinderTests.java
View File

@@ -19,22 +19,18 @@ package org.springframework.cloud.stream.binder;
import java.util.Iterator;
import java.util.Set;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.ErrorInfrastructure;
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
import org.springframework.cloud.stream.provisioning.ProducerDestination;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.stream.binder.integration.SpringIntegrationBinderConfiguration;
import org.springframework.cloud.stream.provisioning.ProvisioningProvider;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.Lifecycle;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.handler.BridgeHandler;
import org.springframework.integration.test.util.TestUtils;
@@ -43,36 +39,43 @@ import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.SubscribableChannel;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.mock;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
/**
* @author Marius Bogoevici
* @author Gary Russell
* @author Oleg Zhurakousky
*
* @since 1.2.2
*/
public class AbstractMessageChannelBinderTests {
private ApplicationContext context;
@Before
public void prepare() {
this.context = new SpringApplicationBuilder(SpringIntegrationBinderConfiguration.getCompleteConfiguration())
.web(WebApplicationType.NONE)
.run();
}
@Test
public void testEndpointLifecycle() throws Exception {
StubMessageChannelBinder binder = new StubMessageChannelBinder();
GenericApplicationContext context = new GenericApplicationContext();
context.refresh();
context.getBeanFactory().registerSingleton("errorChannel", new PublishSubscribeChannel());
binder.setApplicationContext(context);
AbstractMessageChannelBinder<ConsumerProperties, ProducerProperties, ProvisioningProvider<ConsumerProperties, ProducerProperties>> binder =
context.getBean(AbstractMessageChannelBinder.class);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo", "fooGroup", new DirectChannel(),
new ConsumerProperties());
ConsumerProperties consumerProperties = new ConsumerProperties();
consumerProperties.setMaxAttempts(1); // to force error infrastructure creation
Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo", "fooGroup", new DirectChannel(), consumerProperties);
DirectFieldAccessor consumerBindingAccessor = new DirectFieldAccessor(consumerBinding);
Object messageProducer = consumerBindingAccessor.getPropertyValue("lifecycle");
Mockito.verify((Lifecycle) messageProducer).start();
Mockito.verify((InitializingBean) messageProducer).afterPropertiesSet();
Mockito.verify((MessageProducer) messageProducer).setOutputChannel(Mockito.any(MessageChannel.class));
Mockito.verifyNoMoreInteractions(messageProducer);
ErrorInfrastructure errorInfra = binder.errorInfrastructure;
SubscribableChannel errorChannel = errorInfra.getErrorChannel();
MessageProducer messageProducer = (MessageProducer) consumerBindingAccessor.getPropertyValue("lifecycle");
assertTrue(((Lifecycle)messageProducer).isRunning());
assertNotNull(messageProducer.getOutputChannel());
SubscribableChannel errorChannel = (SubscribableChannel) consumerBindingAccessor.getPropertyValue("lifecycle.errorChannel");
assertThat(errorChannel).isNotNull();
Set<MessageHandler> handlers = TestUtils.getPropertyValue(errorChannel, "dispatcher.handlers", Set.class);
assertThat(handlers.size()).isEqualTo(2);
@@ -89,46 +92,37 @@ public class AbstractMessageChannelBinderTests {
assertThat(context.containsBean("foo.fooGroup.errors.handler")).isFalse();
assertThat(context.containsBean("foo.fooGroup.errors.bridge")).isFalse();
Mockito.verify((Lifecycle) messageProducer).stop();
Mockito.verify((DisposableBean) messageProducer).destroy();
Mockito.verifyNoMoreInteractions(messageProducer);
assertFalse(((Lifecycle) messageProducer).isRunning());
ProducerProperties producerProps = new ProducerProperties();
producerProps.setErrorChannelEnabled(true);
Binding<MessageChannel> producerBinding = binder.bindProducer("bar", new DirectChannel(), producerProps);
DirectFieldAccessor producerBindingAccessor = new DirectFieldAccessor(producerBinding);
Object messageHandler = producerBindingAccessor.getPropertyValue("lifecycle");
Mockito.verify((Lifecycle) messageHandler).start();
Mockito.verify((InitializingBean) messageHandler).afterPropertiesSet();
Mockito.verifyNoMoreInteractions(messageHandler);
assertThat(context.containsBean("bar.errors")).isTrue();
assertThat(context.containsBean("bar.errors.bridge")).isTrue();
producerBinding.unbind();
Mockito.verify((Lifecycle) messageHandler).stop();
Mockito.verify((DisposableBean) messageHandler).destroy();
Mockito.verifyNoMoreInteractions(messageHandler);
assertThat(context.containsBean("bar.errors")).isFalse();
assertThat(context.containsBean("bar.errors.bridge")).isFalse();
}
@Test
public void testEndpointBinderHasRecoverer() throws Exception {
StubMessageChannelBinder binder = new StubMessageChannelBinder(true);
GenericApplicationContext context = new GenericApplicationContext();
context.refresh();
context.getBeanFactory().registerSingleton("errorChannel", new PublishSubscribeChannel());
binder.setApplicationContext(context);
ConfigurableApplicationContext context =
new SpringApplicationBuilder(SpringIntegrationBinderConfiguration.getCompleteConfiguration()).web(WebApplicationType.NONE).run();
Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo", "fooGroup", new DirectChannel(),
new ConsumerProperties());
ErrorInfrastructure errorInfra = binder.errorInfrastructure;
SubscribableChannel errorChannel = errorInfra.getErrorChannel();
AbstractMessageChannelBinder<ConsumerProperties, ProducerProperties, ProvisioningProvider<ConsumerProperties, ProducerProperties>> binder =
context.getBean(AbstractMessageChannelBinder.class);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo", "fooGroup", new DirectChannel(), new ConsumerProperties());
DirectFieldAccessor consumerBindingAccessor = new DirectFieldAccessor(consumerBinding);
SubscribableChannel errorChannel = (SubscribableChannel) consumerBindingAccessor.getPropertyValue("lifecycle.errorChannel");
assertThat(errorChannel).isNull();
errorChannel = (SubscribableChannel) consumerBindingAccessor.getPropertyValue("lifecycle.recoveryCallback.channel");
assertThat(errorChannel).isNotNull();
Set<MessageHandler> handlers = TestUtils.getPropertyValue(errorChannel, "dispatcher.handlers", Set.class);
assertThat(handlers.size()).isEqualTo(2);
Iterator<MessageHandler> iterator = handlers.iterator();
assertThat(iterator.next()).isInstanceOf(BridgeHandler.class);
assertThat(iterator.next()).isNotInstanceOf(LastSubscriberMessageHandler.class);
assertThat(iterator.next()).isInstanceOf(LastSubscriberMessageHandler.class);
assertThat(context.containsBean("foo.fooGroup.errors")).isTrue();
assertThat(context.containsBean("foo.fooGroup.errors.recoverer")).isTrue();
assertThat(context.containsBean("foo.fooGroup.errors.handler")).isTrue();
@@ -139,111 +133,4 @@ public class AbstractMessageChannelBinderTests {
assertThat(context.containsBean("foo.fooGroup.errors.handler")).isFalse();
assertThat(context.containsBean("foo.fooGroup.errors.bridge")).isFalse();
}
private static class StubMessageChannelBinder extends
AbstractMessageChannelBinder<ConsumerProperties, ProducerProperties,
ProvisioningProvider<ConsumerProperties, ProducerProperties>> {
private final boolean hasRecoverer;
private ErrorInfrastructure errorInfrastructure;
StubMessageChannelBinder() {
this(false);
}
StubMessageChannelBinder(boolean hasRecoverer) {
super(null, Mockito.mock(ProvisioningProvider.class));
mockProvisioner();
this.hasRecoverer = hasRecoverer;
}
private void mockProvisioner() {
willAnswer(new Answer<SimpleConsumerDestination>() {
@Override
public SimpleConsumerDestination answer(final InvocationOnMock invocation) throws Throwable {
return new SimpleConsumerDestination(invocation.getArgument(0));
}
}).given(this.provisioningProvider).provisionConsumerDestination(anyString(), anyString(),
any(ConsumerProperties.class));
willAnswer(new Answer<SimpleProducerDestination>() {
@Override
public SimpleProducerDestination answer(final InvocationOnMock invocation) throws Throwable {
return new SimpleProducerDestination(invocation.getArgument(0));
}
}).given(this.provisioningProvider).provisionProducerDestination(anyString(),
any(ProducerProperties.class));
}
@Override
protected MessageHandler createProducerMessageHandler(ProducerDestination destination,
ProducerProperties producerProperties, MessageChannel errorChannel) throws Exception {
MessageHandler mock = Mockito.mock(MessageHandler.class, Mockito.withSettings()
.extraInterfaces(Lifecycle.class, InitializingBean.class, DisposableBean.class));
return mock;
}
@Override
protected MessageProducer createConsumerEndpoint(ConsumerDestination destination, String group,
ConsumerProperties properties) throws Exception {
this.errorInfrastructure = registerErrorInfrastructure(destination, group, properties);
MessageProducer adapter = Mockito.mock(MessageProducer.class,
Mockito.withSettings().extraInterfaces(Lifecycle.class, InitializingBean.class,
DisposableBean.class));
return adapter;
}
@Override
protected MessageHandler getErrorMessageHandler(ConsumerDestination destination, String group,
ConsumerProperties consumerProperties) {
if (this.hasRecoverer) {
return mock(MessageHandler.class);
}
else {
return null;
}
}
}
private static class SimpleConsumerDestination implements ConsumerDestination {
private final String name;
SimpleConsumerDestination(String name) {
this.name = name;
}
@Override
public String getName() {
return this.name;
}
}
private static class SimpleProducerDestination implements ProducerDestination {
private final String name;
SimpleProducerDestination(String name) {
this.name = name;
}
@Override
public String getName() {
return this.name;
}
@Override
public String getNameForPartition(int partition) {
return getName() + partition;
}
}
}

6
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesOnlyTest.java
View File

@@ -24,11 +24,10 @@ 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.integration.SpringIntegrationBinderConfiguration;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
import org.springframework.context.annotation.Import;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
@@ -36,7 +35,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
* @author Ilayaperumal Gopinathan
*/
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(classes = SourceBindingWithGlobalPropertiesOnlyTest.TestSource.class, properties = {
@SpringBootTest(classes = {SpringIntegrationBinderConfiguration.class, SourceBindingWithGlobalPropertiesOnlyTest.TestSource.class}, properties = {
"spring.cloud.stream.default.contentType=application/json",
"spring.cloud.stream.default.producer.partitionKeyExpression=key" })
public class SourceBindingWithGlobalPropertiesOnlyTest {
@@ -54,7 +53,6 @@ public class SourceBindingWithGlobalPropertiesOnlyTest {
@EnableBinding(Source.class)
@EnableAutoConfiguration
@Import(MockBinderRegistryConfiguration.class)
public static class TestSource {
}

1
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationChannelBinder.java
View File

@@ -127,6 +127,7 @@ class SpringIntegrationChannelBinder extends AbstractMessageChannelBinder<Consum
ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(destination, groupName, properties);
if (properties.getMaxAttempts() > 1) {
adapter.setRetryTemplate(buildRetryTemplate(properties));
adapter.setRecoveryCallback(errorInfrastructure.getRecoverer());
}
else {
adapter.setErrorMessageStrategy(errorMessageStrategy);