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

GH-1907 Decouple binding control from actuator

Browse Source 
This allows for easy programmatic control of binding lifecycle. The details are documented and as well as javadoced
Resolves #1907
This commit is contained in:
Oleg Zhurakousky
2021-03-19 14:00:01 +01:00
parent 533609ce74
commit 5cd14e4a8d
6 changed files with 262 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
docs/src/main/asciidoc/spring-cloud-stream.adoc
View File

@@ -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]

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

@@ -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<InputBindingLifecycle> inputBindingLifecycles;
private final List<OutputBindingLifecycle> outputBindingsLifecycles;
private final ObjectMapper objectMapper;
public BindingsLifecycleController(List<InputBindingLifecycle> inputBindingLifecycles,
List<OutputBindingLifecycle> 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<Binding<?>> queryStates() {
List<Binding<?>> 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<Binding<?>> gatherInputBindings() {
List<Binding<?>> inputBindings = new ArrayList<>();
for (InputBindingLifecycle inputBindingLifecycle : this.inputBindingLifecycles) {
Collection<Binding<?>> lifecycleInputBindings = (Collection<Binding<?>>) 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<Binding<?>> gatherOutputBindings() {
List<Binding<?>> outputBindings = new ArrayList<>();
for (OutputBindingLifecycle inputBindingLifecycle : this.outputBindingsLifecycles) {
Collection<Binding<?>> lifecycleInputBindings = (Collection<Binding<?>>) new DirectFieldAccessor(
inputBindingLifecycle).getPropertyValue("outputBindings");
outputBindings.addAll(lifecycleInputBindings);
}
return outputBindings;
}
private Binding<?> locateBinding(String name) {
Stream<Binding<?>> 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;
}
}

7
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java
View File

@@ -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<InputBindingLifecycle> inputBindingLifecycles,
List<OutputBindingLifecycle> outputBindingsLifecycles) {
return new BindingsLifecycleController(inputBindingLifecycles, outputBindingsLifecycles);
}
@Bean
@DependsOn("bindingService")
public ContextStartAfterRefreshListener contextStartAfterRefreshListener() {

12
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingsEndpointAutoConfiguration.java
View File

@@ -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<InputBindingLifecycle> inputBindings,
List<OutputBindingLifecycle> outputBindings) {
return new BindingsEndpoint(inputBindings, outputBindings);
public BindingsEndpoint bindingsEndpoint(BindingsLifecycleController bindingsLifecycleController) {
return new BindingsEndpoint(bindingsLifecycleController);
}
}

111
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/endpoint/BindingsEndpoint.java
View File

@@ -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<InputBindingLifecycle> inputBindingLifecycles;
private final BindingsLifecycleController lifecycleController;
private final List<OutputBindingLifecycle> outputBindingsLifecycles;
private final ObjectMapper objectMapper;
public BindingsEndpoint(List<InputBindingLifecycle> inputBindingLifecycles,
List<OutputBindingLifecycle> 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<Binding<?>> 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<Binding<?>> gatherInputBindings() {
List<Binding<?>> inputBindings = new ArrayList<>();
for (InputBindingLifecycle inputBindingLifecycle : this.inputBindingLifecycles) {
Collection<Binding<?>> lifecycleInputBindings = (Collection<Binding<?>>) new DirectFieldAccessor(
inputBindingLifecycle).getPropertyValue("inputBindings");
inputBindings.addAll(lifecycleInputBindings);
}
return inputBindings;
}
@SuppressWarnings("unchecked")
private List<Binding<?>> gatherOutputBindings() {
List<Binding<?>> outputBindings = new ArrayList<>();
for (OutputBindingLifecycle inputBindingLifecycle : this.outputBindingsLifecycles) {
Collection<Binding<?>> lifecycleInputBindings = (Collection<Binding<?>>) new DirectFieldAccessor(
inputBindingLifecycle).getPropertyValue("outputBindings");
outputBindings.addAll(lifecycleInputBindings);
}
return outputBindings;
}
private Binding<?> locateBinding(String name) {
Stream<Binding<?>> 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);
}
}

20
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
View File

@@ -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() {