From f024ba2bb0737e2a2ad37e0ced1d7486d7472ee0 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Fri, 19 Mar 2021 14:00:01 +0100 Subject: [PATCH] GH-1907 Decouple binding control from actuator This allows for easy programmatic control of binding lifecycle. The details are documented and as well as javadoced Resolves #1907 --- .../main/asciidoc/spring-cloud-stream.adoc | 22 +- .../binding/BindingsLifecycleController.java | 202 ++++++++++++++++++ .../config/BindingServiceConfiguration.java | 7 + .../BindingsEndpointAutoConfiguration.java | 12 +- .../stream/endpoint/BindingsEndpoint.java | 111 +--------- .../ImplicitFunctionBindingTests.java | 20 ++ 6 files changed, 262 insertions(+), 112 deletions(-) create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingsLifecycleController.java diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index d01b0aa97..a520d2ca5 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -1833,9 +1833,27 @@ environment: ---- [[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 438d29714..a0b19b89b 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 @@ -43,6 +43,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; @@ -235,6 +236,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 330b4a525..093b7a73f 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; @@ -76,6 +79,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() {