Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix more issues from Sonar report

Browse Source
This commit is contained in:
Artem Bilan
2022-11-03 12:44:26 -04:00
parent 0b9bab313b
commit 5f1c0c17de
13 changed files with 60 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ObservationPropagationChannelInterceptor.java
View File

@@ -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();
}

16
spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java
View File

@@ -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());
}
}
}

2
spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java
View File

@@ -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);

39
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java
View File

@@ -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) {

2
spring-integration-core/src/test/java/org/springframework/integration/config/xml/DelayerUsageTests.java
View File

@@ -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

4
spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java
View File

@@ -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

8
spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/HazelcastClusterMonitorMessageProducer.java
View File

@@ -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<String> monitorTypes =
Set<String> 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

6
spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/store/HazelcastMessageStore.java
View File

@@ -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("\\*", "%")));
}
}

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java
View File

@@ -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.

6
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java
View File

@@ -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.
*
* <p>
* 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;

9
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java
View File

@@ -305,14 +305,13 @@ public class DefaultSftpSessionFactory implements SessionFactory<SftpClient.DirE
initClient();
Duration verifyTimeout = this.timeout != null ? Duration.ofMillis(this.timeout) : null;
HostConfigEntry hostConfig = this.hostConfig;
if (hostConfig == null) {
hostConfig =
new HostConfigEntry(SshdSocketAddress.isIPv6Address(this.host) ? "" : this.host, this.host,
HostConfigEntry config = this.hostConfig;
if (config == null) {
config = new HostConfigEntry(SshdSocketAddress.isIPv6Address(this.host) ? "" : this.host, this.host,
this.port, this.user);
}
ClientSession clientSession =
this.sshClient.connect(hostConfig)
this.sshClient.connect(config)
.verify(verifyTimeout)
.getSession();

6
spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java
View File

@@ -517,9 +517,9 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W
}
private static MediaType selectMoreSpecificMediaType(MediaType acceptable, MediaType producible) {
producible = producible.copyQualityValue(acceptable);
if (acceptable.isLessSpecific(producible)) {
return producible;
MediaType producibleToUse = producible.copyQualityValue(acceptable);
if (acceptable.isLessSpecific(producibleToUse)) {
return producibleToUse;
}
else {
return acceptable;

6
spring-integration-xml/src/main/java/org/springframework/integration/xml/AggregatedXmlMessageValidationException.java
View File

@@ -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.
@@ -41,7 +41,9 @@ public class AggregatedXmlMessageValidationException extends RuntimeException {
public String getMessage() {
StringBuilder message = new StringBuilder("Multiple causes:\n");
for (Throwable exception : this.exceptions) {
message.append(" " + exception.getMessage() + "\n");
message.append(" ")
.append(exception.getMessage())
.append("\n");
}
return message.toString();
}