diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java index 377b958ce1..ad6b7ec913 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2024 the original author or authors. + * Copyright 2002-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -28,7 +28,6 @@ import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.core.convert.ConversionService; -import org.springframework.core.convert.support.DefaultConversionService; import org.springframework.core.log.LogAccessor; import org.springframework.expression.Expression; import org.springframework.expression.ExpressionParser; @@ -72,8 +71,6 @@ public abstract class IntegrationObjectSupport implements ComponentSourceAware, protected final LogAccessor logger = new LogAccessor(getClass()); // NOSONAR protected - private final ConversionService defaultConversionService = DefaultConversionService.getSharedInstance(); - private DestinationResolver channelResolver; private String beanName; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java index 8142d00ee0..ae4000828e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,7 +16,6 @@ package org.springframework.integration.endpoint; -import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; import org.springframework.beans.factory.DisposableBean; @@ -51,8 +50,6 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport protected final ReentrantLock lifecycleLock = new ReentrantLock(); // NOSONAR - protected final Condition lifecycleCondition = this.lifecycleLock.newCondition(); // NOSONAR - private String role; private SmartLifecycleRoleController roleController; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractFetchLimitingMessageSource.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractFetchLimitingMessageSource.java index 98c9e41f37..cb5013ffa4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractFetchLimitingMessageSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractFetchLimitingMessageSource.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2021 the original author or authors. + * Copyright 2016-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -48,7 +48,7 @@ public abstract class AbstractFetchLimitingMessageSource extends AbstractMess } /** - * Subclasses must implement this method. Typically the returned value will be the + * Subclasses must implement this method. Typically, the returned value will be the * payload of type T, but the returned value may also be a Message instance whose * payload is of type T. * @param maxFetchSizeToReceive the maximum number of messages to fetch if a fetch is diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java index 1f33c51e7d..556ff79b24 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -156,11 +156,11 @@ public abstract class AbstractMessageSource extends AbstractExpressionEvaluat } Message message; Map headers = evaluateHeaders(); - if (result instanceof AbstractIntegrationMessageBuilder) { + if (result instanceof AbstractIntegrationMessageBuilder messageBuilder) { if (!CollectionUtils.isEmpty(headers)) { - ((AbstractIntegrationMessageBuilder) result).copyHeaders(headers); + messageBuilder.copyHeaders(headers); } - message = ((AbstractIntegrationMessageBuilder) result).build(); + message = messageBuilder.build(); } else if (result instanceof Message) { message = (Message) result; @@ -216,7 +216,7 @@ public abstract class AbstractMessageSource extends AbstractExpressionEvaluat } /** - * Subclasses must implement this method. Typically the returned value will be the {@code payload} of + * Subclasses must implement this method. Typically, the returned value will be the {@code payload} of * type T, but the returned value may also be a {@link Message} instance whose payload is of type T; * also can be {@link AbstractIntegrationMessageBuilder} which is used for additional headers population. * @return The value returned. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java index c3b7860e46..05a68e88ca 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java @@ -26,7 +26,6 @@ import java.util.concurrent.Executor; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; -import java.util.stream.Collectors; import org.aopalliance.aop.Advice; import org.reactivestreams.Subscription; @@ -153,8 +152,8 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement /** * Configure a cap for messages to poll from the source per scheduling cycle. * A negative number means retrieve unlimited messages until the {@code MessageSource} returns {@code null}. - * Zero means do not poll for any records - it - * can be considered as pausing if 'maxMessagesPerPoll' is later changed to a non-zero value. + * Zero means do not poll for any records - + * it can be considered as pausing if 'maxMessagesPerPoll' is later changed to a non-zero value. * The polling cycle may exit earlier if the source returns null for the current receive call. * @param maxMessagesPerPoll the number of message to poll per schedule. */ @@ -176,8 +175,8 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement this.beanClassLoader = classLoader; } - public void setTransactionSynchronizationFactory(TransactionSynchronizationFactory - transactionSynchronizationFactory) { + public void setTransactionSynchronizationFactory( + TransactionSynchronizationFactory transactionSynchronizationFactory) { this.transactionSynchronizationFactory = transactionSynchronizationFactory; } @@ -189,7 +188,9 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement * @since 4.3 */ public MessageChannel getDefaultErrorChannel() { - if (!this.errorHandlerIsDefault && this.errorHandler instanceof MessagePublishingErrorHandler messagePublishingErrorHandler) { + if (!this.errorHandlerIsDefault && this.errorHandler + instanceof MessagePublishingErrorHandler messagePublishingErrorHandler) { + return messagePublishingErrorHandler.getDefaultErrorChannel(); } else { @@ -323,7 +324,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement if (!CollectionUtils.isEmpty(this.adviceChain)) { receiveOnlyAdviceChain = this.adviceChain.stream() .filter(this::isReceiveOnlyAdvice) - .collect(Collectors.toList()); + .toList(); } Callable> task = this::doPoll; @@ -423,15 +424,15 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement } catch (Exception ex) { pollingTaskError = ex; - if (ex instanceof MessagingException) { // NOSONAR - throw (MessagingException) ex; + if (ex instanceof MessagingException messagingException) { // NOSONAR + throw messagingException; } else { Message failedMessage = null; if (this.transactionSynchronizationFactory != null) { Object resource = TransactionSynchronizationManager.getResource(getResourceToBind()); - if (resource instanceof IntegrationResourceHolder) { - failedMessage = ((IntegrationResourceHolder) resource).getMessage(); + if (resource instanceof IntegrationResourceHolder integrationResourceHolder) { + failedMessage = integrationResourceHolder.getMessage(); } } throw new MessagingException(failedMessage, ex); // NOSONAR (null failedMessage) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ExpressionMessageProducerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ExpressionMessageProducerSupport.java index 1b028d4676..bea37a6e81 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ExpressionMessageProducerSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ExpressionMessageProducerSupport.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -22,7 +22,7 @@ import org.springframework.integration.expression.ExpressionUtils; import org.springframework.util.Assert; /** - * A {@link MessageProducerSupport} sub-class that provides {@linkplain #payloadExpression} + * A {@link MessageProducerSupport} subclass that provides {@linkplain #payloadExpression} * evaluation with result as a payload for Message to send. * * @author David Turanski diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/IntegrationConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/IntegrationConsumer.java index 4f502a3abf..fd4ae19154 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/IntegrationConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/IntegrationConsumer.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -37,7 +37,7 @@ public interface IntegrationConsumer extends NamedComponent { MessageChannel getInputChannel(); /** - * Return the output channel (may be null). + * Return the output channel (maybe null). * @return the output channel. */ MessageChannel getOutputChannel(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MethodInvokingMessageSource.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MethodInvokingMessageSource.java index 32d3c95abc..9e25131db3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MethodInvokingMessageSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MethodInvokingMessageSource.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2024 the original author or authors. + * Copyright 2002-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -93,15 +93,15 @@ public class MethodInvokingMessageSource extends AbstractMessageSource i @Override public void start() { - if (this.object instanceof Lifecycle) { - ((Lifecycle) this.object).start(); + if (this.object instanceof Lifecycle lifecycle) { + lifecycle.start(); } } @Override public void stop() { - if (this.object instanceof Lifecycle) { - ((Lifecycle) this.object).stop(); + if (this.object instanceof Lifecycle lifecycle) { + lifecycle.stop(); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java index a9cb5c39ac..0105d4a010 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2024 the original author or authors. + * Copyright 2002-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -74,8 +74,8 @@ public class PollingConsumer extends AbstractPollingEndpoint implements Integrat } this.inputChannel = inputChannel; this.handler = handler; - if (this.inputChannel instanceof ExecutorChannelInterceptorAware) { - this.channelInterceptors = ((ExecutorChannelInterceptorAware) this.inputChannel).getInterceptors(); + if (this.inputChannel instanceof ExecutorChannelInterceptorAware executorChannelInterceptorAware) { + this.channelInterceptors = executorChannelInterceptorAware.getInterceptors(); } else { this.channelInterceptors = null; @@ -93,11 +93,11 @@ public class PollingConsumer extends AbstractPollingEndpoint implements Integrat @Override public MessageChannel getOutputChannel() { - if (this.handler instanceof MessageProducer) { - return ((MessageProducer) this.handler).getOutputChannel(); + if (this.handler instanceof MessageProducer messageProducer) { + return messageProducer.getOutputChannel(); } - else if (this.handler instanceof MessageRouter) { - return ((MessageRouter) this.handler).getDefaultOutputChannel(); + else if (this.handler instanceof MessageRouter messageRouter) { + return messageRouter.getDefaultOutputChannel(); } else { return null; @@ -127,16 +127,16 @@ public class PollingConsumer extends AbstractPollingEndpoint implements Integrat @Override protected void doStart() { - if (this.handler instanceof Lifecycle) { - ((Lifecycle) this.handler).start(); + if (this.handler instanceof Lifecycle lifecycle) { + lifecycle.start(); } super.doStart(); } @Override protected void doStop() { - if (this.handler instanceof Lifecycle) { - ((Lifecycle) this.handler).stop(); + if (this.handler instanceof Lifecycle lifecycle) { + lifecycle.stop(); } super.doStop(); } @@ -180,8 +180,7 @@ public class PollingConsumer extends AbstractPollingEndpoint implements Integrat private Message applyBeforeHandle(Message message, Deque interceptorStack) { Message theMessage = message; for (ChannelInterceptor interceptor : this.channelInterceptors) { - if (interceptor instanceof ExecutorChannelInterceptor) { - ExecutorChannelInterceptor executorInterceptor = (ExecutorChannelInterceptor) interceptor; + if (interceptor instanceof ExecutorChannelInterceptor executorInterceptor) { theMessage = executorInterceptor.beforeHandle(theMessage, this.inputChannel, this.handler); if (theMessage == null) { logger.debug(() -> executorInterceptor.getClass().getSimpleName() diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java index 1db3f51667..f01663e594 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2024 the original author or authors. + * Copyright 2016-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -100,11 +100,11 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra this.publisher = IntegrationReactiveUtils.messageChannelToFlux(inputChannel); this.subscriber = subscriber; this.lifecycleDelegate = subscriber instanceof Lifecycle ? (Lifecycle) subscriber : null; - if (subscriber instanceof MessageHandlerSubscriber) { - this.handler = ((MessageHandlerSubscriber) subscriber).messageHandler; + if (subscriber instanceof MessageHandlerSubscriber messageHandlerSubscriber) { + this.handler = messageHandlerSubscriber.messageHandler; } - else if (subscriber instanceof MessageHandler) { - this.handler = (MessageHandler) subscriber; + else if (subscriber instanceof MessageHandler messageHandler) { + this.handler = messageHandler; } else { this.handler = this.subscriber::onNext; @@ -126,7 +126,7 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra this.publisher = IntegrationReactiveUtils.messageChannelToFlux(inputChannel); this.subscriber = null; this.lifecycleDelegate = - reactiveMessageHandler instanceof Lifecycle ? (Lifecycle) reactiveMessageHandler : null; + reactiveMessageHandler instanceof Lifecycle lifecycle ? lifecycle : null; } public void setErrorHandler(ErrorHandler errorHandler) { @@ -146,11 +146,11 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra @Override public MessageChannel getOutputChannel() { - if (this.handler instanceof MessageProducer) { - return ((MessageProducer) this.handler).getOutputChannel(); + if (this.handler instanceof MessageProducer messageProducer) { + return messageProducer.getOutputChannel(); } - else if (this.handler instanceof MessageRouter) { - return ((MessageRouter) this.handler).getDefaultOutputChannel(); + else if (this.handler instanceof MessageRouter messageRouter) { + return messageRouter.getDefaultOutputChannel(); } else { return null; @@ -258,15 +258,15 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra @Override public void start() { - if (this.messageHandler instanceof Lifecycle) { - ((Lifecycle) this.messageHandler).start(); + if (this.messageHandler instanceof Lifecycle lifecycle) { + lifecycle.start(); } } @Override public void stop() { - if (this.messageHandler instanceof Lifecycle) { - ((Lifecycle) this.messageHandler).stop(); + if (this.messageHandler instanceof Lifecycle lifecycle) { + lifecycle.stop(); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java index 15fad23d68..b635038dc5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java @@ -62,6 +62,8 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint private final MessagingTemplate messagingTemplate = new MessagingTemplate(); + private final Lock lock = new ReentrantLock(); + private MessageSource originalSource; private ObservationRegistry observationRegistry = ObservationRegistry.NOOP; @@ -76,8 +78,6 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint private volatile boolean shouldTrack; - private final Lock lock = new ReentrantLock(); - /** * Specify the source to be polled for Messages. * @param source The message source.