Skip to content

Commit c570bee

Browse files
garyrussellartembilan
garyrussell
authored and
artembilan
committed
Fix Sonar complexity issues
Address a few complexity issues.
1 parent 9072e92 commit c570bee

File tree

3 files changed

+108
-170
lines changed

3 files changed

+108
-170
lines changed

‎spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java

+90-159
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import org.springframework.amqp.core.MessageDeliveryMode;
2727
import org.springframework.amqp.core.MessageProperties;
2828
import org.springframework.amqp.support.AmqpHeaders;
29+
import org.springframework.amqp.utils.JavaUtils;
2930
import org.springframework.integration.IntegrationMessageHeaderAccessor;
3031
import org.springframework.integration.mapping.AbstractHeaderMapper;
3132
import org.springframework.integration.mapping.support.JsonHeaders;
@@ -107,86 +108,55 @@ protected DefaultAmqpHeaderMapper(String[] requestHeaderNames, String[] replyHea
107108
protected Map<String, Object> extractStandardHeaders(MessageProperties amqpMessageProperties) {
108109
Map<String, Object> headers = new HashMap<String, Object>();
109110
try {
110-
String appId = amqpMessageProperties.getAppId();
111-
if (StringUtils.hasText(appId)) {
112-
headers.put(AmqpHeaders.APP_ID, appId);
113-
}
114-
String clusterId = amqpMessageProperties.getClusterId();
115-
if (StringUtils.hasText(clusterId)) {
116-
headers.put(AmqpHeaders.CLUSTER_ID, clusterId);
117-
}
118-
String contentEncoding = amqpMessageProperties.getContentEncoding();
119-
if (StringUtils.hasText(contentEncoding)) {
120-
headers.put(AmqpHeaders.CONTENT_ENCODING, contentEncoding);
121-
}
111+
JavaUtils.INSTANCE
112+
.acceptIfNotNull(AmqpHeaders.APP_ID, amqpMessageProperties.getAppId(),
113+
(key, value) -> headers.put(key, value))
114+
.acceptIfNotNull(AmqpHeaders.CLUSTER_ID, amqpMessageProperties.getClusterId(),
115+
(key, value) -> headers.put(key, value))
116+
.acceptIfNotNull(AmqpHeaders.CONTENT_ENCODING, amqpMessageProperties.getContentEncoding(),
117+
(key, value) -> headers.put(key, value));
122118
long contentLength = amqpMessageProperties.getContentLength();
123-
if (contentLength > 0) {
124-
headers.put(AmqpHeaders.CONTENT_LENGTH, contentLength);
125-
}
126-
String contentType = amqpMessageProperties.getContentType();
127-
if (StringUtils.hasText(contentType)) {
128-
headers.put(AmqpHeaders.CONTENT_TYPE, contentType);
129-
}
130-
String correlationId = amqpMessageProperties.getCorrelationId();
131-
if (StringUtils.hasText(correlationId)) {
132-
headers.put(AmqpHeaders.CORRELATION_ID, correlationId);
133-
}
134-
MessageDeliveryMode receivedDeliveryMode = amqpMessageProperties.getReceivedDeliveryMode();
135-
if (receivedDeliveryMode != null) {
136-
headers.put(AmqpHeaders.RECEIVED_DELIVERY_MODE, receivedDeliveryMode);
137-
}
119+
JavaUtils.INSTANCE
120+
.acceptIfCondition(contentLength > 0, AmqpHeaders.CONTENT_LENGTH, contentLength,
121+
(key, value) -> headers.put(key, value))
122+
.acceptIfHasText(AmqpHeaders.CONTENT_TYPE, amqpMessageProperties.getContentType(),
123+
(key, value) -> headers.put(key, value))
124+
.acceptIfHasText(AmqpHeaders.CORRELATION_ID, amqpMessageProperties.getCorrelationId(),
125+
(key, value) -> headers.put(key, value))
126+
.acceptIfNotNull(AmqpHeaders.RECEIVED_DELIVERY_MODE, amqpMessageProperties.getReceivedDeliveryMode(),
127+
(key, value) -> headers.put(key, value));
138128
long deliveryTag = amqpMessageProperties.getDeliveryTag();
139-
if (deliveryTag > 0) {
140-
headers.put(AmqpHeaders.DELIVERY_TAG, deliveryTag);
141-
}
142-
String expiration = amqpMessageProperties.getExpiration();
143-
if (StringUtils.hasText(expiration)) {
144-
headers.put(AmqpHeaders.EXPIRATION, expiration);
145-
}
129+
JavaUtils.INSTANCE
130+
.acceptIfCondition(deliveryTag > 0, AmqpHeaders.DELIVERY_TAG, deliveryTag,
131+
(key, value) -> headers.put(key, value))
132+
.acceptIfHasText(AmqpHeaders.EXPIRATION, amqpMessageProperties.getExpiration(),
133+
(key, value) -> headers.put(key, value));
146134
Integer messageCount = amqpMessageProperties.getMessageCount();
147-
if (messageCount != null && messageCount > 0) {
148-
headers.put(AmqpHeaders.MESSAGE_COUNT, messageCount);
149-
}
150-
String messageId = amqpMessageProperties.getMessageId();
151-
if (StringUtils.hasText(messageId)) {
152-
headers.put(AmqpHeaders.MESSAGE_ID, messageId);
153-
}
135+
JavaUtils.INSTANCE
136+
.acceptIfCondition(messageCount != null && messageCount > 0, AmqpHeaders.MESSAGE_COUNT, messageCount,
137+
(key, value) -> headers.put(key, value))
138+
.acceptIfHasText(AmqpHeaders.MESSAGE_ID, amqpMessageProperties.getMessageId(),
139+
(key, value) -> headers.put(key, value));
154140
Integer priority = amqpMessageProperties.getPriority();
155-
if (priority != null && priority > 0) {
156-
headers.put(IntegrationMessageHeaderAccessor.PRIORITY, priority);
157-
}
158-
Integer receivedDelay = amqpMessageProperties.getReceivedDelay();
159-
if (receivedDelay != null) {
160-
headers.put(AmqpHeaders.RECEIVED_DELAY, receivedDelay);
161-
}
162-
String receivedExchange = amqpMessageProperties.getReceivedExchange();
163-
if (StringUtils.hasText(receivedExchange)) {
164-
headers.put(AmqpHeaders.RECEIVED_EXCHANGE, receivedExchange);
165-
}
166-
String receivedRoutingKey = amqpMessageProperties.getReceivedRoutingKey();
167-
if (StringUtils.hasText(receivedRoutingKey)) {
168-
headers.put(AmqpHeaders.RECEIVED_ROUTING_KEY, receivedRoutingKey);
169-
}
170-
Boolean redelivered = amqpMessageProperties.isRedelivered();
171-
if (redelivered != null) {
172-
headers.put(AmqpHeaders.REDELIVERED, redelivered);
173-
}
174-
String replyTo = amqpMessageProperties.getReplyTo();
175-
if (replyTo != null) {
176-
headers.put(AmqpHeaders.REPLY_TO, replyTo);
177-
}
178-
Date timestamp = amqpMessageProperties.getTimestamp();
179-
if (timestamp != null) {
180-
headers.put(AmqpHeaders.TIMESTAMP, timestamp);
181-
}
182-
String type = amqpMessageProperties.getType();
183-
if (StringUtils.hasText(type)) {
184-
headers.put(AmqpHeaders.TYPE, type);
185-
}
186-
String userId = amqpMessageProperties.getReceivedUserId();
187-
if (StringUtils.hasText(userId)) {
188-
headers.put(AmqpHeaders.RECEIVED_USER_ID, userId);
189-
}
141+
JavaUtils.INSTANCE
142+
.acceptIfCondition(priority != null && priority > 0, IntegrationMessageHeaderAccessor.PRIORITY,
143+
priority, (key, value) -> headers.put(key, value))
144+
.acceptIfNotNull(AmqpHeaders.RECEIVED_DELAY, amqpMessageProperties.getReceivedDelay(),
145+
(key, value) -> headers.put(key, value))
146+
.acceptIfNotNull(AmqpHeaders.RECEIVED_EXCHANGE, amqpMessageProperties.getReceivedExchange(),
147+
(key, value) -> headers.put(key, value))
148+
.acceptIfHasText(AmqpHeaders.RECEIVED_ROUTING_KEY, amqpMessageProperties.getReceivedRoutingKey(),
149+
(key, value) -> headers.put(key, value))
150+
.acceptIfNotNull(AmqpHeaders.REDELIVERED, amqpMessageProperties.isRedelivered(),
151+
(key, value) -> headers.put(key, value))
152+
.acceptIfNotNull(AmqpHeaders.REPLY_TO, amqpMessageProperties.getReplyTo(),
153+
(key, value) -> headers.put(key, value))
154+
.acceptIfNotNull(AmqpHeaders.TIMESTAMP, amqpMessageProperties.getTimestamp(),
155+
(key, value) -> headers.put(key, value))
156+
.acceptIfHasText(AmqpHeaders.TYPE, amqpMessageProperties.getType(),
157+
(key, value) -> headers.put(key, value))
158+
.acceptIfHasText(AmqpHeaders.RECEIVED_USER_ID, amqpMessageProperties.getReceivedUserId(),
159+
(key, value) -> headers.put(key, value));
190160

191161
for (String jsonHeader : JsonHeaders.HEADERS) {
192162
Object value = amqpMessageProperties.getHeaders().get(jsonHeader.replaceFirst(JsonHeaders.PREFIX, ""));
@@ -228,54 +198,30 @@ protected void populateStandardHeaders(Map<String, Object> headers, MessagePrope
228198
@Override
229199
protected void populateStandardHeaders(@Nullable Map<String, Object> allHeaders, Map<String, Object> headers,
230200
MessageProperties amqpMessageProperties) {
231-
String appId = getHeaderIfAvailable(headers, AmqpHeaders.APP_ID, String.class);
232-
if (StringUtils.hasText(appId)) {
233-
amqpMessageProperties.setAppId(appId);
234-
}
235-
String clusterId = getHeaderIfAvailable(headers, AmqpHeaders.CLUSTER_ID, String.class);
236-
if (StringUtils.hasText(clusterId)) {
237-
amqpMessageProperties.setClusterId(clusterId);
238-
}
239-
String contentEncoding = getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_ENCODING, String.class);
240-
if (StringUtils.hasText(contentEncoding)) {
241-
amqpMessageProperties.setContentEncoding(contentEncoding);
242-
}
243-
Long contentLength = getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_LENGTH, Long.class);
244-
if (contentLength != null) {
245-
amqpMessageProperties.setContentLength(contentLength);
246-
}
247-
String contentType = this.extractContentTypeAsString(headers);
248201

249-
if (StringUtils.hasText(contentType)) {
250-
amqpMessageProperties.setContentType(contentType);
251-
}
252-
253-
String correlationId = getHeaderIfAvailable(headers, AmqpHeaders.CORRELATION_ID, String.class);
254-
if (StringUtils.hasText(correlationId)) {
255-
amqpMessageProperties.setCorrelationId(correlationId);
256-
}
257-
258-
Integer delay = getHeaderIfAvailable(headers, AmqpHeaders.DELAY, Integer.class);
259-
if (delay != null) {
260-
amqpMessageProperties.setDelay(delay);
261-
}
262-
MessageDeliveryMode deliveryMode = getHeaderIfAvailable(headers, AmqpHeaders.DELIVERY_MODE,
263-
MessageDeliveryMode.class);
264-
if (deliveryMode != null) {
265-
amqpMessageProperties.setDeliveryMode(deliveryMode);
266-
}
267-
Long deliveryTag = getHeaderIfAvailable(headers, AmqpHeaders.DELIVERY_TAG, Long.class);
268-
if (deliveryTag != null) {
269-
amqpMessageProperties.setDeliveryTag(deliveryTag);
270-
}
271-
String expiration = getHeaderIfAvailable(headers, AmqpHeaders.EXPIRATION, String.class);
272-
if (StringUtils.hasText(expiration)) {
273-
amqpMessageProperties.setExpiration(expiration);
274-
}
275-
Integer messageCount = getHeaderIfAvailable(headers, AmqpHeaders.MESSAGE_COUNT, Integer.class);
276-
if (messageCount != null) {
277-
amqpMessageProperties.setMessageCount(messageCount);
278-
}
202+
JavaUtils.INSTANCE
203+
.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.APP_ID, String.class),
204+
appId -> amqpMessageProperties.setAppId(appId))
205+
.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.CLUSTER_ID, String.class),
206+
clusterId -> amqpMessageProperties.setClusterId(clusterId))
207+
.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_ENCODING, String.class),
208+
contentEncoding -> amqpMessageProperties.setContentEncoding(contentEncoding))
209+
.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_LENGTH, Long.class),
210+
contentLength -> amqpMessageProperties.setContentLength(contentLength))
211+
.acceptIfHasText(this.extractContentTypeAsString(headers),
212+
contentType -> amqpMessageProperties.setContentType(contentType))
213+
.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.CORRELATION_ID, String.class),
214+
correlationId -> amqpMessageProperties.setCorrelationId(correlationId))
215+
.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.DELAY, Integer.class),
216+
delay -> amqpMessageProperties.setDelay(delay))
217+
.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.DELIVERY_MODE, MessageDeliveryMode.class),
218+
deliveryMode -> amqpMessageProperties.setDeliveryMode(deliveryMode))
219+
.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.DELIVERY_TAG, Long.class),
220+
deliveryTag -> amqpMessageProperties.setDeliveryTag(deliveryTag))
221+
.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.EXPIRATION, String.class),
222+
expiration -> amqpMessageProperties.setExpiration(expiration))
223+
.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.MESSAGE_COUNT, Integer.class),
224+
messageCount -> amqpMessageProperties.setMessageCount(messageCount));
279225
String messageId = getHeaderIfAvailable(headers, AmqpHeaders.MESSAGE_ID, String.class);
280226
if (StringUtils.hasText(messageId)) {
281227
amqpMessageProperties.setMessageId(messageId);
@@ -286,26 +232,17 @@ else if (allHeaders != null) {
286232
amqpMessageProperties.setMessageId(id.toString());
287233
}
288234
}
289-
Integer priority = getHeaderIfAvailable(headers, IntegrationMessageHeaderAccessor.PRIORITY, Integer.class);
290-
if (priority != null) {
291-
amqpMessageProperties.setPriority(priority);
292-
}
293-
String receivedExchange = getHeaderIfAvailable(headers, AmqpHeaders.RECEIVED_EXCHANGE, String.class);
294-
if (StringUtils.hasText(receivedExchange)) {
295-
amqpMessageProperties.setReceivedExchange(receivedExchange);
296-
}
297-
String receivedRoutingKey = getHeaderIfAvailable(headers, AmqpHeaders.RECEIVED_ROUTING_KEY, String.class);
298-
if (StringUtils.hasText(receivedRoutingKey)) {
299-
amqpMessageProperties.setReceivedRoutingKey(receivedRoutingKey);
300-
}
301-
Boolean redelivered = getHeaderIfAvailable(headers, AmqpHeaders.REDELIVERED, Boolean.class);
302-
if (redelivered != null) {
303-
amqpMessageProperties.setRedelivered(redelivered);
304-
}
305-
String replyTo = getHeaderIfAvailable(headers, AmqpHeaders.REPLY_TO, String.class);
306-
if (replyTo != null) {
307-
amqpMessageProperties.setReplyTo(replyTo);
308-
}
235+
JavaUtils.INSTANCE
236+
.acceptIfNotNull(getHeaderIfAvailable(headers, IntegrationMessageHeaderAccessor.PRIORITY, Integer.class),
237+
priority -> amqpMessageProperties.setPriority(priority))
238+
.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.RECEIVED_EXCHANGE, String.class),
239+
receivedExchange -> amqpMessageProperties.setReceivedExchange(receivedExchange))
240+
.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.RECEIVED_ROUTING_KEY, String.class),
241+
receivedRoutingKey -> amqpMessageProperties.setReceivedRoutingKey(receivedRoutingKey))
242+
.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.REDELIVERED, Boolean.class),
243+
redelivered -> amqpMessageProperties.setRedelivered(redelivered))
244+
.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.REPLY_TO, String.class),
245+
replyTo -> amqpMessageProperties.setReplyTo(replyTo));
309246
Date timestamp = getHeaderIfAvailable(headers, AmqpHeaders.TIMESTAMP, Date.class);
310247
if (timestamp != null) {
311248
amqpMessageProperties.setTimestamp(timestamp);
@@ -316,14 +253,11 @@ else if (allHeaders != null) {
316253
amqpMessageProperties.setTimestamp(new Date(ts));
317254
}
318255
}
319-
String type = getHeaderIfAvailable(headers, AmqpHeaders.TYPE, String.class);
320-
if (type != null) {
321-
amqpMessageProperties.setType(type);
322-
}
323-
String userId = getHeaderIfAvailable(headers, AmqpHeaders.USER_ID, String.class);
324-
if (StringUtils.hasText(userId)) {
325-
amqpMessageProperties.setUserId(userId);
326-
}
256+
JavaUtils.INSTANCE
257+
.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.TYPE, String.class),
258+
type -> amqpMessageProperties.setType(type))
259+
.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.USER_ID, String.class),
260+
userId -> amqpMessageProperties.setUserId(userId));
327261

328262
Map<String, String> jsonHeaders = new HashMap<String, String>();
329263

@@ -346,14 +280,11 @@ else if (allHeaders != null) {
346280
amqpMessageProperties.getHeaders().putAll(jsonHeaders);
347281
}
348282

349-
String replyCorrelation = getHeaderIfAvailable(headers, AmqpHeaders.SPRING_REPLY_CORRELATION, String.class);
350-
if (StringUtils.hasLength(replyCorrelation)) {
351-
amqpMessageProperties.setHeader("spring_reply_correlation", replyCorrelation);
352-
}
353-
String replyToStack = getHeaderIfAvailable(headers, AmqpHeaders.SPRING_REPLY_TO_STACK, String.class);
354-
if (StringUtils.hasLength(replyToStack)) {
355-
amqpMessageProperties.setHeader("spring_reply_to", replyToStack);
356-
}
283+
JavaUtils.INSTANCE
284+
.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.SPRING_REPLY_CORRELATION, String.class),
285+
replyCorrelation -> amqpMessageProperties.setHeader("spring_reply_correlation", replyCorrelation))
286+
.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.SPRING_REPLY_TO_STACK, String.class),
287+
replyToStack -> amqpMessageProperties.setHeader("spring_reply_to", replyToStack));
357288
}
358289

359290
@Override

‎spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java

+13-8
Original file line numberDiff line numberDiff line change
@@ -105,6 +105,18 @@ else if (payload instanceof AbstractIntegrationMessageBuilder) {
105105
*/
106106
protected Map<String, Object> aggregateHeaders(MessageGroup group) {
107107
Map<String, Object> aggregatedHeaders = new HashMap<>();
108+
Set<String> conflictKeys = doAggregateHeaders(group, aggregatedHeaders);
109+
for (String keyToRemove : conflictKeys) {
110+
if (this.logger.isDebugEnabled()) {
111+
this.logger.debug("Excluding header '" + keyToRemove + "' upon aggregation due to conflict(s) "
112+
+ "in MessageGroup with correlation key: " + group.getGroupId());
113+
}
114+
aggregatedHeaders.remove(keyToRemove);
115+
}
116+
return aggregatedHeaders;
117+
}
118+
119+
private Set<String> doAggregateHeaders(MessageGroup group, Map<String, Object> aggregatedHeaders) {
108120
Set<String> conflictKeys = new HashSet<>();
109121
for (Message<?> message : group.getMessages()) {
110122
for (Entry<String, Object> entry : message.getHeaders().entrySet()) {
@@ -126,14 +138,7 @@ protected Map<String, Object> aggregateHeaders(MessageGroup group) {
126138
}
127139
}
128140
}
129-
for (String keyToRemove : conflictKeys) {
130-
if (this.logger.isDebugEnabled()) {
131-
this.logger.debug("Excluding header '" + keyToRemove + "' upon aggregation due to conflict(s) "
132-
+ "in MessageGroup with correlation key: " + group.getGroupId());
133-
}
134-
aggregatedHeaders.remove(keyToRemove);
135-
}
136-
return aggregatedHeaders;
141+
return conflictKeys;
137142
}
138143

139144
protected abstract Object aggregatePayloads(MessageGroup group, Map<String, Object> defaultHeaders);

‎spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java

+5-3
Original file line numberDiff line numberDiff line change
@@ -643,7 +643,7 @@ protected void afterRelease(MessageGroup group, Collection<Message<?>> completed
643643
afterRelease(group, completedMessages);
644644
}
645645

646-
protected void forceComplete(MessageGroup group) {
646+
protected void forceComplete(MessageGroup group) { // NOSONAR Complexity
647647
Object correlationKey = group.getGroupId();
648648
// UUIDConverter is no-op if already converted
649649
UUID groupId = UUIDConverter.getUUID(correlationKey);
@@ -737,7 +737,7 @@ protected void forceComplete(MessageGroup group) {
737737
}
738738
}
739739
}
740-
catch (InterruptedException ie) {
740+
catch (@SuppressWarnings("unused") InterruptedException ie) {
741741
Thread.currentThread().interrupt();
742742
this.logger.debug("Thread was interrupted while trying to obtain lock");
743743
}
@@ -748,7 +748,9 @@ protected void remove(MessageGroup group) {
748748
this.messageStore.removeMessageGroup(correlationKey);
749749
}
750750

751-
protected int findLastReleasedSequenceNumber(Object groupId, Collection<Message<?>> partialSequence) {
751+
protected int findLastReleasedSequenceNumber(@SuppressWarnings("unused") Object groupId,
752+
Collection<Message<?>> partialSequence) {
753+
752754
Message<?> lastReleasedMessage = Collections.max(partialSequence, this.sequenceNumberComparator);
753755
return new IntegrationMessageHeaderAccessor(lastReleasedMessage).getSequenceNumber();
754756
}

0 commit comments

Comments
 (0)