GH-3071 Add support for accessing producer/consumer properties

This commit is contained in:
Oleg Zhurakousky
2025-01-28 09:36:58 +01:00
parent fcd7cba51d
commit adddeffbf3
3 changed files with 43 additions and 1 deletions

View File

@@ -368,6 +368,14 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
return doGetExtendedInfo(destination, producerProperties);
}
@SuppressWarnings({ "unchecked", "hiding" })
public <P> P getExtension() {
if (producerProperties instanceof ExtendedProducerProperties extendedProperties) {
return (P) extendedProperties.getExtension();
}
return null;
}
@Override
public boolean isInput() {
return false;
@@ -551,6 +559,14 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
return doGetExtendedInfo(destination, properties);
}
@SuppressWarnings({ "unchecked", "hiding" })
public <P> P getExtension() {
if (properties instanceof ExtendedConsumerProperties extendedProperties) {
return (P) extendedProperties.getExtension();
}
return null;
}
@Override
public boolean isInput() {
return true;

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2023 the original author or authors.
* Copyright 2016-2025 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.
@@ -41,6 +41,15 @@ public interface Binding<T> extends Pausable {
return Collections.emptyMap();
}
/**
* Will return producer or consumer properties of this binding.
* @param <P> type of producer or consumer properties
* @return producer or consumer properties of this binding
*/
default <P> P getExtension() {
return null;
}
/**
* Starts the target component represented by this instance. NOTE: At the time the
* instance is created the component is already started. This operation is typically

View File

@@ -31,6 +31,7 @@ import org.springframework.beans.DirectFieldAccessor;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;
/**
*
@@ -52,6 +53,7 @@ public class BindingsLifecycleController {
@SuppressWarnings("unchecked")
public BindingsLifecycleController(List<InputBindingLifecycle> inputBindingLifecycles,
List<OutputBindingLifecycle> outputBindingsLifecycles) {
Assert.notEmpty(inputBindingLifecycles,
"'inputBindingLifecycles' must not be null or empty");
this.inputBindingLifecycles = inputBindingLifecycles;
@@ -72,6 +74,21 @@ public class BindingsLifecycleController {
}
}
/**
* Will return producer or consumer properties for a specified binding. For example, calling `getExtensionProperties("foo-in-0")`
* on Kafka binding will return an instance of KafkaConsumerProperties.
* @param <T> type of producer or consumer properties for a specified binding
* @param bindingName name of the binding
* @return producer or consumer properties
*/
public <T> T getExtensionProperties(String bindingName) {
List<Binding<?>> locateBinding = BindingsLifecycleController.this.locateBinding(bindingName);
if (!CollectionUtils.isEmpty(locateBinding)) {
return locateBinding.get(0).getExtension();
}
return null;
}
/**
* Provide an accessor for the custom ObjectMapper created by this controller.
* @return {@link ObjectMapper}