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

Fix general Observablity support as well as Rabbit Observability support

Browse Source
This commit is contained in:
Oleg Zhurakousky
2022-10-24 17:41:57 +02:00
parent 0357d7f67f
commit b86e713a58
5 changed files with 101 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/ObservationAutoConfiguration.java Normal file
View File

@@ -0,0 +1,60 @@
/*
* Copyright 2022-2022 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.rabbit.config;
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.cloud.stream.config.ListenerContainerCustomizer;
import org.springframework.cloud.stream.config.ProducerMessageHandlerCustomizer;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.messaging.MessageHandler;
@Configuration(proxyBeanMethods = false)
@ConditionalOnBean(org.springframework.boot.actuate.autoconfigure.observation.ObservationAutoConfiguration.class)
public class ObservationAutoConfiguration {
@Bean
@Order(Ordered.HIGHEST_PRECEDENCE)
ListenerContainerCustomizer<MessageListenerContainer> observedListenerContainerCustomizer(
ApplicationContext applicationContext) {
return (container, destinationName, group) -> {
if (container instanceof AbstractMessageListenerContainer) {
AbstractMessageListenerContainer abstractMessageListenerContainer = ((AbstractMessageListenerContainer) container);
abstractMessageListenerContainer.setObservationEnabled(true);
abstractMessageListenerContainer.setApplicationContext(applicationContext);
}
};
}
@Bean
@Order(Ordered.HIGHEST_PRECEDENCE)
ProducerMessageHandlerCustomizer<MessageHandler> observedProducerMessageHandlerCustomizer(
ApplicationContext applicationContext) {
return (handler, destinationName) -> {
if (handler instanceof AmqpOutboundEndpoint) {
((AmqpOutboundEndpoint) handler).getRabbitTemplate().setObservationEnabled(true);
((AmqpOutboundEndpoint) handler).getRabbitTemplate().setApplicationContext(applicationContext);
}
};
}
}

31
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java
View File

@@ -44,8 +44,9 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
import org.springframework.integration.amqp.inbound.AmqpMessageSource;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.CollectionUtils;
/**
@@ -78,9 +79,9 @@ public class RabbitMessageChannelBinderConfiguration {
@Bean
RabbitMessageChannelBinder rabbitMessageChannelBinder(
@Nullable ListenerContainerCustomizer<MessageListenerContainer> listenerContainerCustomizer,
@Nullable List<ListenerContainerCustomizer<MessageListenerContainer>> listenerContainerCustomizers,
@Nullable MessageSourceCustomizer<AmqpMessageSource> sourceCustomizer,
@Nullable ProducerMessageHandlerCustomizer<AmqpOutboundEndpoint> producerMessageHandlerCustomizer,
@Nullable List<ProducerMessageHandlerCustomizer<MessageHandler>> producerMessageHandlerCustomizers,
@Nullable ConsumerEndpointCustomizer<AmqpInboundChannelAdapter> consumerCustomizer,
List<DeclarableCustomizer> declarableCustomizers,
@Nullable ConnectionNameStrategy connectionNameStrategy) {
@@ -91,9 +92,31 @@ public class RabbitMessageChannelBinderConfiguration {
((AbstractConnectionFactory) this.rabbitConnectionFactory).setConnectionNameStrategy(f -> connectionNamePrefix
+ "#" + nameIncrementer.getAndIncrement());
}
ListenerContainerCustomizer<MessageListenerContainer> composedCistomizer = new ListenerContainerCustomizer<>() {
@Override
public void configure(MessageListenerContainer container, String destinationName, String group) {
if (!CollectionUtils.isEmpty(listenerContainerCustomizers)) {
for (ListenerContainerCustomizer<MessageListenerContainer> customizer : listenerContainerCustomizers) {
customizer.configure(container, destinationName, group);
}
}
}
};
ProducerMessageHandlerCustomizer<MessageHandler> producerMessageHandlerCustomizer = new ProducerMessageHandlerCustomizer<>() {
@Override
public void configure(MessageHandler handler, String destinationName) {
if (!CollectionUtils.isEmpty(producerMessageHandlerCustomizers)) {
for (ProducerMessageHandlerCustomizer<MessageHandler> customizer : producerMessageHandlerCustomizers) {
customizer.configure(handler, destinationName);
}
}
}
};
RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(
this.rabbitConnectionFactory, this.rabbitProperties,
provisioningProvider(declarableCustomizers), listenerContainerCustomizer, sourceCustomizer);
provisioningProvider(declarableCustomizers), composedCistomizer, sourceCustomizer);
binder.setAdminAddresses(
this.rabbitBinderConfigurationProperties.getAdminAddresses());
binder.setCompressingPostProcessor(gZipPostProcessor());

1
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
View File

@@ -1 +1,2 @@
org.springframework.cloud.stream.binder.rabbit.config.ExtendedBindingHandlerMappingsProviderConfiguration
org.springframework.cloud.stream.binder.rabbit.config.ObservationAutoConfiguration

17
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java
View File

@@ -71,9 +71,10 @@ import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
import org.springframework.integration.amqp.inbound.AmqpMessageSource;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
@@ -421,8 +422,10 @@ public class RabbitBinderModuleTests {
@Bean
public ListenerContainerCustomizer<MessageListenerContainer> containerCustomizer() {
return (c, q, g) -> ((AbstractMessageListenerContainer) c).setBeanName(
"setByCustomizerForQueue:" + q + (g == null ? "" : ",andGroup:" + g));
return (c, q, g) -> {
((AbstractMessageListenerContainer) c).setBeanName(
"setByCustomizerForQueue:" + q + (g == null ? "" : ",andGroup:" + g));
};
}
@Bean
@@ -431,8 +434,12 @@ public class RabbitBinderModuleTests {
}
@Bean
public ProducerMessageHandlerCustomizer<AmqpOutboundEndpoint> messageHandlerCustomizer() {
return (handler, destinationName) -> handler.setBeanName("setByCustomizer:" + destinationName);
public ProducerMessageHandlerCustomizer<MessageHandler> messageHandlerCustomizer() {
return (handler, destinationName) -> {
if (handler instanceof AbstractMessageHandler) {
((AbstractMessageHandler) handler).setBeanName("setByCustomizer:" + destinationName);
}
};
}
@Bean

3
core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
View File

@@ -1475,8 +1475,7 @@ public class ImplicitFunctionBindingTests {
return new FunctionAroundWrapper() {
@Override
protected Object doApply(Object input,
FunctionInvocationWrapper targetFunction) {
protected Object doApply(Object input, FunctionInvocationWrapper targetFunction) {
return targetFunction.apply(input);
}
};