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

INT-4369: Add DeliveryAttempt header

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

Also add `StaticMessageHeaderAccessor`

- avoids object creation when getting type-safe well-known headers.

* Docs
This commit is contained in:
Gary Russell
2018-01-02 16:17:32 -05:00
committed by Artem Bilan
parent 381d3f9483
commit f21f81092d
9 changed files with 245 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java
View File

@@ -17,6 +17,7 @@
package org.springframework.integration.amqp.inbound;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
@@ -27,6 +28,7 @@ import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.core.AttributeAccessor;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.amqp.support.AmqpHeaderMapper;
import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
@@ -34,6 +36,7 @@ import org.springframework.integration.context.OrderlyShutdownCapable;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageUtils;
import org.springframework.integration.support.StaticMessageHeaderAccessor;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
@@ -200,15 +203,18 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
try {
if (AmqpInboundChannelAdapter.this.retryTemplate == null) {
try {
processMessage(message, channel);
createAndSend(message, channel);
}
finally {
attributesHolder.remove();
}
}
else {
final org.springframework.messaging.Message<Object> toSend = createMessage(message, channel);
AmqpInboundChannelAdapter.this.retryTemplate.execute(context -> {
processMessage(message, channel);
StaticMessageHeaderAccessor.getDeliveryAttempt(toSend).incrementAndGet();
setAttributesIfNecessary(message, toSend);
sendMessage(toSend);
return null;
},
(RecoveryCallback<Object>) AmqpInboundChannelAdapter.this.recoveryCallback);
@@ -225,7 +231,13 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
}
}
private void processMessage(Message message, Channel channel) {
private void createAndSend(Message message, Channel channel) {
org.springframework.messaging.Message<Object> messagingMessage = createMessage(message, channel);
setAttributesIfNecessary(message, messagingMessage);
sendMessage(messagingMessage);
}
private org.springframework.messaging.Message<Object> createMessage(Message message, Channel channel) {
Object payload = AmqpInboundChannelAdapter.this.messageConverter.fromMessage(message);
Map<String, Object> headers = AmqpInboundChannelAdapter.this.headerMapper
.toHeadersFromRequest(message.getMessageProperties());
@@ -234,12 +246,14 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
headers.put(AmqpHeaders.DELIVERY_TAG, message.getMessageProperties().getDeliveryTag());
headers.put(AmqpHeaders.CHANNEL, channel);
}
if (AmqpInboundChannelAdapter.this.retryTemplate != null) {
headers.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, new AtomicInteger());
}
final org.springframework.messaging.Message<Object> messagingMessage = getMessageBuilderFactory()
.withPayload(payload)
.copyHeaders(headers)
.build();
setAttributesIfNecessary(message, messagingMessage);
sendMessage(messagingMessage);
return messagingMessage;
}
@Override

118
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java
View File

@@ -17,6 +17,7 @@
package org.springframework.integration.amqp.inbound;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Address;
@@ -32,12 +33,14 @@ import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.core.AttributeAccessor;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.amqp.support.AmqpHeaderMapper;
import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageUtils;
import org.springframework.integration.support.StaticMessageHeaderAccessor;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
@@ -255,23 +258,29 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
public void onMessage(final Message message, final Channel channel) throws Exception {
if (AmqpInboundGateway.this.retryTemplate == null) {
try {
doOnMessage(message, channel);
org.springframework.messaging.Message<Object> converted = convert(message, channel);
if (converted != null) {
process(message, converted);
}
}
finally {
attributesHolder.remove();
}
}
else {
AmqpInboundGateway.this.retryTemplate.execute(context -> {
doOnMessage(message, channel);
org.springframework.messaging.Message<Object> converted = convert(message, channel);
if (converted != null) {
AmqpInboundGateway.this.retryTemplate.execute(context -> {
StaticMessageHeaderAccessor.getDeliveryAttempt(converted).incrementAndGet();
process(message, converted);
return null;
},
(RecoveryCallback<Object>) AmqpInboundGateway.this.recoveryCallback);
}
}
}
private void doOnMessage(Message message, Channel channel) {
boolean error = false;
private org.springframework.messaging.Message<Object> convert(Message message, Channel channel) {
Map<String, Object> headers = null;
Object payload = null;
try {
@@ -281,6 +290,9 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
headers.put(AmqpHeaders.DELIVERY_TAG, message.getMessageProperties().getDeliveryTag());
headers.put(AmqpHeaders.CHANNEL, channel);
}
if (AmqpInboundGateway.this.retryTemplate != null) {
headers.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, new AtomicInteger());
}
}
catch (RuntimeException e) {
if (getErrorChannel() != null) {
@@ -290,60 +302,60 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
else {
throw e;
}
error = true;
return null;
}
if (!error) {
org.springframework.messaging.Message<Object> messagingMessage = getMessageBuilderFactory()
return getMessageBuilderFactory()
.withPayload(payload)
.copyHeaders(headers)
.build();
setAttributesIfNecessary(message, messagingMessage);
final org.springframework.messaging.Message<?> reply = sendAndReceiveMessage(messagingMessage);
if (reply != null) {
Address replyTo;
String replyToProperty = message.getMessageProperties().getReplyTo();
if (replyToProperty != null) {
replyTo = new Address(replyToProperty);
}
private void process(Message message, org.springframework.messaging.Message<Object> messagingMessage) {
setAttributesIfNecessary(message, messagingMessage);
final org.springframework.messaging.Message<?> reply = sendAndReceiveMessage(messagingMessage);
if (reply != null) {
Address replyTo;
String replyToProperty = message.getMessageProperties().getReplyTo();
if (replyToProperty != null) {
replyTo = new Address(replyToProperty);
}
else {
replyTo = AmqpInboundGateway.this.defaultReplyTo;
}
MessagePostProcessor messagePostProcessor =
message1 -> {
MessageProperties messageProperties = message1.getMessageProperties();
String contentEncoding = messageProperties.getContentEncoding();
long contentLength = messageProperties.getContentLength();
String contentType = messageProperties.getContentType();
AmqpInboundGateway.this.headerMapper.fromHeadersToReply(reply.getHeaders(),
messageProperties);
// clear the replyTo from the original message since we are using it now
messageProperties.setReplyTo(null);
// reset the content-* properties as determined by the MessageConverter
if (StringUtils.hasText(contentEncoding)) {
messageProperties.setContentEncoding(contentEncoding);
}
messageProperties.setContentLength(contentLength);
if (contentType != null) {
messageProperties.setContentType(contentType);
}
return message1;
};
if (replyTo != null) {
AmqpInboundGateway.this.amqpTemplate.convertAndSend(replyTo.getExchangeName(),
replyTo.getRoutingKey(), reply.getPayload(), messagePostProcessor);
}
else {
if (!AmqpInboundGateway.this.amqpTemplateExplicitlySet) {
throw new IllegalStateException("There is no 'replyTo' message property " +
"and the `defaultReplyTo` hasn't been configured.");
}
else {
replyTo = AmqpInboundGateway.this.defaultReplyTo;
}
MessagePostProcessor messagePostProcessor =
message1 -> {
MessageProperties messageProperties = message1.getMessageProperties();
String contentEncoding = messageProperties.getContentEncoding();
long contentLength = messageProperties.getContentLength();
String contentType = messageProperties.getContentType();
AmqpInboundGateway.this.headerMapper.fromHeadersToReply(reply.getHeaders(),
messageProperties);
// clear the replyTo from the original message since we are using it now
messageProperties.setReplyTo(null);
// reset the content-* properties as determined by the MessageConverter
if (StringUtils.hasText(contentEncoding)) {
messageProperties.setContentEncoding(contentEncoding);
}
messageProperties.setContentLength(contentLength);
if (contentType != null) {
messageProperties.setContentType(contentType);
}
return message1;
};
if (replyTo != null) {
AmqpInboundGateway.this.amqpTemplate.convertAndSend(replyTo.getExchangeName(),
replyTo.getRoutingKey(), reply.getPayload(), messagePostProcessor);
}
else {
if (!AmqpInboundGateway.this.amqpTemplateExplicitlySet) {
throw new IllegalStateException("There is no 'replyTo' message property " +
"and the `defaultReplyTo` hasn't been configured.");
}
else {
AmqpInboundGateway.this.amqpTemplate.convertAndSend(reply.getPayload(),
messagePostProcessor);
}
AmqpInboundGateway.this.amqpTemplate.convertAndSend(reply.getPayload(),
messagePostProcessor);
}
}
}

14
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java
View File

@@ -19,6 +19,7 @@ package org.springframework.integration.amqp.inbound;
import static org.hamcrest.CoreMatchers.containsString;
import static org.hamcrest.CoreMatchers.instanceOf;
import static org.hamcrest.CoreMatchers.notNullValue;
import static org.hamcrest.Matchers.equalTo;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
@@ -66,6 +67,7 @@ import org.springframework.integration.json.JsonToObjectTransformer;
import org.springframework.integration.json.ObjectToJsonTransformer;
import org.springframework.integration.mapping.support.JsonHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.StaticMessageHeaderAccessor;
import org.springframework.integration.transformer.MessageTransformingHandler;
import org.springframework.integration.transformer.Transformer;
import org.springframework.messaging.Message;
@@ -307,9 +309,11 @@ public class InboundEndpointTests {
Message<?> errorMessage = errors.receive(0);
assertNotNull(errorMessage);
assertThat(errorMessage.getPayload(), instanceOf(MessagingException.class));
assertThat(((MessagingException) errorMessage.getPayload()).getMessage(), containsString("Dispatcher has no"));
MessagingException payload = (MessagingException) errorMessage.getPayload();
assertThat(payload.getMessage(), containsString("Dispatcher has no"));
assertThat(StaticMessageHeaderAccessor.getDeliveryAttempt(payload.getFailedMessage()).get(), equalTo(3));
org.springframework.amqp.core.Message amqpMessage = errorMessage.getHeaders()
.get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE, org.springframework.amqp.core.Message.class);
.get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE, org.springframework.amqp.core.Message.class);
assertThat(amqpMessage, notNullValue());
assertNull(errors.receive(0));
}
@@ -332,9 +336,11 @@ public class InboundEndpointTests {
Message<?> errorMessage = errors.receive(0);
assertNotNull(errorMessage);
assertThat(errorMessage.getPayload(), instanceOf(MessagingException.class));
assertThat(((MessagingException) errorMessage.getPayload()).getMessage(), containsString("Dispatcher has no"));
MessagingException payload = (MessagingException) errorMessage.getPayload();
assertThat(payload.getMessage(), containsString("Dispatcher has no"));
assertThat(StaticMessageHeaderAccessor.getDeliveryAttempt(payload.getFailedMessage()).get(), equalTo(3));
org.springframework.amqp.core.Message amqpMessage = errorMessage.getHeaders()
.get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE, org.springframework.amqp.core.Message.class);
.get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE, org.springframework.amqp.core.Message.class);
assertThat(amqpMessage, notNullValue());
assertNull(errors.receive(0));
}

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

@@ -22,7 +22,9 @@ import java.util.Date;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageHeaderAccessor;
@@ -35,7 +37,7 @@ import org.springframework.util.ObjectUtils;
*
* @author Andy Wilkinson
* @author Artem Bilan
* @author Gary Russel
* @author Gary Russell
*
* @since 4.0
*
@@ -60,6 +62,8 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor {
public static final String CLOSEABLE_RESOURCE = "closeableResource";
public static final String DELIVERY_ATTEMPT = "deliveryAttempt";
private Set<String> readOnlyHeaders = new HashSet<String>();
public IntegrationMessageHeaderAccessor(Message<?> message) {
@@ -81,10 +85,12 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor {
}
}
@Nullable
public Long getExpirationDate() {
return this.getHeader(EXPIRATION_DATE, Long.class);
}
@Nullable
public Object getCorrelationId() {
return this.getHeader(CORRELATION_ID);
}
@@ -99,6 +105,7 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor {
return (sequenceSize != null ? sequenceSize.intValue() : 0);
}
@Nullable
public Integer getPriority() {
Number priority = this.getHeader(PRIORITY, Number.class);
return (priority != null ? priority.intValue() : null);
@@ -113,11 +120,24 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor {
* @return the {@link Closeable}.
* @since 4.3
*/
@Nullable
public Closeable getCloseableResource() {
return this.getHeader(CLOSEABLE_RESOURCE, Closeable.class);
}
/**
* When a message-driven enpoint supports retry implicitly, this
* header is incremented for each delivery attempt.
* @return the delivery attempt.
* @since 5.0.1
*/
@Nullable
public AtomicInteger getDeliveryAttempt() {
return this.getHeader(DELIVERY_ATTEMPT, AtomicInteger.class);
}
@SuppressWarnings("unchecked")
@Nullable
public <T> T getHeader(String key, Class<T> type) {
Object value = getHeader(key);
if (value == null) {

103
spring-integration-core/src/main/java/org/springframework/integration/support/StaticMessageHeaderAccessor.java Normal file
View File

@@ -0,0 +1,103 @@
/*
* Copyright 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.
* 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.support;
import java.io.Closeable;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.util.MimeType;
/**
* Lightweight type-safe header accessor avoiding object
* creation just to access a header.
*
* @author Gary Russell
* @since 5.0.1
*
* @see IntegrationMessageHeaderAccessor
*/
public final class StaticMessageHeaderAccessor {
private StaticMessageHeaderAccessor() {
super();
}
@Nullable
public static UUID getId(Message<?> message) {
Object value = message.getHeaders().get(MessageHeaders.ID);
if (value == null) {
return null;
}
return (value instanceof UUID ? (UUID) value : UUID.fromString(value.toString()));
}
@Nullable
public static Long getTimestamp(Message<?> message) {
Object value = message.getHeaders().get(MessageHeaders.TIMESTAMP);
if (value == null) {
return null;
}
return (value instanceof Long ? (Long) value : Long.parseLong(value.toString()));
}
@Nullable
public static MimeType getContentType(Message<?> message) {
Object value = message.getHeaders().get(MessageHeaders.CONTENT_TYPE);
if (value == null) {
return null;
}
return (value instanceof MimeType ? (MimeType) value : MimeType.valueOf(value.toString()));
}
@Nullable
public static Long getExpirationDate(Message<?> message) {
return message.getHeaders().get(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, Long.class);
}
public static int getSequenceNumber(Message<?> message) {
Number sequenceNumber = message.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER,
Number.class);
return (sequenceNumber != null ? sequenceNumber.intValue() : 0);
}
public static int getSequenceSize(Message<?> message) {
Number sequenceSize = message.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, Number.class);
return (sequenceSize != null ? sequenceSize.intValue() : 0);
}
@Nullable
public static Integer getPriority(Message<?> message) {
Number priority = message.getHeaders().get(IntegrationMessageHeaderAccessor.PRIORITY, Number.class);
return (priority != null ? priority.intValue() : null);
}
@Nullable
public static Closeable getCloseableResource(Message<?> message) {
return message.getHeaders().get(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, Closeable.class);
}
@Nullable
public static AtomicInteger getDeliveryAttempt(Message<?> message) {
return message.getHeaders().get(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, AtomicInteger.class);
}
}

4
spring-integration-core/src/main/java/org/springframework/integration/transformer/StreamTransformer.java
View File

@@ -20,7 +20,7 @@ import java.io.ByteArrayOutputStream;
import java.io.Closeable;
import java.io.InputStream;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.support.StaticMessageHeaderAccessor;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import org.springframework.util.FileCopyUtils;
@@ -60,7 +60,7 @@ public class StreamTransformer extends AbstractTransformer {
InputStream stream = (InputStream) message.getPayload();
ByteArrayOutputStream baos = new ByteArrayOutputStream();
FileCopyUtils.copy(stream, baos);
Closeable closeableResource = new IntegrationMessageHeaderAccessor(message).getCloseableResource();
Closeable closeableResource = StaticMessageHeaderAccessor.getCloseableResource(message);
if (closeableResource != null) {
closeableResource.close();
}

5
spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java
View File

@@ -34,11 +34,11 @@ import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.file.FileHeaders;
import org.springframework.integration.file.splitter.FileSplitter.FileMarker.Mark;
import org.springframework.integration.splitter.AbstractMessageSplitter;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.support.StaticMessageHeaderAccessor;
import org.springframework.integration.support.json.JsonObjectMapper;
import org.springframework.integration.support.json.JsonObjectMapperProvider;
import org.springframework.messaging.Message;
@@ -220,8 +220,7 @@ public class FileSplitter extends AbstractMessageSplitter {
super.close();
}
finally {
Closeable closeableResource = new IntegrationMessageHeaderAccessor(message)
.getCloseableResource();
Closeable closeableResource = StaticMessageHeaderAccessor.getCloseableResource(message);
if (closeableResource != null) {
closeableResource.close();
}

7
spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java
View File

@@ -47,6 +47,7 @@ import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
import org.springframework.integration.file.remote.session.Session;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.file.splitter.FileSplitter;
import org.springframework.integration.support.StaticMessageHeaderAccessor;
import org.springframework.integration.transformer.StreamTransformer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
@@ -85,7 +86,7 @@ public class StreamingInboundTests {
assertThat(fileInfo, containsString("link\":false"));
// close after list, transform
verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource(), times(2)).close();
verify(StaticMessageHeaderAccessor.getCloseableResource(received), times(2)).close();
received = (Message<byte[]>) this.transformer.transform(streamer.receive());
assertEquals("baz\nqux", new String(received.getPayload()));
@@ -101,7 +102,7 @@ public class StreamingInboundTests {
assertThat(fileInfo, containsString("link\":false"));
// close after transform
verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource(), times(3)).close();
verify(StaticMessageHeaderAccessor.getCloseableResource(received), times(3)).close();
verify(sessionFactory.getSession()).list("/foo");
}
@@ -122,7 +123,7 @@ public class StreamingInboundTests {
assertEquals("foo", received.getHeaders().get(FileHeaders.REMOTE_FILE));
// close after list, transform
verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource(), times(2)).close();
verify(StaticMessageHeaderAccessor.getCloseableResource(received), times(2)).close();
received = (Message<byte[]>) this.transformer.transform(streamer.receive());
assertEquals("baz\nqux", new String(received.getPayload()));

60
src/reference/asciidoc/message.adoc
View File

@@ -167,11 +167,6 @@ All (core) Spring Integration specific Message Headers constants are now declare
[cols="5l,3l,5", options="header"]
|===
| Header Name
@@ -181,12 +176,6 @@ All (core) Spring Integration specific Message Headers constants are now declare
| Usage
| IntegrationMessageHeaderAccessor.
CORRELATION_ID
@@ -197,12 +186,6 @@ All (core) Spring Integration specific Message Headers constants are now declare
| Used to correlate two or more messages.
| IntegrationMessageHeaderAccessor.
SEQUENCE_NUMBER
@@ -213,12 +196,6 @@ All (core) Spring Integration specific Message Headers constants are now declare
| Usually a sequence number with a group of messages with a `SEQUENCE_SIZE` but can also be used in a `<resequencer/>` to resequence an unbounded group of messages.
| IntegrationMessageHeaderAccessor.
SEQUENCE_SIZE
@@ -229,12 +206,6 @@ All (core) Spring Integration specific Message Headers constants are now declare
| The number of messages within a group of correlated messages.
| IntegrationMessageHeaderAccessor.
EXPIRATION_DATE
@@ -246,12 +217,6 @@ All (core) Spring Integration specific Message Headers constants are now declare
Not used by the framework directly but can be set with a header enricher and used in a `<filter/>` configured with an `UnexpiredMessageSelector`.
| IntegrationMessageHeaderAccessor.
PRIORITY
@@ -262,12 +227,6 @@ Not used by the framework directly but can be set with a header enricher and use
| Message priority; for example within a `PriorityChannel`
| IntegrationMessageHeaderAccessor.
DUPLICATE_MESSAGE
@@ -278,6 +237,25 @@ Not used by the framework directly but can be set with a header enricher and use
| True if a message was detected as a duplicate by an idempotent receiver interceptor.
See <<idempotent-receiver>>.
| IntegrationMessageHeaderAccessor.
CLOSEABLE_RESOURCE
| java.io.Closeable
| This header is present if the message is associated with a `Closeable` which should be closed when message processing is complete.
An example is the `Session` associated with a streamed file transfer using FTP, SFTP, etc.
| IntegrationMessageHeaderAccessor.
DELIVERY_ATTEMPT
| java.lang.AtomicInteger
| If a message-driven channel adapter supports the configuration of a `RetryTemplate` this header contains the current delivery attempt.
|===
Convenient typed getters for some of these headers are provided on the `IntegrationMessageHeaderAccessor` class: