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

GH-2215 Ensure internal components reuse ObjectMapper

Browse Source 
Resolves #2215
This commit is contained in:
Oleg Zhurakousky
2021-10-27 19:06:43 +02:00
parent 662d126e4f
commit 0a675e3534
6 changed files with 54 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
View File

@@ -65,6 +65,7 @@ import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.retry.RecoveryCallback;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
/**
* {@link AbstractBinder} that serves as base class for {@link MessageChannel} binders.
@@ -101,7 +102,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
private final EmbeddedHeadersChannelInterceptor embeddedHeadersChannelInterceptor = new EmbeddedHeadersChannelInterceptor(
this.logger);
private final ObjectMapper objectMapper = new ObjectMapper();
private volatile ObjectMapper objectMapper;
/**
* Indicates which headers are to be embedded in the payload if a binding requires
@@ -126,14 +127,23 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
this(headersToEmbed, provisioningProvider, null, null);
}
@Override
protected void onInit() throws Exception {
if (!CollectionUtils.isEmpty(this.getApplicationContext().getBeansOfType(ObjectMapper.class))) {
this.objectMapper = this.getApplicationContext().getBean(ObjectMapper.class);
}
else {
this.objectMapper = new ObjectMapper();
}
SimpleModule module = new SimpleModule();
module.addSerializer(Expression.class, new ExpressionSerializer(Expression.class));
this.objectMapper.registerModule(module);
}
public AbstractMessageChannelBinder(String[] headersToEmbed, PP provisioningProvider,
@Nullable ListenerContainerCustomizer<?> containerCustomizer,
@Nullable MessageSourceCustomizer<?> sourceCustomizer) {
SimpleModule module = new SimpleModule();
module.addSerializer(Expression.class, new ExpressionSerializer(Expression.class));
objectMapper.registerModule(module);
this.headersToEmbed = headersToEmbed == null ? new String[0] : headersToEmbed;
this.provisioningProvider = provisioningProvider;
this.containerCustomizer = containerCustomizer == null ? (c, q, g) -> {

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

@@ -78,18 +78,21 @@ public class BindingService {
private final BinderFactory binderFactory;
private final ObjectMapper objectMapper;
public BindingService(BindingServiceProperties bindingServiceProperties,
BinderFactory binderFactory) {
this(bindingServiceProperties, binderFactory, null);
BinderFactory binderFactory, ObjectMapper objectMapper) {
this(bindingServiceProperties, binderFactory, null, objectMapper);
}
public BindingService(BindingServiceProperties bindingServiceProperties,
BinderFactory binderFactory, TaskScheduler taskScheduler) {
BinderFactory binderFactory, TaskScheduler taskScheduler, ObjectMapper objectMapper) {
this.bindingServiceProperties = bindingServiceProperties;
this.binderFactory = binderFactory;
this.validator = new CustomValidatorBean();
this.validator.afterPropertiesSet();
this.taskScheduler = taskScheduler;
this.objectMapper = objectMapper;
}
@SuppressWarnings({ "unchecked", "rawtypes" })
@@ -180,7 +183,7 @@ public class BindingService {
}
catch (RuntimeException e) {
LateBinding<T> late = new LateBinding<T>(target,
e.getCause() == null ? e.toString() : e.getCause().getMessage(), consumerProperties, true);
e.getCause() == null ? e.toString() : e.getCause().getMessage(), consumerProperties, true, this.objectMapper);
rescheduleConsumerBinding(input, inputName, binder, consumerProperties,
target, late, e);
this.consumerBindings.put(inputName, Collections.singletonList(late));
@@ -228,7 +231,7 @@ public class BindingService {
}
catch (RuntimeException e) {
LateBinding<T> late = new LateBinding<T>(target,
e.getCause() == null ? e.toString() : e.getCause().getMessage(), consumerProperties, true);
e.getCause() == null ? e.toString() : e.getCause().getMessage(), consumerProperties, true, this.objectMapper);
reschedulePollableConsumerBinding(input, inputName, binder,
consumerProperties, target, late, e);
return late;
@@ -321,7 +324,7 @@ public class BindingService {
}
catch (RuntimeException e) {
LateBinding<T> late = new LateBinding<T>(bindingTarget,
e.getCause() == null ? e.toString() : e.getCause().getMessage(), producerProperties, false);
e.getCause() == null ? e.toString() : e.getCause().getMessage(), producerProperties, false, this.objectMapper);
rescheduleProducerBinding(output, bindingTarget, binder,
producerProperties, late, e);
return late;
@@ -422,14 +425,15 @@ public class BindingService {
private final boolean isInput;
ObjectMapper mapper = new ObjectMapper();
final ObjectMapper objectMapper;
LateBinding(String bindingName, String error, Object consumerOrProducerproperties, boolean isInput) {
LateBinding(String bindingName, String error, Object consumerOrProducerproperties, boolean isInput, ObjectMapper objectMapper) {
super();
this.error = error;
this.bindingName = bindingName;
this.consumerOrProducerproperties = consumerOrProducerproperties;
this.isInput = isInput;
this.objectMapper = objectMapper;
}
public synchronized void setDelegate(Binding<T> delegate) {
@@ -474,7 +478,7 @@ public class BindingService {
Map<String, Object> extendedInfo = new LinkedHashMap<>();
extendedInfo.put("bindingDestination", this.getBindingName());
extendedInfo.put(consumerOrProducerproperties.getClass().getSimpleName(),
mapper.convertValue(consumerOrProducerproperties, Map.class));
this.objectMapper.convertValue(consumerOrProducerproperties, Map.class));
return extendedInfo;
}

4
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingsLifecycleController.java
View File

@@ -44,12 +44,12 @@ public class BindingsLifecycleController {
private final ObjectMapper objectMapper;
public BindingsLifecycleController(List<InputBindingLifecycle> inputBindingLifecycles,
List<OutputBindingLifecycle> outputBindingsLifecycles) {
List<OutputBindingLifecycle> outputBindingsLifecycles, ObjectMapper objectMapper) {
Assert.notEmpty(inputBindingLifecycles,
"'inputBindingLifecycles' must not be null or empty");
this.inputBindingLifecycles = inputBindingLifecycles;
this.outputBindingsLifecycles = outputBindingsLifecycles;
this.objectMapper = new ObjectMapper();
this.objectMapper = objectMapper;
}
/**

13
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java
View File

@@ -23,6 +23,8 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Autowired;
@@ -219,9 +221,9 @@ public class BindingServiceConfiguration {
@ConditionalOnMissingBean(search = SearchStrategy.CURRENT)
public BindingService bindingService(
BindingServiceProperties bindingServiceProperties,
BinderFactory binderFactory, TaskScheduler taskScheduler) {
return new BindingService(bindingServiceProperties, binderFactory, taskScheduler);
BinderFactory binderFactory, TaskScheduler taskScheduler, @Nullable ObjectMapper objectMapper) {
objectMapper = objectMapper == null ? new ObjectMapper() : objectMapper;
return new BindingService(bindingServiceProperties, binderFactory, taskScheduler, objectMapper);
}
@Bean
@@ -241,8 +243,9 @@ public class BindingServiceConfiguration {
@Bean
public BindingsLifecycleController bindingsLifecycleController(List<InputBindingLifecycle> inputBindingLifecycles,
List<OutputBindingLifecycle> outputBindingsLifecycles) {
return new BindingsLifecycleController(inputBindingLifecycles, outputBindingsLifecycles);
List<OutputBindingLifecycle> outputBindingsLifecycles, @Nullable ObjectMapper objectMapper) {
objectMapper = objectMapper == null ? new ObjectMapper() : objectMapper;
return new BindingsLifecycleController(inputBindingLifecycles, outputBindingsLifecycles, objectMapper);
}
@Bean

3
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java
View File

@@ -23,6 +23,7 @@ import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
@@ -177,7 +178,7 @@ public class BinderAwareChannelResolverTests {
when(mockBinderFactory.getBinder("someTransport",
DirectWithAttributesChannel.class)).thenReturn(binder2);
BindingService bindingService = new BindingService(this.bindingServiceProperties,
mockBinderFactory);
mockBinderFactory, new ObjectMapper());
BinderAwareChannelResolver resolver = new BinderAwareChannelResolver(
bindingService, this.bindingTargetFactory,
new DynamicDestinationsBindable());

27
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java
View File

@@ -30,6 +30,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.Ignore;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
@@ -121,7 +122,7 @@ public class BindingServiceTests {
properties.setBindings(bindingProperties);
DefaultBinderFactory binderFactory = createMockBinderFactory();
Binder binder = binderFactory.getBinder("mock", MessageChannel.class);
BindingService service = new BindingService(properties, binderFactory);
BindingService service = new BindingService(properties, binderFactory, new ObjectMapper());
MessageChannel inputChannel = new DirectChannel();
Binding<MessageChannel> mockBinding = Mockito.mock(Binding.class);
when(binder.bindConsumer(eq("foo"), isNull(), same(inputChannel),
@@ -153,7 +154,7 @@ public class BindingServiceTests {
DefaultBinderFactory binderFactory = createMockBinderFactory();
Binder binder = binderFactory.getBinder("mock", MessageChannel.class);
BindingService service = new BindingService(properties, binderFactory);
BindingService service = new BindingService(properties, binderFactory, new ObjectMapper());
MessageChannel inputChannel = new DirectChannel();
Binding<MessageChannel> mockBinding1 = Mockito.mock(Binding.class);
@@ -209,7 +210,7 @@ public class BindingServiceTests {
DefaultBinderFactory binderFactory = createMockBinderFactory();
Binder binder = binderFactory.getBinder("mock", MessageChannel.class);
BindingService service = new BindingService(properties, binderFactory);
BindingService service = new BindingService(properties, binderFactory, new ObjectMapper());
MessageChannel inputChannel = new DirectChannel();
Binding<MessageChannel> mockBinding1 = Mockito.mock(Binding.class, "FirstBinding");
@@ -268,7 +269,7 @@ public class BindingServiceTests {
DefaultBinderFactory binderFactory = createMockBinderFactory();
Binder binder = binderFactory.getBinder("mock", MessageChannel.class);
BindingService service = new BindingService(properties, binderFactory);
BindingService service = new BindingService(properties, binderFactory, new ObjectMapper());
MessageChannel inputChannel = new DirectChannel();
Binding<MessageChannel> mockBinding1 = Mockito.mock(Binding.class);
@@ -307,7 +308,7 @@ public class BindingServiceTests {
properties.setBindings(bindingProperties);
DefaultBinderFactory binderFactory = createMockBinderFactory();
Binder binder = binderFactory.getBinder("mock", MessageChannel.class);
BindingService service = new BindingService(properties, binderFactory);
BindingService service = new BindingService(properties, binderFactory, new ObjectMapper());
MessageChannel inputChannel = new DirectChannel();
Binding<MessageChannel> mockBinding = Mockito.mock(Binding.class);
when(binder.bindConsumer(eq("foo"), eq("fooGroup"), same(inputChannel),
@@ -340,7 +341,7 @@ public class BindingServiceTests {
final AtomicReference<MessageChannel> dynamic = new AtomicReference<>();
when(binder.bindProducer(matches("foo"), any(DirectChannel.class),
any(ProducerProperties.class))).thenReturn(mockBinding);
BindingService bindingService = new BindingService(properties, binderFactory) {
BindingService bindingService = new BindingService(properties, binderFactory, new ObjectMapper()) {
@Override
protected <T> Binder<T, ?, ?> getBinder(String channelName,
@@ -426,7 +427,7 @@ public class BindingServiceTests {
bindingProperties.put(outputChannelName, props);
serviceProperties.setBindings(bindingProperties);
DefaultBinderFactory binderFactory = createMockBinderFactory();
BindingService service = new BindingService(serviceProperties, binderFactory);
BindingService service = new BindingService(serviceProperties, binderFactory, new ObjectMapper());
MessageChannel outputChannel = new DirectChannel();
try {
service.bindProducer(outputChannel, outputChannelName);
@@ -498,7 +499,7 @@ public class BindingServiceTests {
bindingProperties.put(inputChannelName, props);
serviceProperties.setBindings(bindingProperties);
DefaultBinderFactory binderFactory = createMockBinderFactory();
BindingService service = new BindingService(serviceProperties, binderFactory);
BindingService service = new BindingService(serviceProperties, binderFactory, new ObjectMapper());
MessageChannel inputChannel = new DirectChannel();
try {
service.bindConsumer(inputChannel, inputChannelName);
@@ -520,7 +521,7 @@ public class BindingServiceTests {
BindingServiceProperties bindingServiceProperties = createBindingServiceProperties(
properties);
BindingService bindingService = new BindingService(bindingServiceProperties,
createMockBinderFactory());
createMockBinderFactory(), new ObjectMapper());
bindingService.bindConsumer(new DirectChannel(), "input");
try {
bindingService.bindProducer(new DirectChannel(), "output");
@@ -545,7 +546,7 @@ public class BindingServiceTests {
BinderFactory binderFactory = new BindingServiceConfiguration()
.binderFactory(createMockBinderTypeRegistry(), bindingServiceProperties, Mockito.mock(ObjectProvider.class));
BindingService bindingService = new BindingService(bindingServiceProperties,
binderFactory);
binderFactory, new ObjectMapper());
bindingService.bindConsumer(new DirectChannel(), "input");
bindingService.bindProducer(new DirectChannel(), "output");
}
@@ -565,7 +566,7 @@ public class BindingServiceTests {
BinderFactory binderFactory = new BindingServiceConfiguration()
.binderFactory(createMockBinderTypeRegistry(), bindingServiceProperties, Mockito.mock(ObjectProvider.class));
BindingService bindingService = new BindingService(bindingServiceProperties,
binderFactory);
binderFactory, new ObjectMapper());
bindingService.bindConsumer(new DirectChannel(), "input");
try {
bindingService.bindProducer(new DirectChannel(), "output");
@@ -599,7 +600,7 @@ public class BindingServiceTests {
Binder binder = binderFactory.getBinder("mock", MessageChannel.class);
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.initialize();
BindingService service = new BindingService(properties, binderFactory, scheduler);
BindingService service = new BindingService(properties, binderFactory, scheduler, new ObjectMapper());
MessageChannel inputChannel = new DirectChannel();
final Binding<MessageChannel> mockBinding = Mockito.mock(Binding.class);
final CountDownLatch fail = new CountDownLatch(2);
@@ -644,7 +645,7 @@ public class BindingServiceTests {
Binder binder = binderFactory.getBinder("mock", MessageChannel.class);
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.initialize();
BindingService service = new BindingService(properties, binderFactory, scheduler);
BindingService service = new BindingService(properties, binderFactory, scheduler, new ObjectMapper());
MessageChannel outputChannel = new DirectChannel();
final Binding<MessageChannel> mockBinding = Mockito.mock(Binding.class);
final CountDownLatch fail = new CountDownLatch(2);