diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java index deeb22578..79b061684 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-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. @@ -16,6 +16,8 @@ package org.springframework.cloud.stream.binder; +import org.springframework.context.Lifecycle; + /** * Represents a binding between an input or output and an adapter endpoint that connects * via a Binder. The binding could be for a consumer or a producer. A consumer binding @@ -26,9 +28,46 @@ package org.springframework.cloud.stream.binder; * @author Mark Fisher * @author Gary Russell * @author Marius Bogoevici + * @author Oleg Zhurakousky + * * @see org.springframework.cloud.stream.annotation.EnableBinding */ -public interface Binding { +public interface Binding extends Lifecycle { + + /** + * Stops the target component represented by this instance. + * NOTE: At the time the instance is created the component is already started. + * This operation is typically used by actuator to re-bind/re-start. + * + * @see BindingsEndpoint + */ + default void start() {} + + /** + * Starts the target component represented by this instance. + * NOTE: At the time the instance is created the component is already started. + * This operation is typically used by actuator to re-bind/re-start. + * + * @see BindingsEndpoint + */ + default void stop() {} + + /** + * Returns 'true' if the target component represented by this instance is running. + */ + default boolean isRunning() { + return false; + } + + /** + * Returns the name of this binding (i.e., channel name) + * + * @return binding name + */ + default String getName() { + return null; + } + /** * Unbinds the target component represented by this instance and stops any active diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java index a756fda91..cd2acb49f 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2016 the original author or authors. + * Copyright 2013-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. @@ -16,10 +16,14 @@ package org.springframework.cloud.stream.binder; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + import org.springframework.context.Lifecycle; import org.springframework.integration.support.context.NamedComponent; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; +import org.springframework.util.StringUtils; /** * Default implementation for a {@link Binding}. @@ -27,9 +31,13 @@ import org.springframework.util.ObjectUtils; * @author Mark Fisher * @author Gary Russell * @author Marius Bogoevici + * @author Oleg Zhurakousky + * * @see org.springframework.cloud.stream.annotation.EnableBinding */ public class DefaultBinding implements Binding { + + private final Log logger = LogFactory.getLog(this.getClass().getName()); protected final String name; @@ -64,12 +72,33 @@ public class DefaultBinding implements Binding { public String getGroup() { return this.group; } + + public boolean isRunning() { + return this.lifecycle != null && this.lifecycle.isRunning(); + } + + @Override + public final synchronized void start() { + if (!this.isRunning()) { + if (this.lifecycle != null && StringUtils.hasText(this.group)) { + this.lifecycle.start(); + } + else { + logger.warn("Can not re-bind an anonymous binding"); + } + } + } + + @Override + public final synchronized void stop() { + if (this.isRunning()) { + this.lifecycle.stop(); + } + } @Override public final void unbind() { - if (this.lifecycle != null) { - this.lifecycle.stop(); - } + this.stop(); afterUnbind(); } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/Bindable.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/Bindable.java index d3ad5c7c6..c628c1632 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/Bindable.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/Bindable.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2017 the original author or authors. + * 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. @@ -16,9 +16,12 @@ package org.springframework.cloud.stream.binding; +import java.util.Collection; import java.util.Collections; import java.util.Set; +import org.springframework.cloud.stream.binder.Binding; + /** * Marker interface for instances that can bind/unbind groups of inputs and outputs. * @@ -31,8 +34,23 @@ public interface Bindable { /** * Binds all the inputs associated with this instance. + * @deprecated as of 2.0 in favor of {@link #createAndBindInputs(BindingService)} */ - default void bindInputs(BindingService adapter) {} + @Deprecated + default void bindInputs(BindingService adapter) { + this.createAndBindInputs(adapter); + } + + /** + * Binds all the inputs associated with this instance. + * @param adapter instance of {@link BindingService} + * @return collection of {@link Binding}s + * + * @since 2.0 + */ + default Collection> createAndBindInputs(BindingService adapter) { + return Collections.>emptyList(); + } /** * Binds all the outputs associated with this instance. diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java index 6e58cdf89..ebe078b1b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2017 the original author or authors. + * 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. @@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binding; import java.lang.reflect.Method; import java.util.ArrayList; +import java.util.Collection; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -37,6 +38,7 @@ import org.springframework.cloud.stream.aggregate.SharedBindingTargetRegistry; 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.Binding; import org.springframework.cloud.stream.internal.InternalPropertyNames; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.util.Assert; @@ -50,6 +52,7 @@ import org.springframework.util.StringUtils; * @author Marius Bogoevici * @author David Syer * @author Ilayaperumal Gopinathan + * @author Oleg Zhurakousky * * @see EnableBinding */ @@ -206,8 +209,18 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean> createAndBindInputs(BindingService bindingService) { + List> bindings = new ArrayList<>(); if (log.isDebugEnabled()) { log.debug(String.format("Binding inputs for %s:%s", this.namespace, this.type)); } @@ -218,9 +231,10 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean> inputBindings; public InputBindingLifecycle(BindingService bindingService, Map bindables) { super(bindingService, bindables); @@ -42,7 +49,7 @@ public class InputBindingLifecycle extends AbstractBindingLifecycle { @Override void doStartWithBindable(Bindable bindable) { - bindable.bindInputs(bindingService); + this.inputBindings = bindable.createAndBindInputs(bindingService); } @Override diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingsEndpointAutoConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingsEndpointAutoConfiguration.java new file mode 100644 index 000000000..57fabf691 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingsEndpointAutoConfiguration.java @@ -0,0 +1,57 @@ +/* + * Copyright 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 + * + * 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.config; + +import java.util.List; + +import org.springframework.boot.actuate.autoconfigure.endpoint.EndpointAutoConfiguration; +import org.springframework.boot.autoconfigure.AutoConfigureAfter; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.cloud.stream.binding.BindingService; +import org.springframework.cloud.stream.binding.InputBindingLifecycle; +import org.springframework.cloud.stream.endpoint.BindingsEndpoint; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * @author Oleg Zhurakousky + * + * @since 2.0 + */ +@Configuration +@ConditionalOnClass(name = "org.springframework.boot.actuate.endpoint.annotation.Endpoint") +@ConditionalOnBean(BindingService.class) +@AutoConfigureAfter(EndpointAutoConfiguration.class) +public class BindingsEndpointAutoConfiguration { + + @Bean + public BindingsEndpoint bindingsEndpoint(List inputBindings) { + return new BindingsEndpoint(inputBindings); + } + + @Bean + public BindingsEndpoint.StopEndpoint stopEndpoint(BindingsEndpoint bindingEndpoint) { + return bindingEndpoint.getStopEndpoint(); + } + + @Bean + public BindingsEndpoint.StartEndpoint startEndpoint(BindingsEndpoint bindingEndpoint) { + return bindingEndpoint.getStartEndpoint(); + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/endpoint/BindingsEndpoint.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/endpoint/BindingsEndpoint.java new file mode 100644 index 000000000..2433610f5 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/endpoint/BindingsEndpoint.java @@ -0,0 +1,111 @@ +/* + * Copyright 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 + * + * 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.endpoint; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.boot.actuate.endpoint.annotation.Endpoint; +import org.springframework.boot.actuate.endpoint.annotation.ReadOperation; +import org.springframework.boot.actuate.endpoint.annotation.WriteOperation; +import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.binding.InputBindingLifecycle; +import org.springframework.util.Assert; + +/** + * + * @author Oleg Zhurakousky + * + * @since 2.0 + * + */ +@Endpoint(id = BindingsEndpoint.BASE_ENPOINT_ID) +public class BindingsEndpoint { + + static final String BASE_ENPOINT_ID = "bindings"; + + static final String START_ENPOINT_ID = BASE_ENPOINT_ID + "/" + "start/{name}"; + + static final String STOP_ENPOINT_ID = BASE_ENPOINT_ID + "/" + "stop/{name}"; + + private final List inputBindingLifecycles; + + public BindingsEndpoint(List inputBindingLifecycles) { + Assert.notEmpty(inputBindingLifecycles, "'inputBindingLifecycles' must not be null or empty"); + this.inputBindingLifecycles = inputBindingLifecycles; + } + + @ReadOperation + public List bindings() { + return new ObjectMapper().convertValue(this.gatherInputBindings(), new TypeReference>() {}); + } + + public StopEndpoint getStopEndpoint() { + return new StopEndpoint(); + } + + public StartEndpoint getStartEndpoint() { + return new StartEndpoint(); + } + + @SuppressWarnings("unchecked") + private List> gatherInputBindings() { + List> inputBindings = new ArrayList<>(); + for (InputBindingLifecycle inputBindingLifecycle : inputBindingLifecycles) { + Collection> lifecycleInputBindings = + (Collection>) new DirectFieldAccessor(inputBindingLifecycle).getPropertyValue("inputBindings"); + inputBindings.addAll(lifecycleInputBindings); + } + return inputBindings; + } + + @Endpoint(id = BindingsEndpoint.STOP_ENPOINT_ID) + public class StopEndpoint { + @WriteOperation + public boolean stop(String name) { + List> inputBindings = BindingsEndpoint.this.gatherInputBindings(); + for (Binding binding : inputBindings) { + if (name.equals(binding.getName())) { + binding.stop(); + return true; + } + } + return false; + } + } + + @Endpoint(id = BindingsEndpoint.START_ENPOINT_ID) + public class StartEndpoint { + @WriteOperation + public boolean start(String name) { + List> inputBindings = BindingsEndpoint.this.gatherInputBindings(); + for (Binding binding : inputBindings) { + if (name.equals(binding.getName())) { + binding.start(); + return true; + } + } + return false; + } + } + +} diff --git a/spring-cloud-stream/src/main/resources/META-INF/spring.factories b/spring-cloud-stream/src/main/resources/META-INF/spring.factories index 8a18c68f6..aaedc1cda 100644 --- a/spring-cloud-stream/src/main/resources/META-INF/spring.factories +++ b/spring-cloud-stream/src/main/resources/META-INF/spring.factories @@ -1,5 +1,6 @@ org.springframework.boot.autoconfigure.EnableAutoConfiguration:\ org.springframework.cloud.stream.config.ChannelBindingAutoConfiguration,\ org.springframework.cloud.stream.config.BindersHealthIndicatorAutoConfiguration,\ -org.springframework.cloud.stream.config.ChannelsEndpointAutoConfiguration +org.springframework.cloud.stream.config.ChannelsEndpointAutoConfiguration,\ +org.springframework.cloud.stream.config.BindingsEndpointAutoConfiguration