GH-2664 Add support for displaying binder name/type
Resolves #2664 GH-2664 polishing
This commit is contained in:
@@ -35,7 +35,6 @@ import org.junit.jupiter.api.Disabled;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.params.ParameterizedTest;
|
||||
import org.junit.jupiter.params.provider.ValueSource;
|
||||
import org.springframework.cloud.function.context.FunctionCatalog;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
import reactor.core.publisher.Sinks;
|
||||
@@ -45,6 +44,7 @@ import org.springframework.beans.factory.BeanCreationException;
|
||||
import org.springframework.boot.WebApplicationType;
|
||||
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
|
||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||
import org.springframework.cloud.function.context.FunctionCatalog;
|
||||
import org.springframework.cloud.function.context.FunctionRegistration;
|
||||
import org.springframework.cloud.function.context.catalog.FunctionAroundWrapper;
|
||||
import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
|
||||
|
||||
@@ -308,6 +308,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
|
||||
throw new IllegalStateException("No capable binding targets found.");
|
||||
}
|
||||
|
||||
|
||||
BindingServiceProperties bsp = this.getBindingServiceProperties();
|
||||
Binding<MessageChannel> binding = new DefaultBinding<MessageChannel>(destination,
|
||||
outputChannel, producerMessageHandler instanceof Lifecycle producerMessageHandlerWithLifecycle
|
||||
? producerMessageHandlerWithLifecycle : null) {
|
||||
@@ -322,6 +324,16 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getBinderName() {
|
||||
return bsp == null ? null : bsp.getBinder(destination);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getBinderType() {
|
||||
return bsp == null ? null : bsp.getBinderType(this.getBinderName());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void afterUnbind() {
|
||||
try {
|
||||
@@ -480,6 +492,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
|
||||
consumerEndpointWithLifecycle.start();
|
||||
}
|
||||
|
||||
BindingServiceProperties bsp = this.getBindingServiceProperties();
|
||||
Binding<MessageChannel> binding = new DefaultBinding<MessageChannel>(name,
|
||||
group, inputChannel, consumerEndpoint instanceof Lifecycle consumerEndpointWithLifecycle
|
||||
? consumerEndpointWithLifecycle : null) {
|
||||
@@ -494,6 +507,16 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getBinderName() {
|
||||
return bsp == null ? null : bsp.getBinder(name);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getBinderType() {
|
||||
return bsp == null ? null : bsp.getBinderType(this.getBinderName());
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void afterUnbind() {
|
||||
try {
|
||||
@@ -567,6 +590,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
|
||||
if (properties.isAutoStartup() && resources.getSource() instanceof Lifecycle sourceWithLifecycle) {
|
||||
sourceWithLifecycle.start();
|
||||
}
|
||||
|
||||
BindingServiceProperties bsp = this.getBindingServiceProperties();
|
||||
Binding<PollableSource<MessageHandler>> binding = new DefaultBinding<PollableSource<MessageHandler>>(
|
||||
name, group, inboundBindTarget, resources.getSource() instanceof Lifecycle sourceWithLifecycle
|
||||
? sourceWithLifecycle : null) {
|
||||
@@ -581,6 +606,16 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getBinderName() {
|
||||
return bsp == null ? null : bsp.getBinder(name);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getBinderType() {
|
||||
return bsp == null ? null : bsp.getBinderType(this.getBinderName());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void afterUnbind() {
|
||||
afterUnbindConsumer(destination, this.group, properties);
|
||||
|
||||
@@ -48,6 +48,7 @@ public interface Binding<T> extends Pausable {
|
||||
*
|
||||
* @see BindingsEndpoint
|
||||
*/
|
||||
@Override
|
||||
default void start() {
|
||||
}
|
||||
|
||||
@@ -58,6 +59,7 @@ public interface Binding<T> extends Pausable {
|
||||
*
|
||||
* @see BindingsEndpoint
|
||||
*/
|
||||
@Override
|
||||
default void stop() {
|
||||
}
|
||||
|
||||
@@ -77,6 +79,7 @@ public interface Binding<T> extends Pausable {
|
||||
*
|
||||
* @see BindingsEndpoint
|
||||
*/
|
||||
@Override
|
||||
default void pause() {
|
||||
this.stop();
|
||||
}
|
||||
@@ -89,6 +92,7 @@ public interface Binding<T> extends Pausable {
|
||||
*
|
||||
* @see BindingsEndpoint
|
||||
*/
|
||||
@Override
|
||||
default void resume() {
|
||||
this.start();
|
||||
}
|
||||
@@ -96,6 +100,7 @@ public interface Binding<T> extends Pausable {
|
||||
/**
|
||||
* @return 'true' if the target component represented by this instance is running.
|
||||
*/
|
||||
@Override
|
||||
default boolean isRunning() {
|
||||
return false;
|
||||
}
|
||||
@@ -118,6 +123,26 @@ public interface Binding<T> extends Pausable {
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the name of the binder for this binding.
|
||||
* @return binder name
|
||||
*
|
||||
* @since 4.0.2
|
||||
*/
|
||||
default String getBinderName() {
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the type of the binder for this binding.
|
||||
* @return binder name
|
||||
*
|
||||
* @since 4.0.2
|
||||
*/
|
||||
default String getBinderType() {
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Unbinds the target component represented by this instance and stops any active
|
||||
* components. Implementations must be idempotent. After this method is invoked, the
|
||||
|
||||
@@ -237,6 +237,11 @@ public class BindingServiceProperties
|
||||
}
|
||||
}
|
||||
|
||||
public String getBinderType(String binderName) {
|
||||
BinderProperties bp = this.binders.get(binderName);
|
||||
return bp != null ? bp.getType() : this.bindings.keySet().iterator().next();
|
||||
}
|
||||
|
||||
public String getBinder(String bindingName) {
|
||||
return getBindingProperties(bindingName).getBinder();
|
||||
}
|
||||
|
||||
@@ -135,19 +135,23 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
|
||||
this.streamBridgeFunctionCache = new HashMap<>();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean send(String bindingName, Object data) {
|
||||
BindingProperties bindingProperties = this.bindingServiceProperties.getBindingProperties(bindingName);
|
||||
MimeType contentType = StringUtils.hasText(bindingProperties.getContentType()) ? MimeType.valueOf(bindingProperties.getContentType()) : MimeTypeUtils.APPLICATION_JSON;
|
||||
return this.send(bindingName, data, contentType);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean send(String bindingName, Object data, MimeType outputContentType) {
|
||||
return this.send(bindingName, null, data, outputContentType);
|
||||
}
|
||||
@Override
|
||||
public boolean send(String bindingName, @Nullable String binderName, Object data) {
|
||||
return this.send(bindingName, binderName, data, MimeTypeUtils.APPLICATION_JSON);
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings({ "unchecked"})
|
||||
public boolean send(String bindingName, @Nullable String binderName, Object data, MimeType outputContentType) {
|
||||
if (!this.initialized) {
|
||||
|
||||
Reference in New Issue
Block a user