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 79b061684..57b0b8723 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 @@ -16,7 +16,7 @@ package org.springframework.cloud.stream.binder; -import org.springframework.context.Lifecycle; +import org.springframework.integration.endpoint.Pausable; /** * Represents a binding between an input or output and an adapter endpoint that connects @@ -32,7 +32,7 @@ import org.springframework.context.Lifecycle; * * @see org.springframework.cloud.stream.annotation.EnableBinding */ -public interface Binding extends Lifecycle { +public interface Binding extends Pausable { /** * Stops the target component represented by this instance. @@ -52,6 +52,26 @@ public interface Binding extends Lifecycle { */ default void stop() {} + /** + * Pauses the target component represented by this instance if and only if the component + * implements {@link Pausable} interface + * NOTE: At the time the instance is created the component is already started and active. + * This operation is typically used by actuator to pause/resume. + * + * @see BindingsEndpoint + */ + default void pause() {} + + /** + * Resumes the target component represented by this instance if and only if the component + * implements {@link Pausable} interface + * NOTE: At the time the instance is created the component is already started and active. + * This operation is typically used by actuator to pause/resume. + * + * @see BindingsEndpoint + */ + default void resume() {} + /** * Returns 'true' if the target component represented by this instance is running. */ 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 cd2acb49f..2151dbfaf 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 @@ -20,6 +20,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.context.Lifecycle; +import org.springframework.integration.endpoint.Pausable; import org.springframework.integration.support.context.NamedComponent; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; @@ -95,6 +96,26 @@ public class DefaultBinding implements Binding { this.lifecycle.stop(); } } + + @Override + public final synchronized void pause() { + if (this.lifecycle instanceof Pausable) { + ( (Pausable) this.lifecycle).pause(); + } + else { + logger.warn("Attempted to pause a component that does not support Pausable " + this.lifecycle); + } + } + + @Override + public final synchronized void resume() { + if (this.lifecycle instanceof Pausable) { + ( (Pausable) this.lifecycle).resume(); + } + else { + logger.warn("Attempted to resume a component that does not support Pausable " + this.lifecycle); + } + } @Override public final void unbind() { 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 b33d46b50..904d98988 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 @@ -49,9 +49,19 @@ public class BindingsEndpointAutoConfiguration { return bindingEndpoint.getStopEndpoint(); } + @Bean + public BindingsEndpoint.PauseEndpoint pauseEndpoint(BindingsEndpoint bindingEndpoint) { + return bindingEndpoint.getPauseEndpoint(); + } + @Bean public BindingsEndpoint.StartEndpoint startEndpoint(BindingsEndpoint bindingEndpoint) { return bindingEndpoint.getStartEndpoint(); } + @Bean + public BindingsEndpoint.ResumeEndpoint resumeEndpoint(BindingsEndpoint bindingEndpoint) { + return bindingEndpoint.getResumeEndpoint(); + } + } 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 2bce02571..6f661573d 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 @@ -47,6 +47,10 @@ public class BindingsEndpoint { static final String START_ENPOINT_ID = BASE_ENPOINT_ID + "/start/{name}"; static final String STOP_ENPOINT_ID = BASE_ENPOINT_ID + "/stop/{name}"; + + static final String PAUSE_ENPOINT_ID = BASE_ENPOINT_ID + "/pause/{name}"; + + static final String RESUME_ENPOINT_ID = BASE_ENPOINT_ID + "/resume/{name}"; private final List inputBindingLifecycles; @@ -71,6 +75,14 @@ public class BindingsEndpoint { return new StartEndpoint(); } + public PauseEndpoint getPauseEndpoint() { + return new PauseEndpoint(); + } + + public ResumeEndpoint getResumeEndpoint() { + return new ResumeEndpoint(); + } + @SuppressWarnings("unchecked") private List> gatherInputBindings() { List> inputBindings = new ArrayList<>(); @@ -101,6 +113,18 @@ public class BindingsEndpoint { } } + @Endpoint(id = BindingsEndpoint.PAUSE_ENPOINT_ID) + public class PauseEndpoint { + @WriteOperation + public boolean pause(String name) { + Binding binding = BindingsEndpoint.this.locateBinding(name); + if (binding != null) { + binding.pause(); + } + return binding != null; + } + } + @Endpoint(id = BindingsEndpoint.START_ENPOINT_ID) public class StartEndpoint { @WriteOperation @@ -112,5 +136,17 @@ public class BindingsEndpoint { return binding != null; } } + + @Endpoint(id = BindingsEndpoint.RESUME_ENPOINT_ID) + public class ResumeEndpoint { + @WriteOperation + public boolean resume(String name) { + Binding binding = BindingsEndpoint.this.locateBinding(name); + if (binding != null) { + binding.resume(); + } + return binding != null; + } + } }