diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 6f65f0031..db152117a 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -1707,9 +1707,27 @@ Note that, when there are more than one instance of the same type of the binder, [[binding_visualization_control]] === Binding visualization and control -Since version 2.0, Spring Cloud Stream supports visualization and control of the Bindings through Actuator endpoints. +Spring Cloud Stream supports visualization and control of the Bindings through Actuator endpoints as well as programmatic way. -Starting with version 2.0 actuator and web are optional, you must first add one of the web dependencies as well as add the actuator dependency manually. +==== Programmatic way + +Since version 3.1 we expose `org.springframework.cloud.stream.binding.BindingsLifecycleController` which is registered as bean and once +injected could be used to control the lifecycle of individual bindings + +For example, looks at the fragment from one of the test cases. As you can see we retrieve `BindingsLifecycleController` +from spring application context and execute individual methods to control the lifecycle of `echo-in-0` binding.. + +[source,java] +---- +BindingsLifecycleController bindingsController = context.getBean(BindingsLifecycleController.class); +Binding binding = bindingsController.queryState("echo-in-0"); +assertThat(binding.isRunning()).isTrue(); +bindingsController.changeState("echo-in-0", State.STOPPED); +//bindingsController.stop("echo-in-0") alternative way of changing state. For convenience we expose start/stop and pause/resume operations. +assertThat(binding.isRunning()).isFalse(); + +==== Actuator +Since actuator and web are optional, you must first add one of the web dependencies as well as add the actuator dependency manually. The following example shows how to add the dependency for the Web framework: [source,xml] diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingsLifecycleController.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingsLifecycleController.java new file mode 100644 index 000000000..3d0a8abca --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingsLifecycleController.java @@ -0,0 +1,202 @@ +/* + * Copyright 2021-2021 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.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.stream.Stream; + +import com.fasterxml.jackson.databind.ObjectMapper; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.cloud.stream.binder.Binding; +import org.springframework.util.Assert; + +/** + * + * Lifecycle controller for the bindings. + * It is registered as a bean and once injected could be used to control the lifecycle f the bindings. + * + * @author Oleg Zhurakousky + * @since 3.x + */ +public class BindingsLifecycleController { + + private final List inputBindingLifecycles; + + private final List outputBindingsLifecycles; + + private final ObjectMapper objectMapper; + + public BindingsLifecycleController(List inputBindingLifecycles, + List outputBindingsLifecycles) { + Assert.notEmpty(inputBindingLifecycles, + "'inputBindingLifecycles' must not be null or empty"); + this.inputBindingLifecycles = inputBindingLifecycles; + this.outputBindingsLifecycles = outputBindingsLifecycles; + this.objectMapper = new ObjectMapper(); + } + + /** + * Convenience method to stop the binding with provided `bindingName`. + * @param bindingName the name of the binding. + */ + public void stop(String bindingName) { + this.changeState(bindingName, State.STOPPED); + } + + /** + * Convenience method to start the binding with provided `bindingName`. + * @param bindingName the name of the binding. + */ + public void start(String bindingName) { + this.changeState(bindingName, State.STARTED); + } + + /** + * Convenience method to pause the binding with provided `bindingName`. + * @param bindingName the name of the binding. + */ + public void pause(String bindingName) { + this.changeState(bindingName, State.PAUSED); + } + + /** + * Convenience method to resume the binding with provided `bindingName`. + * @param bindingName the name of the binding. + */ + public void resume(String bindingName) { + this.changeState(bindingName, State.RESUMED); + } + + /** + * General purpose method to change the state of the provided binding. + * @param bindingName the name of the binding. + * @param state the {@link State} you wish to set this binding to + */ + public void changeState(String bindingName, State state) { + Binding binding = BindingsLifecycleController.this.locateBinding(bindingName); + if (binding != null) { + switch (state) { + case STARTED: + binding.start(); + break; + case STOPPED: + binding.stop(); + break; + case PAUSED: + binding.pause(); + break; + case RESUMED: + binding.resume(); + break; + default: + break; + } + } + } + + /** + * Queries the {@link List} of states for all available bindings. The returned list + * consists of {@link Binding} objects which could be further interrogated + * using {@link Binding#isPaused()} and {@link Binding#isRunning()}. + * @return the list of {@link Binding}s + */ + @SuppressWarnings("unchecked") + public List> queryStates() { + List> bindings = new ArrayList<>(gatherInputBindings()); + bindings.addAll(gatherOutputBindings()); + return this.objectMapper.convertValue(bindings, List.class); + } + + /** + * Queries the individual state of a binding. The returned list + * {@link Binding} object could be further interrogated + * using {@link Binding#isPaused()} and {@link Binding#isRunning()}. + * @return instance of {@link Binding} object. + */ + public Binding queryState(String name) { + Assert.notNull(name, "'name' must not be null"); + return this.locateBinding(name); + } + + /** + * Queries for all input {@link Binding}s. + * @return the list of input {@link Binding}s + */ + @SuppressWarnings("unchecked") + private List> gatherInputBindings() { + List> inputBindings = new ArrayList<>(); + for (InputBindingLifecycle inputBindingLifecycle : this.inputBindingLifecycles) { + Collection> lifecycleInputBindings = (Collection>) new DirectFieldAccessor( + inputBindingLifecycle).getPropertyValue("inputBindings"); + inputBindings.addAll(lifecycleInputBindings); + } + return inputBindings; + } + + /** + * Queries for all output {@link Binding}s. + * @return the list of output {@link Binding}s + */ + @SuppressWarnings("unchecked") + private List> gatherOutputBindings() { + List> outputBindings = new ArrayList<>(); + for (OutputBindingLifecycle inputBindingLifecycle : this.outputBindingsLifecycles) { + Collection> lifecycleInputBindings = (Collection>) new DirectFieldAccessor( + inputBindingLifecycle).getPropertyValue("outputBindings"); + outputBindings.addAll(lifecycleInputBindings); + } + return outputBindings; + } + + private Binding locateBinding(String name) { + Stream> bindings = Stream.concat(this.gatherInputBindings().stream(), + this.gatherOutputBindings().stream()); + return bindings.filter(binding -> name.equals(binding.getBindingName())).findFirst() + .orElse(null); + } + + /** + * Binding states. + */ + public enum State { + + /** + * Started state of a binding. + */ + STARTED, + + /** + * Stopped state of a binding. + */ + STOPPED, + + /** + * Paused state of a binding. + */ + PAUSED, + + /** + * Resumed state of a binding. + */ + RESUMED; + + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java index 344d99d22..00f499b97 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java @@ -45,6 +45,7 @@ import org.springframework.cloud.stream.binding.Bindable; import org.springframework.cloud.stream.binding.BinderAwareChannelResolver; import org.springframework.cloud.stream.binding.BinderAwareRouter; import org.springframework.cloud.stream.binding.BindingService; +import org.springframework.cloud.stream.binding.BindingsLifecycleController; import org.springframework.cloud.stream.binding.ContextStartAfterRefreshListener; import org.springframework.cloud.stream.binding.DynamicDestinationsBindable; import org.springframework.cloud.stream.binding.InputBindingLifecycle; @@ -238,6 +239,12 @@ public class BindingServiceConfiguration { return new InputBindingLifecycle(bindingService, bindables); } + @Bean + public BindingsLifecycleController bindingsLifecycleController(List inputBindingLifecycles, + List outputBindingsLifecycles) { + return new BindingsLifecycleController(inputBindingLifecycles, outputBindingsLifecycles); + } + @Bean @DependsOn("bindingService") public ContextStartAfterRefreshListener contextStartAfterRefreshListener() { 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 index 20a4df31c..a951ae82a 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2021 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,16 +16,13 @@ package org.springframework.cloud.stream.config; -import java.util.List; - import org.springframework.boot.actuate.autoconfigure.endpoint.EndpointAutoConfiguration; import org.springframework.boot.actuate.autoconfigure.endpoint.condition.ConditionalOnAvailableEndpoint; 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.binding.OutputBindingLifecycle; +import org.springframework.cloud.stream.binding.BindingsLifecycleController; import org.springframework.cloud.stream.endpoint.BindingsEndpoint; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -43,9 +40,8 @@ public class BindingsEndpointAutoConfiguration { @Bean @ConditionalOnAvailableEndpoint - public BindingsEndpoint bindingsEndpoint(List inputBindings, - List outputBindings) { - return new BindingsEndpoint(inputBindings, outputBindings); + public BindingsEndpoint bindingsEndpoint(BindingsLifecycleController bindingsLifecycleController) { + return new BindingsEndpoint(bindingsLifecycleController); } } 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 index bb4c35c6d..6a31f9dba 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2021 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,22 +16,15 @@ package org.springframework.cloud.stream.endpoint; -import java.util.ArrayList; -import java.util.Collection; import java.util.List; -import java.util.stream.Stream; -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.Selector; 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.cloud.stream.binding.OutputBindingLifecycle; -import org.springframework.util.Assert; +import org.springframework.cloud.stream.binding.BindingsLifecycleController; +import org.springframework.cloud.stream.binding.BindingsLifecycleController.State; /** * @@ -43,111 +36,25 @@ import org.springframework.util.Assert; @Endpoint(id = "bindings") public class BindingsEndpoint { - private final List inputBindingLifecycles; + private final BindingsLifecycleController lifecycleController; - private final List outputBindingsLifecycles; - - private final ObjectMapper objectMapper; - - public BindingsEndpoint(List inputBindingLifecycles, - List outputBindingsLifecycles) { - Assert.notEmpty(inputBindingLifecycles, - "'inputBindingLifecycles' must not be null or empty"); - this.inputBindingLifecycles = inputBindingLifecycles; - this.outputBindingsLifecycles = outputBindingsLifecycles; - this.objectMapper = new ObjectMapper(); + public BindingsEndpoint(BindingsLifecycleController lifecycleController) { + this.lifecycleController = lifecycleController; } @WriteOperation public void changeState(@Selector String name, State state) { - Binding binding = BindingsEndpoint.this.locateBinding(name); - if (binding != null) { - switch (state) { - case STARTED: - binding.start(); - break; - case STOPPED: - binding.stop(); - break; - case PAUSED: - binding.pause(); - break; - case RESUMED: - binding.resume(); - break; - default: - break; - } - } + this.lifecycleController.changeState(name, state); } @ReadOperation public List queryStates() { - List> bindings = new ArrayList<>(gatherInputBindings()); - bindings.addAll(gatherOutputBindings()); - return this.objectMapper.convertValue(bindings, List.class); + return this.lifecycleController.queryStates(); } @ReadOperation public Binding queryState(@Selector String name) { - Assert.notNull(name, "'name' must not be null"); - return this.locateBinding(name); - } - - @SuppressWarnings("unchecked") - private List> gatherInputBindings() { - List> inputBindings = new ArrayList<>(); - for (InputBindingLifecycle inputBindingLifecycle : this.inputBindingLifecycles) { - Collection> lifecycleInputBindings = (Collection>) new DirectFieldAccessor( - inputBindingLifecycle).getPropertyValue("inputBindings"); - inputBindings.addAll(lifecycleInputBindings); - } - return inputBindings; - } - - @SuppressWarnings("unchecked") - private List> gatherOutputBindings() { - List> outputBindings = new ArrayList<>(); - for (OutputBindingLifecycle inputBindingLifecycle : this.outputBindingsLifecycles) { - Collection> lifecycleInputBindings = (Collection>) new DirectFieldAccessor( - inputBindingLifecycle).getPropertyValue("outputBindings"); - outputBindings.addAll(lifecycleInputBindings); - } - return outputBindings; - } - - private Binding locateBinding(String name) { - Stream> bindings = Stream.concat(this.gatherInputBindings().stream(), - this.gatherOutputBindings().stream()); - return bindings.filter(binding -> name.equals(binding.getBindingName())).findFirst() - .orElse(null); - } - - /** - * Binding states. - */ - public enum State { - - /** - * Started state of a binding. - */ - STARTED, - - /** - * Stopped state of a binding. - */ - STOPPED, - - /** - * Paused state of a binding. - */ - PAUSED, - - /** - * Resumed state of a binding. - */ - RESUMED; - + return this.lifecycleController.queryState(name); } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java index ba2af76c7..a9b3ba9af 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java @@ -43,10 +43,13 @@ import org.springframework.cloud.function.context.FunctionType; import org.springframework.cloud.function.context.config.ContextFunctionCatalogAutoConfiguration; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.test.FunctionBindingTestUtils; import org.springframework.cloud.stream.binder.test.InputDestination; import org.springframework.cloud.stream.binder.test.OutputDestination; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; +import org.springframework.cloud.stream.binding.BindingsLifecycleController; +import org.springframework.cloud.stream.binding.BindingsLifecycleController.State; import org.springframework.cloud.stream.messaging.Sink; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; @@ -78,6 +81,23 @@ public class ImplicitFunctionBindingTests { System.clearProperty("spring.cloud.function.definition"); } + @SuppressWarnings({"rawtypes" }) + @Test + public void testBindingControl() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(SendToDestinationConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false")) { + + BindingsLifecycleController ctrl = context.getBean(BindingsLifecycleController.class); + Binding input = ctrl.queryState("echo-in-0"); + Binding output = ctrl.queryState("echo-out-0"); + assertThat(input.isRunning()).isTrue(); + ctrl.changeState("echo-in-0", State.STOPPED); + assertThat(input.isRunning()).isFalse(); + } + } + @SuppressWarnings({ "unchecked", "rawtypes" }) @Test public void dynamicBindingTestWithFunctionRegistrationAndExplicitDestination() {