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

Use 'instanceof' pattern variable

Browse Source 
Update authors and copyright years

Fixing checkstyle issue
This commit is contained in:
Byungjun You
2023-01-22 19:44:16 +09:00
committed by Soby Chacko
parent b49fde97c8
commit 7895025c47
33 changed files with 283 additions and 274 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2022 the original author or authors.
* Copyright 2014-2023 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.
@@ -86,6 +86,7 @@ import org.springframework.util.StringUtils;
* @author Aldo Sinanaj
* @author Yi Liu
* @author Omer Celik
* @author Byungjun You
*/
public class KafkaTopicProvisioner implements
// @checkstyle:off
@@ -330,8 +331,8 @@ public class KafkaTopicProvisioner implements
dlqProducerProperties.getTopic());
}
catch (Throwable throwable) {
if (throwable instanceof Error) {
throw (Error) throwable;
if (throwable instanceof Error throwableError) {
throw throwableError;
}
else {
throw new ProvisioningException("Provisioning exception encountered for " + name, throwable);
@@ -352,8 +353,8 @@ public class KafkaTopicProvisioner implements
// TODO:
// https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/pull/514#discussion_r241075940
catch (Throwable throwable) {
if (throwable instanceof Error) {
throw (Error) throwable;
if (throwable instanceof Error throwableError) {
throw throwableError;
}
else {
// TODO:
@@ -570,8 +571,8 @@ public class KafkaTopicProvisioner implements
describeTopicsResult.all().get();
}
catch (ExecutionException ex) {
if (ex.getCause() instanceof UnknownTopicOrPartitionException) {
throw (UnknownTopicOrPartitionException) ex.getCause();
if (ex.getCause() instanceof UnknownTopicOrPartitionException unknownTopicOrPartitionException) {
throw unknownTopicOrPartitionException;
}
else {
logger.warn("No partitions have been retrieved for the topic "

9
binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2021-2022 the original author or authors.
* Copyright 2021-2023 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.
@@ -28,7 +28,6 @@ import org.apache.commons.logging.LogFactory;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.kafka.receiver.KafkaReceiver;
import reactor.kafka.receiver.ReceiverOptions;
@@ -67,6 +66,7 @@ import org.springframework.util.StringUtils;
/**
* @author Gary Russell
* @author Byungjun You
* @since 4.0
*
*/
@@ -223,10 +223,7 @@ public class ReactorKafkaBinder
protected void handleMessageInternal(Message<?> message) {
Object sendResultHeader = message.getHeaders().get("sendResult");
Sinks.One<RecordMetadata> sink = Sinks.one();
if (sendResultHeader instanceof AtomicReference) {
@SuppressWarnings("unchecked")
AtomicReference<Mono<RecordMetadata>> result =
(AtomicReference<Mono<RecordMetadata>>) sendResultHeader;
if (sendResultHeader instanceof AtomicReference result) {
result.set(sink.asMono());
}
if (this.sender != null) {

12
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2022 the original author or authors.
* Copyright 2017-2023 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.
@@ -81,6 +81,7 @@ import org.springframework.util.StringUtils;
* @author Marius Bogoevici
* @author Soby Chacko
* @author Gary Russell
* @author Byungjun You
*/
@Configuration(proxyBeanMethods = false)
@EnableConfigurationProperties({ KafkaProperties.class, KafkaStreamsExtendedBindingProperties.class })
@@ -155,8 +156,8 @@ public class KafkaStreamsBinderSupportAutoConfiguration {
@SuppressWarnings("unchecked")
private void flatten(String propertyName, Object value,
Map<String, Object> flattenedProperties) {
if (value instanceof Map) {
((Map<Object, Object>) value).forEach((k, v) -> flatten(
if (value instanceof Map valueAsMap) {
valueAsMap.forEach((k, v) -> flatten(
(propertyName != null ? propertyName + "." : "") + k, v,
flattenedProperties));
}
@@ -213,10 +214,7 @@ public class KafkaStreamsBinderSupportAutoConfiguration {
else {
Object bootstrapServerConfig = properties
.get(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG);
if (bootstrapServerConfig instanceof String) {
@SuppressWarnings("unchecked")
String bootStrapServers = (String) properties
.get(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG);
if (bootstrapServerConfig instanceof String bootStrapServers) {
if (bootStrapServers.equals("localhost:9092")) {
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
kafkaConnectionString);

7
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2022 the original author or authors.
* Copyright 2019-2023 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.
@@ -65,6 +65,7 @@ import org.springframework.util.CollectionUtils;
/**
* @author Soby Chacko
* @author Byungjun You
* @since 2.2.0
*/
public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderProcessor implements BeanFactoryAware {
@@ -397,8 +398,8 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro
private Object handleCurriedFunctions(Object[] adaptedInboundArguments, Object result) {
int i = 1;
while (result instanceof Function || result instanceof Consumer) {
if (result instanceof Function) {
result = ((Function<Object, Object>) result).apply(adaptedInboundArguments[i]);
if (result instanceof Function function) {
result = function.apply(adaptedInboundArguments[i]);
}
else {
((Consumer<Object>) result).accept(adaptedInboundArguments[i]);

11
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2019 the original author or authors.
* Copyright 2018-2023 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.
@@ -50,6 +50,7 @@ import org.springframework.util.StringUtils;
* contentType. Based on the contentType, a {@link MessageConverter} will be resolved.
*
* @author Soby Chacko
* @author Byungjun You
*/
public class KafkaStreamsMessageConversionDelegate {
@@ -94,7 +95,7 @@ public class KafkaStreamsMessageConversionDelegate {
final KStream<?, ?> kStreamWithEnrichedHeaders = outboundBindTarget
.filter((k, v) -> v != null)
.mapValues((v) -> {
Message<?> message = v instanceof Message<?> ? (Message<?>) v
Message<?> message = v instanceof Message<?> m ? m
: MessageBuilder.withPayload(v).build();
Map<String, Object> headers = new HashMap<>(message.getHeaders());
if (StringUtils.hasText(contentType)) {
@@ -171,14 +172,14 @@ public class KafkaStreamsMessageConversionDelegate {
if (o2 instanceof Message || o2 instanceof String
|| o2 instanceof byte[]) {
Message<?> m1 = null;
if (o2 instanceof Message) {
if (o2 instanceof Message message) {
m1 = perRecordContentTypeHolder.contentType != null
? MessageBuilder.fromMessage((Message<?>) o2)
? MessageBuilder.fromMessage(message)
.setHeader(
MessageHeaders.CONTENT_TYPE,
perRecordContentTypeHolder.contentType)
.build()
: (Message<?>) o2;
: message;
}
else {
m1 = perRecordContentTypeHolder.contentType != null

11
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/BinderHeaderMapper.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2019 the original author or authors.
* Copyright 2017-2023 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.
@@ -62,6 +62,7 @@ import org.springframework.util.MimeType;
* @author Gary Russell
* @author Artem Bilan
* @author Soby Chacko
* @author Byungjun You
*
* @since 3.0.0
*
@@ -256,8 +257,8 @@ public class BinderHeaderMapper extends AbstractKafkaHeaderMapper {
headers.forEach((key, rawValue) -> {
if (matches(key, rawValue)) {
Object valueToAdd = headerValueToAddOut(key, rawValue);
if (valueToAdd instanceof byte[]) {
target.add(new RecordHeader(key, (byte[]) valueToAdd));
if (valueToAdd instanceof byte[] valueToAddBytes) {
target.add(new RecordHeader(key, valueToAddBytes));
}
else {
try {
@@ -268,8 +269,8 @@ public class BinderHeaderMapper extends AbstractKafkaHeaderMapper {
}
if (!this.encodeStrings
&& !MimeType.class.isAssignableFrom(rawValue.getClass())
&& valueToAdd instanceof String) {
target.add(new RecordHeader(key, ((String) valueToAdd).getBytes(getCharset())));
&& valueToAdd instanceof String stringValueToAdd) {
target.add(new RecordHeader(key, stringValueToAdd.getBytes(getCharset())));
className = JAVA_LANG_STRING;
}
else {

53
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2022 the original author or authors.
* Copyright 2014-2023 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.
@@ -155,6 +155,7 @@ import org.springframework.util.backoff.FixedBackOff;
* @author Taras Danylchuk
* @author Yi Liu
* @author Chris Bono
* @author Byungjun You
*/
public class KafkaMessageChannelBinder extends
// @checkstyle:off
@@ -430,13 +431,11 @@ public class KafkaMessageChannelBinder extends
List<ChannelInterceptor> interceptors = ((InterceptableChannel) channel)
.getInterceptors();
interceptors.forEach((interceptor) -> {
if (interceptor instanceof PartitioningInterceptor) {
((PartitioningInterceptor) interceptor)
.setPartitionCount(partitions.size());
if (interceptor instanceof PartitioningInterceptor partitioningInterceptor) {
partitioningInterceptor.setPartitionCount(partitions.size());
}
else if (interceptor instanceof DefaultPartitioningInterceptor) {
((DefaultPartitioningInterceptor) interceptor)
.setPartitionCount(partitions.size());
else if (interceptor instanceof DefaultPartitioningInterceptor defaultPartitioningInterceptor) {
defaultPartitioningInterceptor.setPartitionCount(partitions.size());
}
});
}
@@ -721,9 +720,8 @@ public class KafkaMessageChannelBinder extends
if (!extendedConsumerProperties.isBatchMode()
&& extendedConsumerProperties.getMaxAttempts() > 1
&& transMan == null) {
if (!(customizer instanceof ListenerContainerWithDlqAndRetryCustomizer)
|| ((ListenerContainerWithDlqAndRetryCustomizer) customizer)
.retryAndDlqInBinding(destination.getName(), group)) {
if (!(customizer instanceof ListenerContainerWithDlqAndRetryCustomizer c)
|| c.retryAndDlqInBinding(destination.getName(), group)) {
kafkaMessageDrivenChannelAdapter
.setRetryTemplate(buildRetryTemplate(extendedConsumerProperties));
kafkaMessageDrivenChannelAdapter
@@ -770,15 +768,14 @@ public class KafkaMessageChannelBinder extends
CommonErrorHandler.class);
messageListenerContainer.setCommonErrorHandler(commonErrorHandler);
}
if (customizer instanceof ListenerContainerWithDlqAndRetryCustomizer) {
if (customizer instanceof ListenerContainerWithDlqAndRetryCustomizer c) {
BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> destinationResolver = createDestResolver(
extendedConsumerProperties.getExtension());
BackOff createBackOff = extendedConsumerProperties.getMaxAttempts() > 1
? createBackOff(extendedConsumerProperties)
: null;
((ListenerContainerWithDlqAndRetryCustomizer) customizer)
.configure(messageListenerContainer, destination.getName(), consumerGroup, destinationResolver,
c.configure(messageListenerContainer, destination.getName(), consumerGroup, destinationResolver,
createBackOff);
}
else {
@@ -1067,8 +1064,8 @@ public class KafkaMessageChannelBinder extends
MessageConverter messageConverter = BindingUtils.getConsumerMessageConverter(getApplicationContext(),
extendedConsumerProperties, this.configurationProperties);
if (messageConverter instanceof MessagingMessageConverter) {
((MessagingMessageConverter) messageConverter).setHeaderMapper(getHeaderMapper(extendedConsumerProperties));
if (messageConverter instanceof MessagingMessageConverter messagingMessageConverter) {
messagingMessageConverter.setHeaderMapper(getHeaderMapper(extendedConsumerProperties));
}
return messageConverter;
}
@@ -1139,11 +1136,11 @@ public class KafkaMessageChannelBinder extends
Object timeout = producerFactory.getConfigurationProperties().get(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG);
Long sendTimeout = null;
if (timeout instanceof Number) {
sendTimeout = ((Number) timeout).longValue() + 2000L;
if (timeout instanceof Number timeoutAsNumber) {
sendTimeout = timeoutAsNumber.longValue() + 2000L;
}
else if (timeout instanceof String) {
sendTimeout = Long.parseLong((String) timeout) + 2000L;
else if (timeout instanceof String timeoutAsString) {
sendTimeout = Long.parseLong(timeoutAsString) + 2000L;
}
if (timeout == null) {
sendTimeout = ((Integer) ProducerConfig.configDef()
@@ -1197,9 +1194,9 @@ public class KafkaMessageChannelBinder extends
AtomicReference<ConsumerRecord<?, ?>> recordToSend = new AtomicReference<>(
record);
Throwable throwable = null;
if (message.getPayload() instanceof Throwable) {
if (message.getPayload() instanceof Throwable throwablePayload) {
throwable = (Throwable) message.getPayload();
throwable = throwablePayload;
HeaderMode headerMode = properties.getHeaderMode();
@@ -1338,11 +1335,9 @@ public class KafkaMessageChannelBinder extends
}
}
else {
if (message.getPayload() instanceof MessagingException) {
if (message.getPayload() instanceof MessagingException messagingException) {
AcknowledgmentCallback ack = StaticMessageHeaderAccessor
.getAcknowledgmentCallback(
((MessagingException) message.getPayload())
.getFailedMessage());
.getAcknowledgmentCallback(messagingException.getFailedMessage());
if (ack != null) {
if (isAutoCommitOnError(properties)) {
ack.acknowledge(AcknowledgmentCallback.Status.REJECT);
@@ -1514,9 +1509,9 @@ public class KafkaMessageChannelBinder extends
@Override
public void stop() {
if (this.producerFactory instanceof DisposableBean) {
if (this.producerFactory instanceof DisposableBean disposableProducerFactory) {
try {
((DisposableBean) producerFactory).destroy();
disposableProducerFactory.destroy();
}
catch (Exception ex) {
this.logger.error(ex, "Error destroying the producer factory bean: ");
@@ -1636,8 +1631,8 @@ public class KafkaMessageChannelBinder extends
}
private String keyOrValue(Object keyOrValue) {
if (keyOrValue instanceof byte[]) {
return "byte[" + ((byte[]) keyOrValue).length + "]";
if (keyOrValue instanceof byte[] keyOrValueBytes) {
return "byte[" + keyOrValueBytes.length + "]";
}
else {
return toDisplayString(ObjectUtils.nullSafeToString(keyOrValue), 50);

15
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2022 the original author or authors.
* Copyright 2015-2023 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.
@@ -82,6 +82,7 @@ import org.springframework.messaging.converter.MessageConverter;
* @author Aldo Sinanaj
* @author Chris Bono
* @author Yi Liu
* @author Byungjun You
*/
@Configuration(proxyBeanMethods = false)
@ConditionalOnMissingBean(Binder.class)
@@ -200,16 +201,14 @@ public class KafkaBinderConfiguration {
@Override
public void configure(ProducerFactory<?, ?> pf) {
if (pf instanceof DefaultKafkaProducerFactory) {
((DefaultKafkaProducerFactory<?, ?>) pf)
.addListener(new MicrometerProducerListener<>(meterRegistry));
pf.addListener(new MicrometerProducerListener<>(meterRegistry));
}
}
@Override
public void configure(ConsumerFactory<?, ?> cf) {
if (cf instanceof DefaultKafkaConsumerFactory) {
((DefaultKafkaConsumerFactory<?, ?>) cf)
.addListener(new MicrometerConsumerListener<>(meterRegistry));
cf.addListener(new MicrometerConsumerListener<>(meterRegistry));
}
}
@@ -255,16 +254,14 @@ public class KafkaBinderConfiguration {
@Override
public void configure(ProducerFactory<?, ?> pf) {
if (pf instanceof DefaultKafkaProducerFactory) {
((DefaultKafkaProducerFactory<?, ?>) pf)
.addListener(new MicrometerProducerListener<>(this.meterRegistry));
pf.addListener(new MicrometerProducerListener<>(this.meterRegistry));
}
}
@Override
public void configure(ConsumerFactory<?, ?> cf) {
if (cf instanceof DefaultKafkaConsumerFactory) {
((DefaultKafkaConsumerFactory<?, ?>) cf)
.addListener(new MicrometerConsumerListener<>(this.meterRegistry));
cf.addListener(new MicrometerConsumerListener<>(this.meterRegistry));
}
}

29
binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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.
@@ -78,6 +78,7 @@ import org.springframework.util.StringUtils;
* @author Gary Russell
* @author Oleg Zhurakousky
* @author Michael Michailidis
* @author Byungjun You
*/
// @checkstyle:off
public class RabbitExchangeQueueProvisioner
@@ -380,14 +381,14 @@ public class RabbitExchangeQueueProvisioner
bindingKey += "-" + index;
Map<String, Object> arguments = new HashMap<>();
arguments.putAll(extendedProperties.getQueueBindingArguments());
if (exchange instanceof TopicExchange) {
Binding binding = BindingBuilder.bind(queue).to((TopicExchange) exchange)
if (exchange instanceof TopicExchange topicExchange) {
Binding binding = BindingBuilder.bind(queue).to(topicExchange)
.with(bindingKey);
declareBinding(queue.getName(), binding);
return binding;
}
else if (exchange instanceof DirectExchange) {
Binding binding = BindingBuilder.bind(queue).to((DirectExchange) exchange)
else if (exchange instanceof DirectExchange directExchange) {
Binding binding = BindingBuilder.bind(queue).to(directExchange)
.with(bindingKey);
declareBinding(queue.getName(), binding);
return binding;
@@ -497,8 +498,8 @@ public class RabbitExchangeQueueProvisioner
: properties.getDeadLetterRoutingKey(),
arguments);
declareBinding(dlqName, dlqBinding);
if (properties instanceof RabbitConsumerProperties
&& ((RabbitConsumerProperties) properties).isRepublishToDlq()) {
if (properties instanceof RabbitConsumerProperties rabbitConsumerProperties
&& rabbitConsumerProperties.isRepublishToDlq()) {
/*
* Also bind with the base queue name when republishToDlq is used, which
* does not know about partitioning
@@ -760,9 +761,9 @@ public class RabbitExchangeQueueProvisioner
true)).forEach(name -> {
String group = null;
String bindingName = null;
if (destination instanceof RabbitConsumerDestination) {
group = ((RabbitConsumerDestination) destination).getGroup();
bindingName = ((RabbitConsumerDestination) destination).getBindingName();
if (destination instanceof RabbitConsumerDestination rabbitConsumerDestination) {
group = rabbitConsumerDestination.getGroup();
bindingName = rabbitConsumerDestination.getBindingName();
}
RabbitConsumerProperties properties = consumerProperties.getExtension();
String toRemove = properties.isQueueNameGroupOnly() ? bindingName + "." + group : name.trim();
@@ -780,8 +781,8 @@ public class RabbitExchangeQueueProvisioner
ExtendedProducerProperties<RabbitProducerProperties> properties) {
synchronized (this.autoDeclareContext) {
if (dest instanceof RabbitProducerDestination) {
String qual = ((RabbitProducerDestination) dest).getBeanNameQualifier();
if (dest instanceof RabbitProducerDestination rabbitProducerDestination) {
String qual = rabbitProducerDestination.getBeanNameQualifier();
removeSingleton(dest.getName() + "." + qual + ".exchange");
String[] requiredGroups = properties.getRequiredGroups();
if (!ObjectUtils.isEmpty(requiredGroups)) {
@@ -854,8 +855,8 @@ public class RabbitExchangeQueueProvisioner
if (this.autoDeclareContext.containsBean(name)) {
ConfigurableListableBeanFactory beanFactory = this.autoDeclareContext
.getBeanFactory();
if (beanFactory instanceof DefaultListableBeanFactory) {
((DefaultListableBeanFactory) beanFactory).destroySingleton(name);
if (beanFactory instanceof DefaultListableBeanFactory defaultListableBeanFactory) {
defaultListableBeanFactory.destroySingleton(name);
}
}
}

17
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2022 the original author or authors.
* Copyright 2013-2023 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.
@@ -134,6 +134,7 @@ import org.springframework.util.StringUtils;
* @author Soby Chacko
* @author Oleg Zhurakousky
* @author Christian Tzolov
* @author Byungjun You
*/
// @checkstyle:off
public class RabbitMessageChannelBinder extends
@@ -271,9 +272,9 @@ public class RabbitMessageChannelBinder extends
@Override
public void destroy() throws Exception {
if (this.connectionFactory instanceof DisposableBean) {
if (this.connectionFactory instanceof DisposableBean disposableConnectionFactory) {
if (this.destroyConnectionFactory) {
((DisposableBean) this.connectionFactory).destroy();
disposableConnectionFactory.destroy();
}
}
}
@@ -968,11 +969,9 @@ public class RabbitMessageChannelBinder extends
}
}
else {
if (message.getPayload() instanceof MessagingException) {
if (message.getPayload() instanceof MessagingException messagingException) {
AcknowledgmentCallback ack = StaticMessageHeaderAccessor
.getAcknowledgmentCallback(
((MessagingException) message.getPayload())
.getFailedMessage());
.getAcknowledgmentCallback(messagingException.getFailedMessage());
if (ack != null) {
if (properties.getExtension().isRequeueRejected()) {
ack.acknowledge(Status.REQUEUE);
@@ -1088,8 +1087,8 @@ public class RabbitMessageChannelBinder extends
@Override
protected Message createMessage(Object object,
MessageProperties messageProperties) {
if (object instanceof byte[]) {
return new Message((byte[]) object, messageProperties);
if (object instanceof byte[] bytes) {
return new Message(bytes, messageProperties);
}
else {
// just for safety (backwards compatibility)

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

@@ -1,5 +1,5 @@
/*
* Copyright 2022-2022 the original author or authors.
* Copyright 2022-2023 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.
@@ -29,6 +29,10 @@ import org.springframework.core.annotation.Order;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.messaging.MessageHandler;
/**
* @author Oleg Zhurakousky
* @author Byungjun You
*/
@Configuration(proxyBeanMethods = false)
@ConditionalOnBean(org.springframework.boot.actuate.autoconfigure.observation.ObservationAutoConfiguration.class)
public class ObservationAutoConfiguration {
@@ -50,9 +54,9 @@ public class ObservationAutoConfiguration {
ProducerMessageHandlerCustomizer<MessageHandler> observedProducerMessageHandlerCustomizer(
ApplicationContext applicationContext) {
return (handler, destinationName) -> {
if (handler instanceof AmqpOutboundEndpoint) {
((AmqpOutboundEndpoint) handler).getRabbitTemplate().setObservationEnabled(true);
((AmqpOutboundEndpoint) handler).getRabbitTemplate().setApplicationContext(applicationContext);
if (handler instanceof AmqpOutboundEndpoint amqpOutboundEndpoint) {
amqpOutboundEndpoint.getRabbitTemplate().setObservationEnabled(true);
amqpOutboundEndpoint.getRabbitTemplate().setApplicationContext(applicationContext);
}
};
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2022 the original author or authors.
* Copyright 2015-2023 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.
@@ -58,6 +58,7 @@ import org.springframework.util.CollectionUtils;
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Ben Blinebury
* @author Byungjun You
*/
@Configuration(proxyBeanMethods = false)
@Import({ PropertyPlaceholderAutoConfiguration.class })
@@ -87,9 +88,10 @@ public class RabbitMessageChannelBinderConfiguration {
@Nullable ConnectionNameStrategy connectionNameStrategy) {
String connectionNamePrefix = this.rabbitBinderConfigurationProperties.getConnectionNamePrefix();
if (this.rabbitConnectionFactory instanceof AbstractConnectionFactory && connectionNamePrefix != null && connectionNameStrategy == null) {
if (this.rabbitConnectionFactory instanceof AbstractConnectionFactory connectionFactory
&& connectionNamePrefix != null && connectionNameStrategy == null) {
final AtomicInteger nameIncrementer = new AtomicInteger();
((AbstractConnectionFactory) this.rabbitConnectionFactory).setConnectionNameStrategy(f -> connectionNamePrefix
connectionFactory.setConnectionNameStrategy(f -> connectionNamePrefix
+ "#" + nameIncrementer.getAndIncrement());
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2022 the original author or authors.
* Copyright 2015-2023 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.
@@ -94,6 +94,7 @@ import static org.mockito.Mockito.verify;
* @author Artem Bilan
* @author Soby Chacko
* @author Chris Bono
* @author Byungjun You
*/
public class RabbitBinderModuleTests {
@@ -436,8 +437,8 @@ public class RabbitBinderModuleTests {
@Bean
public ProducerMessageHandlerCustomizer<MessageHandler> messageHandlerCustomizer() {
return (handler, destinationName) -> {
if (handler instanceof AbstractMessageHandler) {
((AbstractMessageHandler) handler).setBeanName("setByCustomizer:" + destinationName);
if (handler instanceof AbstractMessageHandler messageHandler) {
messageHandler.setBeanName("setByCustomizer:" + destinationName);
}
};
}

11
core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2022 the original author or authors.
* Copyright 2017-2023 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.
@@ -57,6 +57,7 @@ import static org.assertj.core.api.Assertions.assertThat;
*
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Byungjun You
*
*/
public class ContentTypeTckTests {
@@ -586,8 +587,8 @@ public class ContentTypeTckTests {
@Bean
public Function<Object, Person> echo(JsonMapper mapper) {
return value -> value instanceof byte[]
? mapper.fromJson((byte[]) value, Person.class)
return value -> value instanceof byte[] valueAsBytes
? mapper.fromJson(valueAsBytes, Person.class)
: mapper.fromJson((String) value, Person.class);
}
}
@@ -598,8 +599,8 @@ public class ContentTypeTckTests {
@Bean
public Function<Message<?>, Person> echo(JsonMapper mapper) {
return message -> message.getPayload() instanceof byte[]
? mapper.fromJson((byte[]) message.getPayload(), Person.class)
return message -> message.getPayload() instanceof byte[] payloadAsBytes
? mapper.fromJson(payloadAsBytes, Person.class)
: mapper.fromJson((String) message.getPayload(), Person.class);
}
}

7
core/spring-cloud-stream-test-binder/src/main/java/org/springframework/cloud/stream/binder/test/FunctionBindingTestUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2020 the original author or authors.
* Copyright 2019-2023 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.
@@ -39,6 +39,7 @@ import org.springframework.messaging.MessageChannel;
* It is intended for internal framework testing and is NOT for public use. Breaking changes are likely!!!
*
* @author Oleg Zhurakousky
* @author Byungjun You
* @since 3.0.2
*
*/
@@ -52,8 +53,8 @@ public final class FunctionBindingTestUtils {
public static void bind(ConfigurableApplicationContext applicationContext, Object function) {
try {
Object targetFunction = function;
if (function instanceof FunctionRegistration) {
targetFunction = ((FunctionRegistration) function).getTarget();
if (function instanceof FunctionRegistration functionRegistration) {
targetFunction = functionRegistration.getTarget();
}
String functionName = targetFunction instanceof Function ? "function" : (targetFunction instanceof Consumer ? "consumer" : "supplier");

8
core/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/binder/AbstractTestBinder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2017 the original author or authors.
* Copyright 2014-2023 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.
@@ -34,6 +34,7 @@ import org.springframework.util.CollectionUtils;
* @author Gary Russell
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Byungjun You
*/
// @checkstyle:off
public abstract class AbstractTestBinder<C extends AbstractBinder<MessageChannel, CP, PP>, CP extends ConsumerProperties, PP extends ProducerProperties>
@@ -86,12 +87,11 @@ public abstract class AbstractTestBinder<C extends AbstractBinder<MessageChannel
* org.springframework.cloud.stream.binding.MessageConverterConfigurer
*/
private void checkChannelIsConfigured(MessageChannel messageChannel, CP properties) {
if (messageChannel instanceof AbstractSubscribableChannel
if (messageChannel instanceof AbstractSubscribableChannel subscribableMessageChannel
&& !properties.isUseNativeDecoding()) {
Assert.isTrue(
!CollectionUtils
.isEmpty(((AbstractSubscribableChannel) messageChannel)
.getInterceptors()),
.isEmpty(subscribableMessageChannel.getInterceptors()),
"'messageChannel' appears to be misconfigured. "
+ "Consider creating channel via AbstractBinderTest.createBindableChannel(..)");
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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.
@@ -97,7 +97,7 @@ import org.springframework.util.StringUtils;
* @author Oleg Zhurakousky
* @author Artem Bilan
* @author Gary Russell
*
* @author Byungjun You
* @since 1.1
*/
public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties, P extends ProducerProperties, PP extends ProvisioningProvider<C, P>>
@@ -268,16 +268,16 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
producerMessageHandler = createProducerMessageHandler(producerDestination,
producerProperties, outputChannel, errorChannel);
customizeProducerMessageHandler(producerMessageHandler, producerDestination.getName());
if (producerMessageHandler instanceof InitializingBean) {
((InitializingBean) producerMessageHandler).afterPropertiesSet();
if (producerMessageHandler instanceof InitializingBean initializingHandler) {
initializingHandler.afterPropertiesSet();
}
}
catch (Exception e) {
if (e instanceof BinderException) {
throw (BinderException) e;
if (e instanceof BinderException binderException) {
throw binderException;
}
else if (e instanceof ProvisioningException) {
throw (ProvisioningException) e;
else if (e instanceof ProvisioningException provisioningException) {
throw provisioningException;
}
else {
throw new BinderException(
@@ -286,15 +286,15 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
}
if (producerProperties.isAutoStartup()
&& producerMessageHandler instanceof Lifecycle) {
((Lifecycle) producerMessageHandler).start();
&& producerMessageHandler instanceof Lifecycle ProducerMessageHandlerWithLifeCycle) {
ProducerMessageHandlerWithLifeCycle.start();
}
this.postProcessOutputChannel(outputChannel, producerProperties);
AtomicReference<ReactiveStreamsConsumer> reactiveStreamsConsumerRef = new AtomicReference<>();
if (outputChannel instanceof SubscribableChannel) {
((SubscribableChannel) outputChannel)
if (outputChannel instanceof SubscribableChannel subscribableOutputChannel) {
subscribableOutputChannel
.subscribe(new SendingHandler(producerMessageHandler,
HeaderMode.embeddedHeaders
.equals(producerProperties.getHeaderMode()),
@@ -310,8 +310,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
}
Binding<MessageChannel> binding = new DefaultBinding<MessageChannel>(destination,
outputChannel, producerMessageHandler instanceof Lifecycle
? (Lifecycle) producerMessageHandler : null) {
outputChannel, producerMessageHandler instanceof Lifecycle producerMessageHandlerWithLifecycle
? producerMessageHandlerWithLifecycle : null) {
@Override
public Map<String, Object> getExtendedInfo() {
@@ -331,8 +331,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
if (rsc != null && rsc.isRunning()) {
rsc.destroy();
}
if (producerMessageHandler instanceof DisposableBean) {
((DisposableBean) producerMessageHandler).destroy();
if (producerMessageHandler instanceof DisposableBean disposableProducerMessageHandler) {
disposableProducerMessageHandler.destroy();
}
}
catch (Exception e) {
@@ -474,16 +474,16 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
consumerEndpoint = createConsumerEndpoint(destination, group, properties);
consumerEndpoint.setOutputChannel(inputChannel);
this.consumerCustomizer.configure(consumerEndpoint, name, group);
if (consumerEndpoint instanceof InitializingBean) {
((InitializingBean) consumerEndpoint).afterPropertiesSet();
if (consumerEndpoint instanceof InitializingBean initializingConsumerEndpoint) {
initializingConsumerEndpoint.afterPropertiesSet();
}
if (properties.isAutoStartup() && consumerEndpoint instanceof Lifecycle) {
((Lifecycle) consumerEndpoint).start();
if (properties.isAutoStartup() && consumerEndpoint instanceof Lifecycle consumerEndpointWithLifecycle) {
consumerEndpointWithLifecycle.start();
}
Binding<MessageChannel> binding = new DefaultBinding<MessageChannel>(name,
group, inputChannel, consumerEndpoint instanceof Lifecycle
? (Lifecycle) consumerEndpoint : null) {
group, inputChannel, consumerEndpoint instanceof Lifecycle consumerEndpointWithLifecycle
? consumerEndpointWithLifecycle : null) {
@Override
public Map<String, Object> getExtendedInfo() {
@@ -498,8 +498,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
@Override
protected void afterUnbind() {
try {
if (getEndpoint() instanceof DisposableBean) {
((DisposableBean) getEndpoint()).destroy();
if (getEndpoint() instanceof DisposableBean disposableEndpoint) {
disposableEndpoint.destroy();
}
}
catch (Exception e) {
@@ -515,14 +515,14 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
return binding;
}
catch (Exception e) {
if (consumerEndpoint instanceof Lifecycle) {
((Lifecycle) consumerEndpoint).stop();
if (consumerEndpoint instanceof Lifecycle consumerEndpointWithLifecycle) {
consumerEndpointWithLifecycle.stop();
}
if (e instanceof BinderException) {
throw (BinderException) e;
if (e instanceof BinderException binderException) {
throw binderException;
}
else if (e instanceof ProvisioningException) {
throw (ProvisioningException) e;
else if (e instanceof ProvisioningException provisioningException) {
throw provisioningException;
}
else {
throw new BinderException("Exception thrown while starting consumer: ", e);
@@ -545,8 +545,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
group, destination, properties);
MessageSource<?> messageSource = resources.getSource();
if (messageSource instanceof BeanFactoryAware) {
((BeanFactoryAware) messageSource).setBeanFactory(getApplicationContext().getBeanFactory());
if (messageSource instanceof BeanFactoryAware beanFactoryAwareMessageSource) {
beanFactoryAwareMessageSource.setBeanFactory(getApplicationContext().getBeanFactory());
}
bindingTarget.setSource(messageSource);
if (resources.getErrorInfrastructure() != null) {
@@ -565,12 +565,12 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
resources.getErrorInfrastructure(), properties));
}
postProcessPollableSource(bindingTarget);
if (properties.isAutoStartup() && resources.getSource() instanceof Lifecycle) {
((Lifecycle) resources.getSource()).start();
if (properties.isAutoStartup() && resources.getSource() instanceof Lifecycle sourceWithLifecycle) {
sourceWithLifecycle.start();
}
Binding<PollableSource<MessageHandler>> binding = new DefaultBinding<PollableSource<MessageHandler>>(
name, group, inboundBindTarget, resources.getSource() instanceof Lifecycle
? (Lifecycle) resources.getSource() : null) {
name, group, inboundBindTarget, resources.getSource() instanceof Lifecycle sourceWithLifecycle
? sourceWithLifecycle : null) {
@Override
public Map<String, Object> getExtendedInfo() {
@@ -818,9 +818,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
if (errorChannel instanceof PublishSubscribeChannel) {
return true;
}
return errorChannel instanceof AbstractSubscribableChannel
? ((AbstractSubscribableChannel) errorChannel).getSubscriberCount() == 0
: true;
return !(errorChannel instanceof AbstractSubscribableChannel subscribableErrorChannel)
|| subscribableErrorChannel.getSubscriberCount() == 0;
}
private void destroyErrorInfrastructure(ProducerDestination destination, String bindingName) {
@@ -1039,11 +1038,11 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
@Override
@SuppressWarnings("unchecked")
public Message<?> preSend(Message<?> message, MessageChannel channel) {
if (message.getPayload() instanceof byte[]
if (message.getPayload() instanceof byte[] messagePayload
&& !message.getHeaders()
.containsKey(BinderHeaders.NATIVE_HEADERS_PRESENT)
&& EmbeddedHeaderUtils
.mayHaveEmbeddedHeaders((byte[]) message.getPayload())) {
.mayHaveEmbeddedHeaders(messagePayload)) {
MessageValues messageValues;
try {
@@ -1140,22 +1139,22 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
@Override
public void start() {
if (this.delegate instanceof Lifecycle) {
((Lifecycle) this.delegate).start();
if (this.delegate instanceof Lifecycle delegateWithLifecycle) {
delegateWithLifecycle.start();
}
}
@Override
public void stop() {
if (this.delegate instanceof Lifecycle) {
((Lifecycle) this.delegate).stop();
if (this.delegate instanceof Lifecycle delegateWithLifecycle) {
delegateWithLifecycle.stop();
}
}
@Override
public boolean isRunning() {
return this.delegate instanceof Lifecycle
&& ((Lifecycle) this.delegate).isRunning();
return this.delegate instanceof Lifecycle delegateWithLifecycle
&& delegateWithLifecycle.isRunning();
}
}

7
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderErrorChannel.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2021 the original author or authors.
* Copyright 2017-2023 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.
@@ -27,6 +27,7 @@ import org.springframework.messaging.MessageHandler;
*
* @author Gary Russell
* @author Oleg Zhurakousky
* @author Byungjun You
* @since 1.3
*
*/
@@ -56,9 +57,9 @@ class BinderErrorChannel extends PublishSubscribeChannel
if (this.finalHandler != null) {
super.subscribe(this.finalHandler);
}
if (handler instanceof LastSubscriberMessageHandler
if (handler instanceof LastSubscriberMessageHandler lastSubscriberMessageHandler
&& this.finalHandler == null) {
this.finalHandler = (LastSubscriberMessageHandler) handler;
this.finalHandler = lastSubscriberMessageHandler;
}
return result;
}

17
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2022 the original author or authors.
* Copyright 2015-2023 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.
@@ -73,6 +73,7 @@ import org.springframework.util.StringUtils;
* @author Artem Bilan
* @author Anshul Mehra
* @author Chris Bono
* @author Byungjun You
*/
public class DefaultBinderFactory implements BinderFactory, DisposableBean, ApplicationContextAware {
@@ -312,8 +313,8 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
Map<String, MessageConverter> messageConverters = binderProducingContext.getBeansOfType(MessageConverter.class);
if (!CollectionUtils.isEmpty(messageConverters) && !ObjectUtils.isEmpty(context.getBeansOfType(FunctionCatalog.class))) {
FunctionCatalog functionCatalog = this.context.getBean(FunctionCatalog.class);
if (functionCatalog instanceof SimpleFunctionRegistry) {
((SimpleFunctionRegistry) functionCatalog).addMessageConverters(messageConverters.values());
if (functionCatalog instanceof SimpleFunctionRegistry simpleFunctionRegistry) {
simpleFunctionRegistry.addMessageConverters(messageConverters.values());
}
}
@@ -323,8 +324,8 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
* accessible within binder's context (see
* https://github.com/spring-cloud/spring-cloud-stream/issues/1384)
*/
if (this.context != null && binder instanceof ApplicationContextAware) {
((ApplicationContextAware) binder).setApplicationContext(this.context);
if (this.context != null && binder instanceof ApplicationContextAware applicationContextAwareBinder) {
applicationContextAwareBinder.setApplicationContext(this.context);
}
if (!CollectionUtils.isEmpty(this.listeners)) {
for (Listener binderFactoryListener : this.listeners) {
@@ -555,12 +556,12 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
* foo={bar=baz}).
* @param propertyName property name to flatten
* @param value value that contains the property name
* @param flattenedProperties map to which we'll add the falttened property
* @param flattenedProperties map to which we'll add the flattened property
*/
@SuppressWarnings("unchecked")
private void flatten(String propertyName, Object value, Map<String, Object> flattenedProperties) {
if (value instanceof Map) {
((Map<Object, Object>) value).forEach((k, v) -> flatten(
if (value instanceof Map valueAsMap) {
valueAsMap.forEach((k, v) -> flatten(
(propertyName != null ? propertyName + "." : "") + k, v, flattenedProperties));
}
else {

19
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2021 the original author or authors.
* Copyright 2013-2023 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.
@@ -40,6 +40,7 @@ import org.springframework.util.StringUtils;
* @author Oleg Zhurakousky
* @author Myeonghyeon Lee
* @author Soby Chacko
* @author Byungjun You
* @see org.springframework.cloud.stream.annotation.EnableBinding
*/
@@ -93,8 +94,8 @@ public class DefaultBinding<T> implements Binding<T> {
@Override
public String getBindingName() {
String resolvedName = (this.target instanceof IntegrationObjectSupport)
? ((IntegrationObjectSupport) this.target).getComponentName() : getName();
String resolvedName = (this.target instanceof IntegrationObjectSupport integrationObjectSupportTarget)
? integrationObjectSupportTarget.getComponentName() : getName();
return resolvedName == null ? getName() : resolvedName;
}
@@ -156,8 +157,8 @@ public class DefaultBinding<T> implements Binding<T> {
@Override
public synchronized void pause() {
if (this.lifecycle instanceof Pausable) {
((Pausable) this.lifecycle).pause();
if (this.lifecycle instanceof Pausable pausableLifecycle) {
pausableLifecycle.pause();
this.paused = true;
}
else {
@@ -169,8 +170,8 @@ public class DefaultBinding<T> implements Binding<T> {
@Override
public synchronized void resume() {
if (this.lifecycle instanceof Pausable) {
((Pausable) this.lifecycle).resume();
if (this.lifecycle instanceof Pausable pausableLifecycle) {
pausableLifecycle.resume();
this.paused = false;
}
else {
@@ -193,8 +194,8 @@ public class DefaultBinding<T> implements Binding<T> {
@Override
public String toString() {
return " Binding [name=" + this.name + ", target=" + this.target + ", lifecycle="
+ ((this.lifecycle instanceof NamedComponent)
? ((NamedComponent) this.lifecycle).getComponentName()
+ ((this.lifecycle instanceof NamedComponent namedComponentWithLifeCycle)
? namedComponentWithLifeCycle.getComponentName()
: ObjectUtils.nullSafeToString(this.lifecycle))
+ "]";
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2020 the original author or authors.
* Copyright 2018-2023 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.
@@ -62,6 +62,7 @@ import org.springframework.util.Assert;
* @author Gary Russell
* @author Oleg Zhurakousky
* @author David Turanski
* @author Byungjun You
* @since 2.0
*
*/
@@ -115,8 +116,7 @@ public class DefaultPollableMessageSource
@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
Object result = invocation.proceed();
if (result instanceof Message) {
Message<?> received = (Message<?>) result;
if (result instanceof Message received) {
for (ChannelInterceptor interceptor : this.interceptors) {
received = interceptor.preSend(received, dummyChannel);
if (received == null) {
@@ -181,16 +181,16 @@ public class DefaultPollableMessageSource
@Override
public synchronized void start() {
if (!this.running && this.source instanceof Lifecycle) {
((Lifecycle) this.source).start();
if (!this.running && this.source instanceof Lifecycle sourceWithLifecycle) {
sourceWithLifecycle.start();
}
this.running = true;
}
@Override
public synchronized void stop() {
if (this.running && this.source instanceof Lifecycle) {
((Lifecycle) this.source).stop();
if (this.running && this.source instanceof Lifecycle sourceWithLifeCycle) {
sourceWithLifeCycle.stop();
}
this.running = false;
}
@@ -244,8 +244,8 @@ public class DefaultPollableMessageSource
}
catch (Exception e) {
AckUtils.autoNack(ackCallback);
if (e instanceof MessageHandlingException && ((MessageHandlingException) e)
.getFailedMessage().equals(message)) {
if (e instanceof MessageHandlingException messageHandlingException &&
messageHandlingException.getFailedMessage().equals(message)) {
throw (MessageHandlingException) e;
}
throw new MessageHandlingException(message, e);
@@ -309,7 +309,7 @@ public class DefaultPollableMessageSource
Message<?> message = this.source.receive();
if (message != null && type != null && this.messageConverter != null) {
Class<?> targetType = type == null ? Object.class
: type.getType() instanceof Class ? (Class<?>) type.getType()
: type.getType() instanceof Class clazz ? clazz
: Object.class;
Object payload = this.messageConverter.fromMessage(message, targetType, type);
if (payload == null) {

7
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/FinalRethrowingErrorMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2019 the original author or authors.
* Copyright 2017-2023 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.
@@ -27,6 +27,7 @@ import org.springframework.messaging.MessagingException;
* handled from {@code getErrorMessageHandler()}.
*
* @author Gary Russell
* @author Byungjun You
* @since 1.3
*
*/
@@ -35,8 +36,8 @@ class FinalRethrowingErrorMessageHandler
@Override
public void handleMessage(Message<?> message) throws MessagingException {
if (message.getPayload() instanceof MessagingException) {
throw (MessagingException) message.getPayload();
if (message.getPayload() instanceof MessagingException messagingException) {
throw messagingException;
}
else {
throw new MessagingException((Message<?>) null,

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

@@ -62,6 +62,7 @@ import org.springframework.validation.DataBinder;
* @author Michael Michailidis
* @author Chris Bono
* @author Artem Bilan
* @author Byungjun You
*/
public class BindingService {
@@ -100,16 +101,16 @@ public class BindingService {
public <T> Collection<Binding<T>> bindConsumer(T input, String inputName) {
Collection<Binding<T>> bindings = new ArrayList<>();
Class<?> inputClass = input.getClass();
if (input instanceof Advised) {
inputClass = Stream.of(((Advised) input).getProxiedInterfaces()).filter(c -> !c.getName().contains("org.springframework")).findFirst()
if (input instanceof Advised advisedInput) {
inputClass = Stream.of(advisedInput.getProxiedInterfaces()).filter(c -> !c.getName().contains("org.springframework")).findFirst()
.orElse(inputClass);
}
Binder<T, ConsumerProperties, ?> binder = (Binder<T, ConsumerProperties, ?>) getBinder(
inputName, inputClass);
ConsumerProperties consumerProperties = this.bindingServiceProperties
.getConsumerProperties(inputName);
if (binder instanceof ExtendedPropertiesBinder) {
Object extension = ((ExtendedPropertiesBinder) binder)
if (binder instanceof ExtendedPropertiesBinder extendedPropertiesBinder) {
Object extension = extendedPropertiesBinder
.getExtendedConsumerProperties(inputName);
ExtendedConsumerProperties extendedConsumerProperties = new ExtendedConsumerProperties(
extension);
@@ -146,11 +147,11 @@ public class BindingService {
continue;
}
Object extensiion = consumerProperties instanceof ExtendedConsumerProperties
? ((ExtendedConsumerProperties) consumerProperties).getExtension()
Object extension = consumerProperties instanceof ExtendedConsumerProperties extendedProperties
? extendedProperties.getExtension()
: null;
ConsumerProperties consumerPropertiesTemp = new ExtendedConsumerProperties<>(extensiion);
ConsumerProperties consumerPropertiesTemp = new ExtendedConsumerProperties<>(extension);
BeanUtils.copyProperties(consumerProperties, consumerPropertiesTemp);
consumerPropertiesTemp.setInstanceIndex(index);
@@ -270,8 +271,8 @@ public class BindingService {
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) {
outputClass = Stream.of(((Advised) output).getProxiedInterfaces()).filter(c -> !c.getName().contains("org.springframework")).findFirst()
if (output instanceof Advised advisedOutput) {
outputClass = Stream.of(advisedOutput.getProxiedInterfaces()).filter(c -> !c.getName().contains("org.springframework")).findFirst()
.orElse(outputClass);
}
if (binder == null) {
@@ -280,9 +281,8 @@ public class BindingService {
ProducerProperties producerProperties = this.bindingServiceProperties
.getProducerProperties(outputName);
if (binder instanceof ExtendedPropertiesBinder) {
Object extension = ((ExtendedPropertiesBinder) binder)
.getExtendedProducerProperties(outputName);
if (binder instanceof ExtendedPropertiesBinder extendedPropertiesBinder) {
Object extension = extendedPropertiesBinder.getExtendedProducerProperties(outputName);
ExtendedProducerProperties extendedProducerProperties = new ExtendedProducerProperties<>(
extension);
BeanUtils.copyProperties(producerProperties, extendedProducerProperties);
@@ -310,9 +310,8 @@ public class BindingService {
@SuppressWarnings("rawtypes")
public Object getExtendedProducerProperties(Object output, String outputName) {
Binder binder = getBinder(outputName, output.getClass());
if (binder instanceof ExtendedPropertiesBinder) {
return ((ExtendedPropertiesBinder) binder)
.getExtendedProducerProperties(outputName);
if (binder instanceof ExtendedPropertiesBinder extendedPropertiesBinder) {
return extendedPropertiesBinder.getExtendedProducerProperties(outputName);
}
return null;
}

15
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SubscribableChannelBindingTargetFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2018 the original author or authors.
* Copyright 2015-2023 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.
@@ -30,6 +30,7 @@ import org.springframework.messaging.SubscribableChannel;
* @author David Syer
* @author Ilayaperumal Gopinathan
* @author Oleg Zhurakousky
* @author Byungjun You
*/
public class SubscribableChannelBindingTargetFactory
extends AbstractBindingTargetFactory<SubscribableChannel> {
@@ -68,9 +69,9 @@ public class SubscribableChannelBindingTargetFactory
}
subscribableChannel = channel;
}
if (subscribableChannel instanceof DirectWithAttributesChannel) {
((DirectWithAttributesChannel) subscribableChannel).setAttribute("type", "input");
this.messageChannelConfigurer.configureInputChannel(subscribableChannel, name);
if (subscribableChannel instanceof DirectWithAttributesChannel directWithAttributesChannel) {
directWithAttributesChannel.setAttribute("type", "input");
this.messageChannelConfigurer.configureInputChannel(directWithAttributesChannel, name);
}
return subscribableChannel;
@@ -95,9 +96,9 @@ public class SubscribableChannelBindingTargetFactory
}
subscribableChannel = channel;
}
if (subscribableChannel instanceof DirectWithAttributesChannel) {
((DirectWithAttributesChannel) subscribableChannel).setAttribute("type", "output");
this.messageChannelConfigurer.configureOutputChannel(subscribableChannel, name);
if (subscribableChannel instanceof DirectWithAttributesChannel directWithAttributesChannel) {
directWithAttributesChannel.setAttribute("type", "output");
this.messageChannelConfigurer.configureOutputChannel(directWithAttributesChannel, name);
}
return subscribableChannel;

17
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ObjectStringMessageConverter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2018 the original author or authors.
* Copyright 2017-2023 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.
@@ -35,6 +35,7 @@ import org.springframework.util.MimeType;
*
* @author Marius Bogoevici
* @author Oleg Zhurakousky
* @author Byungjun You
* @since 1.2
*/
public class ObjectStringMessageConverter extends AbstractMessageConverter {
@@ -74,30 +75,28 @@ public class ObjectStringMessageConverter extends AbstractMessageConverter {
// Assert.isTrue(String.class.isAssignableFrom(targetClass) || targetClass == Object.class, "This converter can only convert byte[] to String");
if (message.getPayload() != null) {
if (message.getPayload() instanceof byte[]) {
if (message.getPayload() instanceof byte[] payloadAsBytes) {
if (byte[].class.isAssignableFrom(targetClass)) {
return message.getPayload();
}
else {
return new String((byte[]) message.getPayload(),
return new String(payloadAsBytes,
StandardCharsets.UTF_8);
}
}
else if (message.getPayload() instanceof Collection) {
Collection<?> collection = ((Collection<?>) message.getPayload()).stream()
else if (message.getPayload() instanceof Collection<?> payloadAsCollection) {
return payloadAsCollection.stream()
.map(value -> {
if (byte[].class.isAssignableFrom(targetClass)) {
return value;
}
else if (value instanceof byte[]) {
return new String((byte[]) value, StandardCharsets.UTF_8);
else if (value instanceof byte[] valueAsBytes) {
return new String(valueAsBytes, StandardCharsets.UTF_8);
}
else {
return value; // String
}
}).collect(Collectors.toList());
return collection;
}
else {
if (byte[].class.isAssignableFrom(targetClass)) {

46
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2022 the original author or authors.
* Copyright 2018-2023 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.
@@ -119,6 +119,7 @@ import org.springframework.util.StringUtils;
* @author Ilayaperumal Gopinathan
* @author Soby Chacko
* @author Chris Bono
* @author Byungjun You
* @since 2.1
*/
@AutoConfiguration
@@ -297,8 +298,9 @@ public class FunctionConfiguration {
boolean splittable = pollable != null
&& (boolean) AnnotationUtils.getAnnotationAttributes(pollable).get("splittable");
FunctionInvocationWrapper function = (supplier instanceof PartitionAwareFunctionWrapper)
? (FunctionInvocationWrapper) ((PartitionAwareFunctionWrapper) supplier).function : (FunctionInvocationWrapper) supplier;
FunctionInvocationWrapper function =
(supplier instanceof PartitionAwareFunctionWrapper partitionAwareFunctionWrapper)
? (FunctionInvocationWrapper) partitionAwareFunctionWrapper.function : (FunctionInvocationWrapper) supplier;
boolean reactive = FunctionTypeUtils.isPublisher(function.getOutputType());
if (pollable == null && reactive) {
@@ -361,17 +363,17 @@ public class FunctionConfiguration {
String supplierFunctionName = StringUtils
.delimitedListToStringArray(proxyFactory.getFunctionDefinition().replaceAll(",", "|").trim(), "|")[0];
BeanDefinition bd = context.getBeanDefinition(supplierFunctionName);
if (!(bd instanceof RootBeanDefinition)) {
if (!(bd instanceof RootBeanDefinition rootBeanDefinition)) {
return null;
}
Method factoryMethod = ((RootBeanDefinition) bd).getResolvedFactoryMethod();
Method factoryMethod = rootBeanDefinition.getResolvedFactoryMethod();
if (factoryMethod == null) {
Object source = bd.getSource();
if (source instanceof MethodMetadata) {
Class<?> factory = ClassUtils.resolveClassName(((MethodMetadata) source).getDeclaringClassName(), null);
if (source instanceof MethodMetadata methodMetadata) {
Class<?> factory = ClassUtils.resolveClassName(methodMetadata.getDeclaringClassName(), null);
Class<?>[] params = FunctionContextUtils.getParamTypesFromBeanDefinitionFactory(factory, (RootBeanDefinition) bd);
factoryMethod = ReflectionUtils.findMethod(factory, ((MethodMetadata) source).getMethodName(), params);
factoryMethod = ReflectionUtils.findMethod(factory, methodMetadata.getMethodName(), params);
}
}
Assert.notNull(factoryMethod, "Failed to introspect factory method since it was not discovered for function '"
@@ -384,8 +386,8 @@ public class FunctionConfiguration {
@SuppressWarnings("unchecked")
private <T> Message<T> wrapToMessageIfNecessary(T value) {
return value instanceof Message
? (Message<T>) value
return value instanceof Message message
? message
: MessageBuilder.withPayload(value).build();
}
@@ -436,8 +438,9 @@ public class FunctionConfiguration {
Map<String, BindableProxyFactory> beansOfType = applicationContext.getBeansOfType(BindableProxyFactory.class);
this.bindableProxyFactories = beansOfType.values().toArray(new BindableProxyFactory[0]);
for (BindableProxyFactory bindableProxyFactory : this.bindableProxyFactories) {
String functionDefinition = bindableProxyFactory instanceof BindableFunctionProxyFactory && ((BindableFunctionProxyFactory) bindableProxyFactory).isFunctionExist()
? ((BindableFunctionProxyFactory) bindableProxyFactory).getFunctionDefinition()
String functionDefinition = bindableProxyFactory instanceof BindableFunctionProxyFactory functionFactory
&& functionFactory.isFunctionExist()
? functionFactory.getFunctionDefinition()
: null; /*this.functionProperties.getDefinition();*/
boolean shouldNotProcess = false;
@@ -558,17 +561,14 @@ public class FunctionConfiguration {
if (!CollectionUtils.isEmpty(outputBindingNames)) {
MessageChannel outputChannel = this.applicationContext.getBean(outputBindingIter.next(), MessageChannel.class);
flux = flux.doOnNext(message -> {
if (message instanceof Message && ((Message<?>) message).getHeaders().get("spring.cloud.stream.sendto.destination") != null) {
String destinationName = (String) ((Message<?>) message).getHeaders().get("spring.cloud.stream.sendto.destination");
if (message instanceof Message m && m.getHeaders().get("spring.cloud.stream.sendto.destination") != null) {
String destinationName = (String) m.getHeaders().get("spring.cloud.stream.sendto.destination");
ProducerProperties producerProperties = this.serviceProperties.getBindings().get(outputBindingNames.iterator().next()).getProducer();
MessageChannel dynamicChannel = streamBridge.resolveDestination(destinationName, producerProperties, null);
if (logger.isInfoEnabled()) {
logger.info("Output message is sent to '" + destinationName + "' destination");
}
if (!(message instanceof Message)) {
message = MessageBuilder.withPayload(message).build();
}
dynamicChannel.send((Message) message);
dynamicChannel.send(m);
}
else {
if (!(message instanceof Message)) {
@@ -623,8 +623,8 @@ public class FunctionConfiguration {
logger.debug("Function execution resulted in null. No message will be sent");
return;
}
if (result instanceof Iterable) {
for (Object resultElement : (Iterable<?>) result) {
if (result instanceof Iterable iterableResult) {
for (Object resultElement : iterableResult) {
this.doSendMessage(resultElement, message);
}
}
@@ -639,13 +639,13 @@ public class FunctionConfiguration {
}
private void doSendMessage(Object result, Message<?> requestMessage) {
if (result instanceof Message && ((Message<?>) result).getHeaders().get("spring.cloud.stream.sendto.destination") != null) {
String destinationName = (String) ((Message<?>) result).getHeaders().get("spring.cloud.stream.sendto.destination");
if (result instanceof Message messageResult && messageResult.getHeaders().get("spring.cloud.stream.sendto.destination") != null) {
String destinationName = (String) messageResult.getHeaders().get("spring.cloud.stream.sendto.destination");
MessageChannel outputChannel = streamBridge.resolveDestination(destinationName, producerProperties, null);
if (logger.isInfoEnabled()) {
logger.info("Output message is sent to '" + destinationName + "' destination");
}
outputChannel.send(((Message<?>) result));
outputChannel.send(messageResult);
}
else if (StringUtils.hasText(outputChannelName)) {
if (!(result instanceof Message)) {

14
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/PartitionAwareFunctionWrapper.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2021 the original author or authors.
* Copyright 2020-2023 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.
@@ -37,6 +37,10 @@ import org.springframework.util.ObjectUtils;
* This class is effectively a wrapper which is aware of the stream related partition information
* for outgoing messages. It has only one responsibility and that is to modify the result message
* with 'scst_partition' header if necessary.
*
* @author Oleg Zhurakousky
* @author Soby Chacko
* @author Byungjun You
*/
class PartitionAwareFunctionWrapper implements Function<Object, Object>, Supplier<Object> {
@@ -90,16 +94,16 @@ class PartitionAwareFunctionWrapper implements Function<Object, Object>, Supplie
@Override
public Object get() {
if (this.function instanceof FunctionInvocationWrapper) {
if (this.function instanceof FunctionInvocationWrapper functionInvocationWrapper) {
this.setEnhancerIfNecessary();
return ((FunctionInvocationWrapper) this.function).get();
return functionInvocationWrapper.get();
}
throw new IllegalStateException("Call to get() is not allowed since this function is not a Supplier.");
}
private void setEnhancerIfNecessary() {
if (this.function instanceof FunctionInvocationWrapper) {
((FunctionInvocationWrapper) this.function).setEnhancer(this.outputMessageEnricher);
if (this.function instanceof FunctionInvocationWrapper functionInvocationWrapper) {
functionInvocationWrapper.setEnhancer(this.outputMessageEnricher);
}
}
}

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

@@ -72,6 +72,7 @@ import org.springframework.util.StringUtils;
*
* @author Oleg Zhurakousky
* @author Soby Chacko
* @author Byungjun You
* @since 3.0.3
*
*/
@@ -164,8 +165,8 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
String targetType = this.resolveBinderTargetType(bindingName, binderName, MessageChannel.class,
this.applicationContext.getBean(BinderFactory.class));
Message<?> messageToSend = data instanceof Message
? MessageBuilder.fromMessage((Message) data).setHeaderIfAbsent(MessageUtils.TARGET_PROTOCOL, targetType).build()
Message<?> messageToSend = data instanceof Message messageData
? MessageBuilder.fromMessage(messageData).setHeaderIfAbsent(MessageUtils.TARGET_PROTOCOL, targetType).build()
: new GenericMessage<>(data, Collections.singletonMap(MessageUtils.TARGET_PROTOCOL, targetType));
Message<?> resultMessage;

4
docs/src/main/asciidoc/kafka/kafka_overview.adoc
View File

@@ -944,8 +944,8 @@ to apply, you need to use a 'BinderCustomizer' to set the container customizer (
@Bean
public BinderCustomizer binderCustomizer(ListenerContainerWithDlqAndRetryCustomizer containerCustomizer) {
return (binder, binderName) -> {
if (binder instanceof KafkaMessageChannelBinder) {
((KafkaMessageChannelBinder) binder).setContainerCustomizer(containerCustomizer);
if (binder instanceof KafkaMessageChannelBinder kafkaMessageChannelBinder) {
kafkaMessageChannelBinder.setContainerCustomizer(containerCustomizer);
}
else if (binder instanceof KStreamBinder) {
...

4
docs/src/main/asciidoc/spring-cloud-stream.adoc
View File

@@ -1879,8 +1879,8 @@ Here is an example of providing a `BinderCustomizer` bean.
@Bean
public BinderCustomizer binderCustomizer() {
return (binder, binderName) -> {
if (binder instanceof KafkaMessageChannelBinder) {
((KafkaMessageChannelBinder) binder).setRebalanceListener(...);
if (binder instanceof KafkaMessageChannelBinder kafkaMessageChannelBinder) {
kafkaMessageChannelBinder.setRebalanceListener(...);
}
else if (binder instanceof KStreamBinder) {
...

11
schema-registry/spring-cloud-stream-schema-registry-client/src/main/java/org/springframework/cloud/stream/schema/registry/avro/AbstractAvroMessageConverter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2018 the original author or authors.
* Copyright 2016-2023 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,7 @@ import org.springframework.util.MimeType;
* @author Vinicius Carvalho
* @author Sercan Karaoglu
* @author Ish Mahajan
* @author Byungjun You
*/
public abstract class AbstractAvroMessageConverter extends AbstractMessageConverter {
@@ -93,8 +94,8 @@ public abstract class AbstractAvroMessageConverter extends AbstractMessageConver
MimeType mimeType = getContentTypeResolver().resolve(message.getHeaders());
if (mimeType == null) {
if (conversionHint instanceof MimeType) {
mimeType = (MimeType) conversionHint;
if (conversionHint instanceof MimeType hintedMimeType) {
mimeType = hintedMimeType;
}
else {
return null;
@@ -117,8 +118,8 @@ public abstract class AbstractAvroMessageConverter extends AbstractMessageConver
ByteArrayOutputStream baos = new ByteArrayOutputStream();
try {
MimeType hintedContentType = null;
if (conversionHint instanceof MimeType) {
hintedContentType = (MimeType) conversionHint;
if (conversionHint instanceof MimeType mimeType) {
hintedContentType = mimeType;
}
Schema schema = resolveSchemaForWriting(payload, headers, hintedContentType);
@SuppressWarnings("unchecked")

10
schema-registry/spring-cloud-stream-schema-registry-client/src/main/java/org/springframework/cloud/stream/schema/registry/avro/OriginalContentTypeResolver.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2018 the original author or authors.
* Copyright 2017-2023 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.
@@ -25,6 +25,7 @@ import org.springframework.util.MimeType;
/**
* @author Vinicius Carvalho
* @author Byungjun You
*
* Resolves contentType looking for a originalContentType header first. If not found
* returns the contentType
@@ -43,13 +44,12 @@ class OriginalContentTypeResolver implements ContentTypeResolver {
? headers.get(BINDER_ORIGINAL_CONTENT_TYPE)
: headers.get(MessageHeaders.CONTENT_TYPE);
MimeType mimeType = null;
if (contentType instanceof MimeType) {
mimeType = (MimeType) contentType;
if (contentType instanceof MimeType mimeContentType) {
mimeType = mimeContentType;
}
else if (contentType instanceof String) {
else if (contentType instanceof String valueAsString) {
mimeType = this.mimeTypeCache.get(contentType);
if (mimeType == null) {
String valueAsString = (String) contentType;
mimeType = MimeType.valueOf(valueAsString);
this.mimeTypeCache.put(valueAsString, mimeType);
}

7
schema-registry/spring-cloud-stream-schema-registry-core/src/main/java/org/springframework/cloud/stream/schema/registry/config/SchemaServerConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-2023 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.
@@ -37,6 +37,7 @@ import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
/**
* @author Vinicius Carvalho
* @author Soby Chacko
* @author Byungjun You
*/
@Configuration(proxyBeanMethods = false)
@EnableJpaRepositories(basePackageClasses = SchemaRepository.class)
@@ -47,8 +48,8 @@ public class SchemaServerConfiguration {
@Bean
public static BeanFactoryPostProcessor entityScanPackagesPostProcessor() {
return beanFactory -> {
if (beanFactory instanceof BeanDefinitionRegistry) {
EntityScanPackages.register((BeanDefinitionRegistry) beanFactory,
if (beanFactory instanceof BeanDefinitionRegistry beanDefinitionRegistry) {
EntityScanPackages.register(beanDefinitionRegistry,
Collections.singletonList(Schema.class.getPackage().getName()));
}
};