diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ObservationPropagationChannelInterceptor.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ObservationPropagationChannelInterceptor.java index f12ca5cdd3..f73ce05445 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ObservationPropagationChannelInterceptor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ObservationPropagationChannelInterceptor.java @@ -68,7 +68,7 @@ public class ObservationPropagationChannelInterceptor extends ThreadStatePropaga @Override public void afterMessageHandled(Message message, MessageChannel channel, MessageHandler handler, Exception ex) { Observation.Scope scope = this.scopes.get(); - if (scope != null && scope == this.observationRegistry.getCurrentObservationScope()) { + if (scope != null && scope.equals(this.observationRegistry.getCurrentObservationScope())) { scope.close(); this.scopes.remove(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java index 6d7a2e093d..a282e9a917 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -32,6 +32,7 @@ import org.springframework.util.StringUtils; * @author Mark Fisher * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan */ public class EventDrivenConsumer extends AbstractEndpoint implements IntegrationConsumer { @@ -95,9 +96,15 @@ public class EventDrivenConsumer extends AbstractEndpoint implements Integration String componentType = ((NamedComponent) this.handler).getComponentType(); componentType = StringUtils.hasText(componentType) ? componentType : ""; String componentName = getComponentName(); - componentName = (StringUtils.hasText(componentName) && componentName.contains("#")) ? "" : ":" + componentName; - StringBuffer buffer = new StringBuffer(); - buffer.append("{" + componentType + componentName + "} as a subscriber to the '" + channelName + "' channel"); + componentName = + (StringUtils.hasText(componentName) && componentName.contains("#")) ? "" : ":" + componentName; + StringBuilder buffer = new StringBuilder(); + buffer.append("{") + .append(componentType) + .append(componentName) + .append("} as a subscriber to the '") + .append(channelName) + .append("' channel"); if (add) { buffer.insert(0, "Adding "); } @@ -107,4 +114,5 @@ public class EventDrivenConsumer extends AbstractEndpoint implements Integration logger.info(buffer.toString()); } } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java index 6370c18f8c..87e4324927 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java @@ -272,7 +272,7 @@ public abstract class MessageProducerSupport extends AbstractEndpoint .map(this::trackMessageIfAny) .doOnComplete(this::stop) .doOnCancel(this::stop) - .doOnSubscribe((subscription) -> this.subscription = subscription); + .doOnSubscribe((subs) -> this.subscription = subs); if (channelForSubscription instanceof ReactiveStreamsSubscribableChannel) { ((ReactiveStreamsSubscribableChannel) channelForSubscription).subscribeTo(messageFlux); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java index 4aea97325a..76badb9bde 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java @@ -309,29 +309,30 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan replyChannel = getOutputChannel(); } - ReactiveAdapter reactiveAdapter = null; + if (this.async) { + ReactiveAdapter reactiveAdapter = ReactiveAdapterRegistry.getSharedInstance().getAdapter(null, reply); + if (reply instanceof org.springframework.util.concurrent.ListenableFuture + || reply instanceof CompletableFuture + || reactiveAdapter != null) { - if (this.async && - (reply instanceof org.springframework.util.concurrent.ListenableFuture - || reply instanceof CompletableFuture - || (reactiveAdapter = ReactiveAdapterRegistry.getSharedInstance().getAdapter(null, reply)) != null)) { + if (replyChannel instanceof ReactiveStreamsSubscribableChannel reactiveStreamsSubscribableChannel) { + Publisher reactiveReply = toPublisherReply(reply, reactiveAdapter); + reactiveStreamsSubscribableChannel + .subscribeTo( + Flux.from(reactiveReply) + .doOnError((ex) -> sendErrorMessage(requestMessage, ex)) + .map(result -> createOutputMessage(result, requestHeaders))); + } + else { + CompletableFuture futureReply = toFutureReply(reply, reactiveAdapter); + futureReply.whenComplete(new ReplyFutureCallback(requestMessage, replyChannel)); + } - if (replyChannel instanceof ReactiveStreamsSubscribableChannel reactiveStreamsSubscribableChannel) { - Publisher reactiveReply = toPublisherReply(reply, reactiveAdapter); - reactiveStreamsSubscribableChannel - .subscribeTo( - Flux.from(reactiveReply) - .doOnError((ex) -> sendErrorMessage(requestMessage, ex)) - .map(result -> createOutputMessage(result, requestHeaders))); - } - else { - CompletableFuture futureReply = toFutureReply(reply, reactiveAdapter); - futureReply.whenComplete(new ReplyFutureCallback(requestMessage, replyChannel)); + return; } } - else { - sendOutput(createOutputMessage(reply, requestHeaders), replyChannel, false); - } + + sendOutput(createOutputMessage(reply, requestHeaders), replyChannel, false); } private static Publisher toPublisherReply(Object reply, @Nullable ReactiveAdapter reactiveAdapter) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests.java index d2eaf394a7..c23bd6251e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests.java @@ -91,7 +91,7 @@ public class DelayerUsageTests { long start = System.currentTimeMillis(); inputA.send(builder.build()); assertThat(outputA.receive(10000)).isNotNull(); - assertThat(System.currentTimeMillis() - start).isCloseTo(2000, withinPercentage(5)); + assertThat(System.currentTimeMillis() - start).isCloseTo(2000, withinPercentage(10)); } @Test diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java index 981d786dc1..4e1e94b4d9 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java @@ -17,6 +17,7 @@ package org.springframework.integration.dsl.flowservices; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.withinPercentage; import java.time.Instant; import java.util.Collection; @@ -25,7 +26,6 @@ import java.util.List; import java.util.Optional; import java.util.concurrent.atomic.AtomicReference; -import org.assertj.core.data.Percentage; import org.junit.jupiter.api.Test; import org.springframework.aop.framework.Advised; @@ -145,7 +145,7 @@ public class FlowServiceTests { .isEqualTo("B"); assertThat(receive2.getHeaders().getTimestamp() - receive1.getHeaders().getTimestamp()) - .isCloseTo(500, Percentage.withPercentage(10)); + .isCloseTo(500, withinPercentage(20)); } @Configuration diff --git a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/HazelcastClusterMonitorMessageProducer.java b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/HazelcastClusterMonitorMessageProducer.java index 309de1a2fe..ce51d3dcec 100644 --- a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/HazelcastClusterMonitorMessageProducer.java +++ b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/HazelcastClusterMonitorMessageProducer.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -64,10 +64,10 @@ public class HazelcastClusterMonitorMessageProducer extends MessageProducerSuppo } public void setMonitorEventTypes(String monitorEventTypes) { - final Set monitorTypes = + Set types = HazelcastIntegrationDefinitionValidator.validateEnumType(ClusterMonitorType.class, monitorEventTypes); - Assert.notEmpty(monitorTypes, "'monitorTypes' must have elements"); - this.monitorTypes = monitorTypes; + Assert.notEmpty(types, "'monitorTypes' must have elements"); + this.monitorTypes = types; } @Override diff --git a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/store/HazelcastMessageStore.java b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/store/HazelcastMessageStore.java index 2ff361d2e1..851875f041 100644 --- a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/store/HazelcastMessageStore.java +++ b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/store/HazelcastMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2021 the original author or authors. + * Copyright 2017-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -76,8 +76,8 @@ public class HazelcastMessageStore extends AbstractKeyValueMessageStore { @Override protected Collection doListKeys(String keyPattern) { Assert.hasText(keyPattern, "'keyPattern' must not be empty"); - keyPattern = keyPattern.replaceAll("\\*", "%"); - return this.map.keySet(Predicates.like(QueryConstants.KEY_ATTRIBUTE_NAME.value(), keyPattern)); + return this.map.keySet(Predicates.like(QueryConstants.KEY_ATTRIBUTE_NAME.value(), + keyPattern.replaceAll("\\*", "%"))); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java index 88beffb380..647ac15ddf 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -59,7 +59,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece private int soSendBufferSize = -1; - private SocketCustomizer socketCustomizer = socket -> { }; + private SocketCustomizer socketCustomizer = (aSocket) -> { }; /** * Constructs a UnicastReceivingChannelAdapter that listens on the specified port. diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java index a654784fe5..d3e396669a 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2001-2020 the original author or authors. + * Copyright 2001-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -50,7 +50,7 @@ import org.springframework.util.StringUtils; /** * A {@link org.springframework.messaging.MessageHandler} implementation that maps a Message into * a UDP datagram packet and sends that to the specified host and port. - * + *

* Messages can be basic, with no support for reliability, can be prefixed * by a length so the receiving end can detect truncation, and can require * a UDP acknowledgment to confirm delivery. @@ -101,7 +101,7 @@ public class UnicastSendingMessageHandler extends private EvaluationContext evaluationContext; - private SocketCustomizer socketCustomizer = socket -> { }; + private SocketCustomizer socketCustomizer = (aSocket) -> { }; private volatile CountDownLatch ackLatch; diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java index 1450d9dd36..f131a76b20 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java @@ -305,14 +305,13 @@ public class DefaultSftpSessionFactory implements SessionFactory