GH-1055 Added support for Pausable

Added support for Pausable lifecycle to Binding and its implementation
Added additional actuator endpoints

Addressed PR comments

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<T> extends Lifecycle {
+public interface Binding<T> extends Pausable {
 	
 	/**
 	 * Stops the target component represented by this instance.
@@ -52,6 +52,26 @@ public interface Binding<T> 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.
 	 */

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<T> implements Binding<T> {
 			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() {

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();
+	}
+	
 }

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<InputBindingLifecycle> 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<Binding<?>> gatherInputBindings() {
 		List<Binding<?>> 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;
+		}
+	}
 
 }