SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Message Channel should be closed in specified cases : application shutdown, unbinding or exceeding cache size.

Browse Source 
checkstyle fixes

Resolves #2869
Resolves #3026
This commit is contained in:
Ömer Çelik
2024-11-02 19:03:26 +03:00
committed by Oleg Zhurakousky
parent a9fe0c2c3c
commit dccf8ae16a
7 changed files with 184 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java
View File

@@ -90,12 +90,13 @@ import static org.mockito.Mockito.when;
* @author Chris Bono
* @author Artem Bilan
* @author Kotaro Matsumoto
* @author Omer Celik
*/
class BindingServiceTests {
@SuppressWarnings({ "unchecked", "rawtypes" })
@Test
void defaultGroup() throws Exception {
void defaultGroup() {
BindingServiceProperties properties = new BindingServiceProperties();
Map<String, BindingProperties> bindingProperties = new HashMap<>();
BindingProperties props = new BindingProperties();
@@ -175,7 +176,7 @@ class BindingServiceTests {
@SuppressWarnings({ "unchecked", "rawtypes" })
@Test
void multipleConsumerBindingsFromIndexList() throws Exception {
void multipleConsumerBindingsFromIndexList() {
BindingServiceProperties properties = new BindingServiceProperties();
Map<String, BindingProperties> bindingProperties = new HashMap<>();
BindingProperties props = new BindingProperties();
@@ -236,7 +237,7 @@ class BindingServiceTests {
@SuppressWarnings({ "unchecked", "rawtypes" })
@Test
void consumerBindingWhenMultiplexingIsEnabled() throws Exception {
void consumerBindingWhenMultiplexingIsEnabled() {
BindingServiceProperties properties = new BindingServiceProperties();
Map<String, BindingProperties> bindingProperties = new HashMap<>();
BindingProperties props = new BindingProperties();
@@ -282,7 +283,7 @@ class BindingServiceTests {
@SuppressWarnings({ "unchecked", "rawtypes" })
@Test
void explicitGroup() throws Exception {
void explicitGroup() {
BindingServiceProperties properties = new BindingServiceProperties();
Map<String, BindingProperties> bindingProperties = new HashMap<>();
BindingProperties props = new BindingProperties();
@@ -528,7 +529,7 @@ class BindingServiceTests {
assertThat(service.getProducerBinding("output")).isSameAs(binding);
service.unbindProducers(outputChannelName);
service.unbindProducers(null, outputChannelName);
verify(binder, times(2)).bindProducer(eq("foo"), same(outputChannel),
any(ProducerProperties.class));
verify(delegate).unbind();
@@ -552,9 +553,8 @@ class BindingServiceTests {
assertThat(inputBinding.isRunning()).isFalse();
}
@SuppressWarnings("unchecked")
@Test
void bindingNameAsTopLevelProperty() throws Exception {
void bindingNameAsTopLevelProperty() {
ApplicationContext context = new SpringApplicationBuilder(BarConfiguration.class)
.web(WebApplicationType.NONE).run();

42
core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java
View File

@@ -84,6 +84,7 @@ import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
*
* @author Oleg Zhurakousky
* @author Soby Chacko
* @author Omer Celik
*
*/
class StreamBridgeTests {
@@ -384,7 +385,7 @@ class StreamBridgeTests {
// See this issue for more details: https://github.com/spring-cloud/spring-cloud-stream/issues/2805
@Test
void streamBridgeSendWithBinderNameAndCustomContentType() throws Exception {
void streamBridgeSendWithBinderNameAndCustomContentType() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
.getCompleteConfiguration(ConsumerConfiguration.class, EmptyConfigurationWithCustomConverters.class))
.web(WebApplicationType.NONE).run(
@@ -767,6 +768,45 @@ class StreamBridgeTests {
assertThat(bindingService.getProducerBindingNames().length).isEqualTo(0);
}
@Test
void dynamicDestinationWithBinderNameDestroy() {
BindingService bindingService;
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
.getCompleteConfiguration(InterceptorConfiguration.class))
.web(WebApplicationType.NONE).run(
"--spring.jmx.enabled=false",
"--spring.cloud.stream.binders.kafka1.type=kafka",
"--spring.cloud.stream.binders.anotherKafka.type=kafka"
)) {
StreamBridge bridge = context.getBean(StreamBridge.class);
bridge.send("binding1", "kafka1", "Omer Celik");
bridge.send("binding2", "anotherKafka", "Omer Celik");
bindingService = context.getBean(BindingService.class);
}
assertThat(bindingService.getProducerBindingNames().length).isEqualTo(0);
}
@Test
void dynamicDestinationWithBinderNameDestroyForCacheSize() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
.getCompleteConfiguration(InterceptorConfiguration.class))
.web(WebApplicationType.NONE).run(
"--spring.jmx.enabled=false",
"--spring.cloud.stream.dynamic-destination-cache-size=1",
"--spring.cloud.stream.binders.kafka1.type=kafka",
"--spring.cloud.stream.binders.anotherKafka.type=kafka"
)) {
StreamBridge bridge = context.getBean(StreamBridge.class);
bridge.send("binding1", "kafka1", "Omer Celik");
bridge.send("binding2", "anotherKafka", "Omer Celik");
BindingService bindingService = context.getBean(BindingService.class);
assertThat(bindingService.getProducerBindingNames().length).isEqualTo(1);
assertThat(bindingService.getProducerBindingNames()[0]).isEqualTo("anotherKafka:binding2");
}
}
@Test
void withIntegrationFlowBecauseMarcinSaidSo() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration

25
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderWrapper.java Normal file
View File

@@ -0,0 +1,25 @@
/*
* Copyright 2024-2024 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder;
/**
* Holds information about the binder and channel.
*
* @author Omer Celik
*/
public record BinderWrapper(Binder binder, String destinationName, String cacheKey) {
}

2
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindableProxyFactory.java
View File

@@ -135,7 +135,7 @@ public class AbstractBindableProxyFactory implements Bindable {
for (Map.Entry<String, BoundTargetHolder> boundTargetHolderEntry : this.outputHolders
.entrySet()) {
if (boundTargetHolderEntry.getValue().bindable()) {
bindingService.unbindProducers(boundTargetHolderEntry.getKey());
bindingService.unbindProducers(null, boundTargetHolderEntry.getKey());
}
}
}

61
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java
View File

@@ -35,6 +35,7 @@ import org.springframework.aop.framework.Advised;
import org.springframework.beans.BeanUtils;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.binder.BinderWrapper;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.ConsumerProperties;
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
@@ -50,6 +51,9 @@ import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import org.springframework.validation.DataBinder;
import static org.springframework.cloud.stream.utils.CacheKeyCreatorUtils.createChannelCacheKey;
import static org.springframework.cloud.stream.utils.CacheKeyCreatorUtils.getBinderNameIfNeeded;
/**
* Handles binding of input/output targets by delegating to an underlying {@link Binder}.
*
@@ -270,47 +274,44 @@ public class BindingService {
}
@SuppressWarnings({ "unchecked", "rawtypes" })
public <T> Binding<T> bindProducer(T output, String outputName, boolean cache, @Nullable Binder<T, ?, ProducerProperties> binder) {
String bindingTarget = this.bindingServiceProperties.getBindingDestination(outputName);
Class<?> outputClass = output.getClass();
if (output instanceof Advised advisedOutput) {
outputClass = Stream.of(advisedOutput.getProxiedInterfaces()).filter(c -> !c.getName().contains("org.springframework")).findFirst()
.orElse(outputClass);
}
if (binder == null) {
binder = (Binder<T, ?, ProducerProperties>) getBinder(outputName, outputClass);
}
public <T> Binding<T> bindProducer(T output, boolean cache, BinderWrapper binderWrapper) {
ProducerProperties producerProperties = this.bindingServiceProperties
.getProducerProperties(outputName);
if (binder instanceof ExtendedPropertiesBinder extendedPropertiesBinder) {
Object extension = extendedPropertiesBinder.getExtendedProducerProperties(outputName);
.getProducerProperties(binderWrapper.destinationName());
if (binderWrapper.binder() instanceof ExtendedPropertiesBinder extendedPropertiesBinder) {
Object extension = extendedPropertiesBinder.getExtendedProducerProperties(binderWrapper.destinationName());
ExtendedProducerProperties extendedProducerProperties = new ExtendedProducerProperties<>(
extension);
BeanUtils.copyProperties(producerProperties, extendedProducerProperties);
producerProperties = extendedProducerProperties;
}
producerProperties.populateBindingName(outputName);
producerProperties.populateBindingName(binderWrapper.destinationName());
validate(producerProperties);
Binding<T> binding = doBindProducer(output, bindingTarget, binder,
String bindingTarget = this.bindingServiceProperties.getBindingDestination(binderWrapper.destinationName());
Binding<T> binding = doBindProducer(output, bindingTarget, binderWrapper.binder(),
producerProperties);
// If the downstream binder modified the partition count in the extended producer properties
// based on the higher number of partitions provisioned on the target middleware, update that
// in the original producer properties.
ProducerProperties originalProducerProperties = this.bindingServiceProperties
.getProducerProperties(outputName);
.getProducerProperties(binderWrapper.destinationName());
if (originalProducerProperties.getPartitionCount() < producerProperties.getPartitionCount()) {
originalProducerProperties.setPartitionCount(producerProperties.getPartitionCount());
}
if (cache) {
this.producerBindings.put(outputName, binding);
this.producerBindings.put(binderWrapper.cacheKey(), binding);
}
return binding;
}
public <T> Binding<T> bindProducer(T output, String outputName, boolean cache) {
return this.bindProducer(output, outputName, cache, null);
Class<?> outputClass = output.getClass();
if (output instanceof Advised advisedOutput) {
outputClass = Stream.of(advisedOutput.getProxiedInterfaces()).filter(c -> !c.getName().contains("org.springframework")).findFirst()
.orElse(outputClass);
}
BinderWrapper binderWrapper = createBinderWrapper(null, outputName, outputClass);
return this.bindProducer(output, cache, binderWrapper);
}
public <T> Binding<T> bindProducer(T output, String outputName) {
@@ -318,8 +319,7 @@ public class BindingService {
}
@SuppressWarnings("rawtypes")
public Object getExtendedProducerProperties(Object output, String outputName) {
Binder binder = getBinder(outputName, output.getClass());
public Object getExtendedProducerProperties(Binder binder, String outputName) {
if (binder instanceof ExtendedPropertiesBinder extendedPropertiesBinder) {
return extendedPropertiesBinder.getExtendedProducerProperties(outputName);
}
@@ -398,8 +398,13 @@ public class BindingService {
}
}
public void unbindProducers(String outputName) {
Binding<?> binding = this.producerBindings.remove(outputName);
public void unbindProducers(@Nullable String binderName, String outputName) {
String cacheKey = createChannelCacheKey(binderName, outputName, bindingServiceProperties);
unbindProducers(cacheKey);
}
public void unbindProducers(String cacheKey) {
Binding<?> binding = this.producerBindings.remove(cacheKey);
if (binding != null) {
binding.stop();
@@ -407,7 +412,7 @@ public class BindingService {
binding.unbind();
}
else if (this.log.isWarnEnabled()) {
this.log.warn("Trying to unbind '" + outputName + "', but no binding found.");
this.log.warn("Trying to unbind '" + cacheKey + "', but no binding found.");
}
}
@@ -443,6 +448,14 @@ public class BindingService {
}
}
public BinderWrapper createBinderWrapper(@Nullable String binderName, String destinationName, Class<?> outputClass) {
binderName = getBinderNameIfNeeded(binderName, destinationName, bindingServiceProperties);
Binder binder = binderFactory.getBinder(binderName, outputClass);
String channelCacheKey = createChannelCacheKey(binderName, destinationName);
return new BinderWrapper(binder, destinationName, channelCacheKey);
}
public static class LateBinding<T> implements Binding<T> {
private volatile Binding<T> delegate;

31
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java
View File

@@ -43,6 +43,7 @@ import org.springframework.cloud.function.context.message.MessageUtils;
import org.springframework.cloud.function.core.FunctionInvocationHelper;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.binder.BinderWrapper;
import org.springframework.cloud.stream.binder.ProducerProperties;
import org.springframework.cloud.stream.binding.BindingService;
import org.springframework.cloud.stream.binding.DefaultPartitioningInterceptor;
@@ -68,6 +69,8 @@ import org.springframework.util.MimeTypeUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import static org.springframework.cloud.stream.utils.CacheKeyCreatorUtils.createChannelCacheKey;
/**
* A class which allows user to send data to an output binding.
@@ -85,6 +88,7 @@ import org.springframework.util.StringUtils;
* @author Soby Chacko
* @author Byungjun You
* @author Michał Rowicki
* @author Omer Celik
* @since 3.0.3
*
*/
@@ -128,7 +132,6 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
* @param bindingServiceProperties instance of {@link BindingServiceProperties}
* @param applicationContext instance of {@link ConfigurableApplicationContext}
*/
@SuppressWarnings("serial")
StreamBridge(FunctionCatalog functionCatalog, BindingServiceProperties bindingServiceProperties,
ConfigurableApplicationContext applicationContext, @Nullable NewDestinationBindingCallback destinationBindingCallback) {
this.executorService = Executors.newCachedThreadPool();
@@ -268,13 +271,7 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
MessageChannel resolveDestination(String destinationName, ProducerProperties producerProperties, String binderName) {
lock.lock();
try {
MessageChannel messageChannel = null;
if (StringUtils.hasText(binderName)) {
messageChannel = this.channelCache.get(binderName + ":" + destinationName);
}
else {
messageChannel = this.channelCache.get(destinationName);
}
MessageChannel messageChannel = this.channelCache.get(createChannelCacheKey(binderName, destinationName, bindingServiceProperties));
if (messageChannel == null) {
if (this.applicationContext.containsBean(destinationName)) {
messageChannel = this.applicationContext.getBean(destinationName, MessageChannel.class);
@@ -291,28 +288,20 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
messageChannel = this.isAsync() ? new ExecutorChannel(this.executorService) : new DirectWithAttributesChannel();
((AbstractSubscribableChannel) messageChannel).setApplicationContext(applicationContext);
((AbstractSubscribableChannel) messageChannel).setComponentName(destinationName);
BinderWrapper binderWrapper = bindingService.createBinderWrapper(binderName, destinationName, messageChannel.getClass());
if (this.destinationBindingCallback != null) {
Object extendedProducerProperties = this.bindingService
.getExtendedProducerProperties(messageChannel, destinationName);
.getExtendedProducerProperties(binderWrapper.binder(), destinationName);
this.destinationBindingCallback.configure(destinationName, messageChannel,
producerProperties, extendedProducerProperties);
}
Binder binder = null;
if (StringUtils.hasText(binderName)) {
BinderFactory binderFactory = this.applicationContext.getBean(BinderFactory.class);
binder = binderFactory.getBinder(binderName, messageChannel.getClass());
}
addPartitioningInterceptorIfNeedBe(producerProperties, destinationName, (AbstractMessageChannel) messageChannel);
addGlobalChannelInterceptorProcessor((AbstractMessageChannel) messageChannel, destinationName);
this.bindingService.bindProducer(messageChannel, destinationName, true, binder);
if (StringUtils.hasText(binderName)) {
this.channelCache.put(binderName + ":" + destinationName, messageChannel);
}
else {
this.channelCache.put(destinationName, messageChannel);
}
this.bindingService.bindProducer(messageChannel, true, binderWrapper);
this.channelCache.put(binderWrapper.cacheKey(), messageChannel);
}
}

63
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/utils/CacheKeyCreatorUtils.java Normal file
View File

@@ -0,0 +1,63 @@
/*
* Copyright 2024-2024 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.utils;
import java.util.Objects;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.lang.Nullable;
import org.springframework.util.StringUtils;
/**
* Utility methods related to Output Channel Cache Key.
*
* @author Omer Celik
*/
public final class CacheKeyCreatorUtils {
private CacheKeyCreatorUtils() {
}
public static String createChannelCacheKey(@Nullable String binderName, String outputName,
BindingServiceProperties bindingServiceProperties) {
String finalBinderName = getBinderNameIfNeeded(binderName, outputName, bindingServiceProperties);
return createChannelCacheKey(finalBinderName, outputName);
}
public static String getBinderNameIfNeeded(@Nullable String binderName, String outputName,
BindingServiceProperties bindingServiceProperties) {
if (Objects.isNull(binderName)) {
return bindingServiceProperties.getBinder(outputName);
}
return binderName;
}
public static String createChannelCacheKey(String outputName, BindingServiceProperties bindingServiceProperties) {
String binderName = bindingServiceProperties.getBinder(outputName);
return createChannelCacheKey(binderName, outputName);
}
public static String createChannelCacheKey(String binderName, String outputName) {
if (StringUtils.hasText(binderName)) {
return binderName + ":" + outputName;
}
else {
return outputName;
}
}
}