diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java index e78eb29cb0..c50c672a2d 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java @@ -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 toSend = createMessage(message, channel); AmqpInboundChannelAdapter.this.retryTemplate.execute(context -> { - processMessage(message, channel); + StaticMessageHeaderAccessor.getDeliveryAttempt(toSend).incrementAndGet(); + setAttributesIfNecessary(message, toSend); + sendMessage(toSend); return null; }, (RecoveryCallback) 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 messagingMessage = createMessage(message, channel); + setAttributesIfNecessary(message, messagingMessage); + sendMessage(messagingMessage); + } + + private org.springframework.messaging.Message createMessage(Message message, Channel channel) { Object payload = AmqpInboundChannelAdapter.this.messageConverter.fromMessage(message); Map 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 messagingMessage = getMessageBuilderFactory() .withPayload(payload) .copyHeaders(headers) .build(); - setAttributesIfNecessary(message, messagingMessage); - sendMessage(messagingMessage); + return messagingMessage; } @Override diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java index 0397fa4cc0..2a958d93fd 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java @@ -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 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 converted = convert(message, channel); + if (converted != null) { + AmqpInboundGateway.this.retryTemplate.execute(context -> { + StaticMessageHeaderAccessor.getDeliveryAttempt(converted).incrementAndGet(); + process(message, converted); return null; }, (RecoveryCallback) AmqpInboundGateway.this.recoveryCallback); + } } } - private void doOnMessage(Message message, Channel channel) { - boolean error = false; + private org.springframework.messaging.Message convert(Message message, Channel channel) { Map 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 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 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); } } } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java index 25f5b29e67..639fe22a4f 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java @@ -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)); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java b/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java index d9153bba0b..78ca11f599 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java @@ -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 readOnlyHeaders = new HashSet(); 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 getHeader(String key, Class type) { Object value = getHeader(key); if (value == null) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/StaticMessageHeaderAccessor.java b/spring-integration-core/src/main/java/org/springframework/integration/support/StaticMessageHeaderAccessor.java new file mode 100644 index 0000000000..1bf23b4534 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/StaticMessageHeaderAccessor.java @@ -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); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/StreamTransformer.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/StreamTransformer.java index 24196313ea..f59e9224f4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/StreamTransformer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/StreamTransformer.java @@ -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(); } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java index bf23b85b78..6d75092d80 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/splitter/FileSplitter.java @@ -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(); } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java index cfc5bd13ca..f068ae305c 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java @@ -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) 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) this.transformer.transform(streamer.receive()); assertEquals("baz\nqux", new String(received.getPayload())); diff --git a/src/reference/asciidoc/message.adoc b/src/reference/asciidoc/message.adoc index 670a88f5fe..8f8ea545a6 100644 --- a/src/reference/asciidoc/message.adoc +++ b/src/reference/asciidoc/message.adoc @@ -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 `` 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 `` 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 <>. +| 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: