From a6fb3c3dbd07f47b5e93eee9ba77733ac22c5795 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 27 Feb 2018 20:05:36 -0500 Subject: [PATCH] GH-1054 Added support for display extra binding information Added support for displaying - state - pausable - name/group - destination info - extended binding info - etc Resolves #1054 --- .../binder/AbstractMessageChannelBinder.java | 32 ++++++++++- .../cloud/stream/binder/Binding.java | 15 +++++- .../cloud/stream/binder/DefaultBinding.java | 53 +++++++++++++++---- 3 files changed, 85 insertions(+), 15 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index a9cb100b2..91f6a94d9 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -16,6 +16,11 @@ package org.springframework.cloud.stream.binder; +import java.util.LinkedHashMap; +import java.util.Map; + +import com.fasterxml.jackson.databind.ObjectMapper; + import org.apache.commons.logging.Log; import org.springframework.beans.factory.DisposableBean; @@ -71,6 +76,8 @@ public abstract class AbstractMessageChannelBinder(destination, null, outputChannel, producerMessageHandler instanceof Lifecycle ? (Lifecycle) producerMessageHandler : null) { + @Override + public Map getExtendedInfo() { + return doGetExtendedInfo(destination, producerProperties); + } + @Override public void afterUnbind() { try { @@ -255,7 +267,12 @@ public abstract class AbstractMessageChannelBinder(name, group, inputChannel, consumerEndpoint instanceof Lifecycle ? (Lifecycle) consumerEndpoint : null) { - + + @Override + public Map getExtendedInfo() { + return doGetExtendedInfo(destination, properties); + } + @Override protected void afterUnbind() { try { @@ -289,7 +306,6 @@ public abstract class AbstractMessageChannelBinder> bindPollableConsumer(String name, String group, final PollableSource inboundBindTarget, C properties) { @@ -322,6 +338,11 @@ public abstract class AbstractMessageChannelBinder>(name, group, inboundBindTarget, resources.getSource() instanceof Lifecycle ? (Lifecycle) resources.getSource() : null) { + + @Override + public Map getExtendedInfo() { + return doGetExtendedInfo(destination, properties); + } @Override public void afterUnbind() { @@ -638,6 +659,13 @@ public abstract class AbstractMessageChannelBinder doGetExtendedInfo(Object destination, Object properties) { + Map extendedInfo = new LinkedHashMap<>(); + extendedInfo.put("bindingDestination", destination.toString()); + extendedInfo.put(properties.getClass().getSimpleName(), objectMapper.convertValue(properties, Map.class)); + return extendedInfo; + } private final class SendingHandler extends AbstractMessageHandler implements Lifecycle { 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 57b0b8723..093b619e5 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,6 +16,9 @@ package org.springframework.cloud.stream.binder; +import java.util.Collections; +import java.util.Map; + import org.springframework.integration.endpoint.Pausable; /** @@ -34,6 +37,10 @@ import org.springframework.integration.endpoint.Pausable; */ public interface Binding extends Pausable { + default Map getExtendedInfo() { + return Collections.emptyMap(); + } + /** * Stops the target component represented by this instance. * NOTE: At the time the instance is created the component is already started. @@ -60,7 +67,9 @@ public interface Binding extends Pausable { * * @see BindingsEndpoint */ - default void pause() {} + default void pause() { + this.stop(); + } /** * Resumes the target component represented by this instance if and only if the component @@ -70,7 +79,9 @@ public interface Binding extends Pausable { * * @see BindingsEndpoint */ - default void resume() {} + default void resume() { + this.start(); + } /** * 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 2151dbfaf..150f291f0 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 @@ -16,6 +16,9 @@ package org.springframework.cloud.stream.binder; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonPropertyOrder; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -36,6 +39,8 @@ import org.springframework.util.StringUtils; * * @see org.springframework.cloud.stream.annotation.EnableBinding */ +@JsonPropertyOrder({ "name", "group", "pausable", "state"}) +@JsonIgnoreProperties("running") public class DefaultBinding implements Binding { private final Log logger = LogFactory.getLog(this.getClass().getName()); @@ -47,7 +52,9 @@ public class DefaultBinding implements Binding { protected final T target; protected final Lifecycle lifecycle; - + + private boolean paused; + /** * Creates an instance that associates a given name, group and binding target with an * optional {@link Lifecycle} component, which will be stopped during unbinding. @@ -55,8 +62,9 @@ public class DefaultBinding implements Binding { * @param name the name of the binding target * @param group the group (only for input targets) * @param target the binding target - * @param lifecycle {@link Lifecycle} that runs while the binding is active and will - * be stopped during unbinding + * @param lifecycle {@link Lifecycle} that runs while the binding is active and will be stopped during unbinding + * @param extBindingInfo additional information related to binding + * */ public DefaultBinding(String name, String group, T target, Lifecycle lifecycle) { Assert.notNull(target, "target must not be null"); @@ -65,7 +73,7 @@ public class DefaultBinding implements Binding { this.target = target; this.lifecycle = lifecycle; } - + public String getName() { return this.name; } @@ -74,10 +82,27 @@ public class DefaultBinding implements Binding { return this.group; } + public String getState() { + String state = "N/A"; + if (this.lifecycle != null) { + if (isPausable()) { + state = this.paused ? "paused" : this.getRunningState(); + } + else { + state = this.getRunningState(); + } + } + return state; + } + public boolean isRunning() { return this.lifecycle != null && this.lifecycle.isRunning(); } + public boolean isPausable() { + return this.lifecycle instanceof Pausable; + } + @Override public final synchronized void start() { if (!this.isRunning()) { @@ -101,6 +126,7 @@ public class DefaultBinding implements Binding { public final synchronized void pause() { if (this.lifecycle instanceof Pausable) { ( (Pausable) this.lifecycle).pause(); + this.paused = true; } else { logger.warn("Attempted to pause a component that does not support Pausable " + this.lifecycle); @@ -111,6 +137,7 @@ public class DefaultBinding implements Binding { public final synchronized void resume() { if (this.lifecycle instanceof Pausable) { ( (Pausable) this.lifecycle).resume(); + this.paused = false; } else { logger.warn("Attempted to resume a component that does not support Pausable " + this.lifecycle); @@ -127,13 +154,6 @@ public class DefaultBinding implements Binding { return this.lifecycle; } - /** - * Listener method that executes after unbinding. Subclasses can implement their own - * behaviour on unbinding by overriding this method. - */ - protected void afterUnbind() { - } - @Override public String toString() { return " Binding [name=" + this.name + ", target=" + this.target + ", lifecycle=" @@ -142,4 +162,15 @@ public class DefaultBinding implements Binding { : ObjectUtils.nullSafeToString(this.lifecycle)) + "]"; } + + /** + * Listener method that executes after unbinding. Subclasses can implement their own + * behaviour on unbinding by overriding this method. + */ + protected void afterUnbind() { + } + + private String getRunningState() { + return isRunning() ? "running" : "stopped"; + } }