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

XD-3597 Separate input, output and context start

Browse Source 
JIRA: https://jira.spring.io/browse/XD-3597

Separates the functionality currently provided by the ChannelBindingLifecycle,
binding inputs and outputs separately at the latest, and earliest phases,
respectively - allowing for SmartLifecycle beans that subscribe to inputs at start
to subscribe before inputs start producing data.

The current functionality of automatically starting the context
on the ChannelBindingLifecycle auto-startup is deferred to a refresh listener.

Removes the call to application.stop() within the ChannelBindingLifecycle, which
is redundant (the ChannelBindingLifecycle would be stopped when the context itself
is stopped anyway).

Adding a test for binding lifecycle

Polishing
This commit is contained in:
Marius Bogoevici
2015-10-07 12:17:01 -04:00
committed by Gary Russell
parent 6c8e6313a4
commit d992e0b852
6 changed files with 415 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

128
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingLifecycle.java
View File

@@ -1,128 +0,0 @@
/*
* Copyright 2015 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 java.util.concurrent.atomic.AtomicBoolean;
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/output channels in accordance to the lifecycle of the host context.
*
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
public class ChannelBindingLifecycle implements SmartLifecycle, ApplicationContextAware {
private volatile boolean running = false;
private ConfigurableApplicationContext applicationContext;
private final AtomicBoolean active = new AtomicBoolean(false);
@Override
public void setApplicationContext(ApplicationContext applicationContext)
throws BeansException {
this.applicationContext = (ConfigurableApplicationContext) applicationContext;
}
@Override
public void start() {
if (!running) {
if (!this.active.get()) {
if (this.active.compareAndSet(false, true)) {
// retrieve the ChannelBindingService lazily, avoiding early initialization
try {
ChannelBindingService channelBindingService = this.applicationContext.getBean(ChannelBindingService.class);
Map<String, Bindable> bindables = this.applicationContext.getBeansOfType(Bindable.class);
for (Bindable bindable : bindables.values()) {
bindable.bindOutputs(channelBindingService);
}
for (Bindable bindable : bindables.values()) {
bindable.bindInputs(channelBindingService);
}
}
catch (BeansException e) {
throw new IllegalStateException("Cannot perform binding, no proper implementation found", e);
}
this.running = true;
this.applicationContext.start();
this.active.set(false);
}
}
}
}
@Override
public void stop() {
if (running) {
if (!this.active.get()) {
if (this.active.compareAndSet(false, true)) {
try {
// retrieve the ChannelBindingService lazily, avoiding early initialization
ChannelBindingService channelBindingService = this.applicationContext.getBean(ChannelBindingService.class);
Map<String, Bindable> bindables = this.applicationContext.getBeansOfType(Bindable.class);
for (Bindable bindable : bindables.values()) {
bindable.unbindInputs(channelBindingService);
}
for (Bindable bindable : bindables.values()) {
bindable.unbindOutputs(channelBindingService);
}
}
catch (BeansException e) {
throw new IllegalStateException("Cannot perform binding, no proper implementation found", e);
}
this.applicationContext.stop();
this.active.set(false);
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();
}
}
/**
* Return the lowest value to start this bean before any message producing lifecycle
* beans.
*/
@Override
public int getPhase() {
return Integer.MIN_VALUE;
}
}

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

@@ -0,0 +1,47 @@
/*
* Copyright 2015 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 org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationListener;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.event.ContextRefreshedEvent;
/**
* Automatically starts the context after a refresh.
*
* @author Marius Bogoevici
*/
public class ContextStartAfterRefreshListener implements ApplicationListener<ContextRefreshedEvent>, ApplicationContextAware {
private ApplicationContext applicationContext;
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
}
@Override
public void onApplicationEvent(ContextRefreshedEvent event) {
ConfigurableApplicationContext source = (ConfigurableApplicationContext) event.getSource();
if (source == this.applicationContext && !source.isRunning()) {
source.start();
}
}
}

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

@@ -0,0 +1,115 @@
/*
* Copyright 2015 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 java.util.concurrent.atomic.AtomicBoolean;
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 channels in accordance to the lifecycle
* of the host context.
*
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
public class InputBindingLifecycle implements SmartLifecycle, ApplicationContextAware {
private volatile boolean running = false;
private ConfigurableApplicationContext applicationContext;
@Override
public void setApplicationContext(ApplicationContext applicationContext)
throws BeansException {
this.applicationContext = (ConfigurableApplicationContext) applicationContext;
}
@Override
public void start() {
if (!running) {
// retrieve the ChannelBindingService lazily, avoiding early initialization
try {
ChannelBindingService channelBindingService = this.applicationContext
.getBean(ChannelBindingService.class);
Map<String, Bindable> bindables = this.applicationContext
.getBeansOfType(Bindable.class);
for (Bindable bindable : bindables.values()) {
bindable.bindInputs(channelBindingService);
}
}
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 ChannelBindingService lazily, avoiding early
// initialization
ChannelBindingService channelBindingService = this.applicationContext
.getBean(ChannelBindingService.class);
Map<String, Bindable> bindables = this.applicationContext
.getBeansOfType(Bindable.class);
for (Bindable bindable : bindables.values()) {
bindable.unbindInputs(channelBindingService);
}
}
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();
}
}
/**
* Return a high value so that this bean is started after receiving Lifecycle beans are started. Beans that need to start after bindings will set a higher phase value.
*/
@Override
public int getPhase() {
return Integer.MAX_VALUE - 1000;
}
}

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

@@ -0,0 +1,117 @@
/*
* Copyright 2015 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 java.util.concurrent.atomic.AtomicBoolean;
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 channels in accordance to the lifecycle
* of the host context.
*
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
public class OutputBindingLifecycle implements SmartLifecycle, ApplicationContextAware {
private volatile boolean running = false;
private ConfigurableApplicationContext applicationContext;
@Override
public void setApplicationContext(ApplicationContext applicationContext)
throws BeansException {
this.applicationContext = (ConfigurableApplicationContext) applicationContext;
}
@Override
public void start() {
if (!running) {
// retrieve the ChannelBindingService lazily, avoiding early initialization
try {
ChannelBindingService channelBindingService = this.applicationContext
.getBean(ChannelBindingService.class);
Map<String, Bindable> bindables = this.applicationContext
.getBeansOfType(Bindable.class);
for (Bindable bindable : bindables.values()) {
bindable.bindOutputs(channelBindingService);
}
}
catch (BeansException e) {
throw new IllegalStateException(
"Cannot perform binding, no proper implementation found", e);
}
this.running = true;
this.applicationContext.start();
}
}
@Override
public void stop() {
if (running) {
try {
// retrieve the ChannelBindingService lazily, avoiding early
// initialization
ChannelBindingService channelBindingService = this.applicationContext
.getBean(ChannelBindingService.class);
Map<String, Bindable> bindables = this.applicationContext
.getBeansOfType(Bindable.class);
for (Bindable bindable : bindables.values()) {
bindable.unbindOutputs(channelBindingService);
}
}
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();
}
}
/**
* Return a low value so that this bean is started after receiving Lifecycle beans are started. Beans that need to start before bindings will set a lower phase value.
*/
@Override
public int getPhase() {
return Integer.MIN_VALUE + 1000;
}
}

20
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java
View File

@@ -26,7 +26,9 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
import org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor;
import org.springframework.cloud.stream.binding.ChannelBindingLifecycle;
import org.springframework.cloud.stream.binding.ContextStartAfterRefreshListener;
import org.springframework.cloud.stream.binding.InputBindingLifecycle;
import org.springframework.cloud.stream.binding.OutputBindingLifecycle;
import org.springframework.cloud.stream.binding.ChannelBindingService;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -61,8 +63,20 @@ public class ChannelBindingServiceConfiguration {
@Bean
@DependsOn("bindingService")
public ChannelBindingLifecycle channelBindingLifecycle() {
return new ChannelBindingLifecycle();
public OutputBindingLifecycle outputBindingLifecycle() {
return new OutputBindingLifecycle();
}
@Bean
@DependsOn("bindingService")
public InputBindingLifecycle inputBindingLifecycle() {
return new InputBindingLifecycle();
}
@Bean
@DependsOn("bindingService")
public ContextStartAfterRefreshListener contextStartAfterRefreshListener() {
return new ContextStartAfterRefreshListener();
}
@Bean

119
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/InputOutputBindingOrderTest.java Normal file
View File

@@ -0,0 +1,119 @@
/*
* Copyright 2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.verifyZeroInteractions;
import java.util.Properties;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.utils.MockBinderConfiguration;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.SmartLifecycle;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
/**
* @author Marius Bogoevici
*/
public class InputOutputBindingOrderTest {
@SuppressWarnings("unchecked")
@Test
public void testInputOutputBindingOrder() {
ConfigurableApplicationContext applicationContext = SpringApplication.run(TestSource.class);
Binder binder = applicationContext.getBean(Binder.class);
Processor processor = applicationContext.getBean(Processor.class);
// input is bound after the context has been started
verify(binder).bindConsumer(eq("input"), eq(processor.input()), Mockito.<Properties>any());
SomeLifecycle someLifecycle = applicationContext.getBean(SomeLifecycle.class);
assertTrue(someLifecycle.isRunning());
applicationContext.close();
assertFalse(someLifecycle.isRunning());
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
@Import(MockBinderConfiguration.class)
public static class TestSource {
@Bean
public SomeLifecycle someLifecycle() {
return new SomeLifecycle();
}
}
public static class SomeLifecycle implements SmartLifecycle {
private boolean running = false;
@SuppressWarnings("rawtypes")
@Autowired
private Binder binder;
@Autowired
private Processor processor;
@Override
public synchronized void start() {
verify(binder).bindProducer(eq("output"), eq(processor.output()), Mockito.<Properties>any());
// input was not bound yet
verifyNoMoreInteractions(binder);
this.running = true;
}
@Override
public synchronized void stop() {
running = false;
}
@Override
public synchronized boolean isRunning() {
return running;
}
@Override
public boolean isAutoStartup() {
return true;
}
@Override
public void stop(Runnable callback) {
stop();
if (callback != null) {
callback.run();
}
}
@Override
public int getPhase() {
return 0;
}
}
}