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

INT-4349: Allow Number type for int headers

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4349

When headers come from the external system there is no guarantee
that special headers (e.g. `sequenceNumber`, `priority` etc.) in the
expected (`Integer`) type.

* Widen `int` headers setting value to the `Number` type
* Return primitive `int` for the `sequenceNumber` and `sequenceSize`
headers since for them `IntegrationMessageHeaderAccessor` never return
null for them
* Remove `SequenceNumberComparator` in favor of `MessageSequenceComparator`
since they are essentially duplicate each other
* Fix tests to deal with primitive `int` already
This commit is contained in:
Artem Bilan
2017-09-28 18:34:52 -04:00
committed by Gary Russell
parent e6225926c4
commit c65584a007
24 changed files with 176 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java
View File

@@ -89,18 +89,19 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor {
return this.getHeader(CORRELATION_ID);
}
public Integer getSequenceNumber() {
Integer sequenceNumber = this.getHeader(SEQUENCE_NUMBER, Integer.class);
return (sequenceNumber != null ? sequenceNumber : 0);
public int getSequenceNumber() {
Number sequenceNumber = this.getHeader(SEQUENCE_NUMBER, Number.class);
return (sequenceNumber != null ? sequenceNumber.intValue() : 0);
}
public Integer getSequenceSize() {
Integer sequenceSize = this.getHeader(SEQUENCE_SIZE, Integer.class);
return (sequenceSize != null ? sequenceSize : 0);
public int getSequenceSize() {
Number sequenceSize = this.getHeader(SEQUENCE_SIZE, Number.class);
return (sequenceSize != null ? sequenceSize.intValue() : 0);
}
public Integer getPriority() {
return this.getHeader(PRIORITY, Integer.class);
Number priority = this.getHeader(PRIORITY, Number.class);
return (priority != null ? priority.intValue() : null);
}
/**
@@ -140,8 +141,8 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor {
else if (IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER.equals(headerName)
|| IntegrationMessageHeaderAccessor.SEQUENCE_SIZE.equals(headerName)
|| IntegrationMessageHeaderAccessor.PRIORITY.equals(headerName)) {
Assert.isTrue(Integer.class.isAssignableFrom(headerValue.getClass()), "The '" + headerName
+ "' header value must be an Integer.");
Assert.isTrue(Number.class.isAssignableFrom(headerValue.getClass()), "The '" + headerName
+ "' header value must be a Number.");
}
else if (IntegrationMessageHeaderAccessor.ROUTING_SLIP.equals(headerName)) {
Assert.isTrue(Map.class.isAssignableFrom(headerValue.getClass()), "The '" + headerName

2
spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java
View File

@@ -90,7 +90,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
protected final Log logger = LogFactory.getLog(getClass());
private final Comparator<Message<?>> sequenceNumberComparator = new SequenceNumberComparator();
private final Comparator<Message<?>> sequenceNumberComparator = new MessageSequenceComparator();
private final Map<UUID, ScheduledFuture<?>> expireGroupScheduledFutures = new HashMap<>();

23
spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageSequenceComparator.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* Copyright 2002-2017 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,23 +22,18 @@ import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.messaging.Message;
/**
* A {@link Comparator} implementation based on the 'sequence number'
* property of a {@link Message Message's} header.
*
* @author Mark Fisher
* @author Dave Syer
* @author Artem Bilan
*/
public class MessageSequenceComparator implements Comparator<Message<?>> {
public int compare(Message<?> message1, Message<?> message2) {
Integer s1 = new IntegrationMessageHeaderAccessor(message1).getSequenceNumber();
Integer s2 = new IntegrationMessageHeaderAccessor(message2).getSequenceNumber();
if (s1 == null) {
s1 = 0;
}
if (s2 == null) {
s2 = 0;
}
return s1.compareTo(s2);
@Override
public int compare(Message<?> o1, Message<?> o2) {
int sequenceNumber1 = new IntegrationMessageHeaderAccessor(o1).getSequenceNumber();
int sequenceNumber2 = new IntegrationMessageHeaderAccessor(o2).getSequenceNumber();
return Integer.compare(sequenceNumber1, sequenceNumber2);
}
}

2
spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java
View File

@@ -36,7 +36,7 @@ import org.springframework.messaging.Message;
*/
public class ResequencingMessageGroupProcessor implements MessageGroupProcessor {
private final Comparator<Message<?>> comparator = new SequenceNumberComparator();
private final Comparator<Message<?>> comparator = new MessageSequenceComparator();
public Object processMessageGroup(MessageGroup group) {
Collection<Message<?>> messages = group.getMessages();

53
spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceNumberComparator.java
View File

@@ -1,53 +0,0 @@
/*
* Copyright 2002-2016 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aggregator;
import java.util.Comparator;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.messaging.Message;
/**
* @author Dave Syer
*
* @since 2.0
*
*/
public class SequenceNumberComparator implements Comparator<Message<?>> {
/**
* If both messages have a sequence number then compare that, otherwise if one has a sequence number and the other
* doesn't then the numbered message comes first, or finally of neither has a sequence number then they are equal in
* rank.
*/
@Override
public int compare(Message<?> o1, Message<?> o2) {
Integer sequenceNumber1 = new IntegrationMessageHeaderAccessor(o1).getSequenceNumber();
Integer sequenceNumber2 = new IntegrationMessageHeaderAccessor(o2).getSequenceNumber();
if (sequenceNumber1 == sequenceNumber2) { //NOSONAR - early exit optimization
return 0;
}
if (sequenceNumber1 == null) {
return -sequenceNumber2;
}
if (sequenceNumber2 == null) {
return sequenceNumber1;
}
return sequenceNumber1.compareTo(sequenceNumber2);
}
}

2
spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java
View File

@@ -44,7 +44,7 @@ public class SequenceSizeReleaseStrategy implements ReleaseStrategy {
private static final Log logger = LogFactory.getLog(SequenceSizeReleaseStrategy.class);
private final Comparator<Message<?>> comparator = new SequenceNumberComparator();
private final Comparator<Message<?>> comparator = new MessageSequenceComparator();
private volatile boolean releasePartialSequences;