From d992e0b8526f54f2caf0234cd12a4d87d182bd36 Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Wed, 7 Oct 2015 12:17:01 -0400 Subject: [PATCH] XD-3597 Separate input, output and context start 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 --- .../binding/ChannelBindingLifecycle.java | 128 ------------------ .../ContextStartAfterRefreshListener.java | 47 +++++++ .../stream/binding/InputBindingLifecycle.java | 115 ++++++++++++++++ .../binding/OutputBindingLifecycle.java | 117 ++++++++++++++++ .../ChannelBindingServiceConfiguration.java | 20 ++- .../binder/InputOutputBindingOrderTest.java | 119 ++++++++++++++++ 6 files changed, 415 insertions(+), 131 deletions(-) delete mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingLifecycle.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ContextStartAfterRefreshListener.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/InputBindingLifecycle.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/OutputBindingLifecycle.java create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/InputOutputBindingOrderTest.java diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingLifecycle.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingLifecycle.java deleted file mode 100644 index b11abd58d..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingLifecycle.java +++ /dev/null @@ -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 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 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; - } -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ContextStartAfterRefreshListener.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ContextStartAfterRefreshListener.java new file mode 100644 index 000000000..6f2311cbd --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ContextStartAfterRefreshListener.java @@ -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, 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(); + } + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/InputBindingLifecycle.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/InputBindingLifecycle.java new file mode 100644 index 000000000..f2eeaacf7 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/InputBindingLifecycle.java @@ -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 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 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; + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/OutputBindingLifecycle.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/OutputBindingLifecycle.java new file mode 100644 index 000000000..a8c878dfc --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/OutputBindingLifecycle.java @@ -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 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 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; + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java index 08b200722..f9a19a609 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java @@ -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 diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/InputOutputBindingOrderTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/InputOutputBindingOrderTest.java new file mode 100644 index 000000000..3de6254dc --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/InputOutputBindingOrderTest.java @@ -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.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.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; + } + } +}