From 1c051416ce75e69caa202d1c9153a7f77588a87b Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Sat, 2 Aug 2014 22:20:10 +0300 Subject: [PATCH] INT-3489 Reschedule Aggregator's group-timeout JIRA: https://jira.spring.io/browse/INT-3489 * Catch `MessageDeliveryException` and reschedule `group-time out task`: `AbstractCorrelatingMessageHandler#scheduleGroupToForceComplete` * Apply `send-timeout` for the `discardChannel` * Mark `groupRemove = false` in case of `MessageDeliveryException` * Improve `send-timeout` docs INT-3489: Recalculate `groupTimeout` on each rescheduling Add Test for Zero Timeout --- .../AbstractCorrelatingMessageHandler.java | 120 +++++++++++------- .../router/RecipientListRouterManagement.java | 2 + .../config/xml/spring-integration-4.1.xsd | 6 +- ...elatingMessageHandlerIntegrationTests.java | 2 + .../AggregatorIntegrationTests-context.xml | 14 ++ .../AggregatorIntegrationTests.java | 61 ++++++++- .../json/ObjectToJsonTransformerTests.java | 26 ++-- src/reference/docbook/aggregator.xml | 11 +- src/reference/docbook/resequencer.xml | 11 +- 9 files changed, 190 insertions(+), 63 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index 54b337124a..19a4233158 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -50,6 +50,7 @@ import org.springframework.integration.support.locks.LockRegistry; import org.springframework.integration.util.UUIDConverter; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessagingException; import org.springframework.messaging.core.DestinationResolutionException; import org.springframework.scheduling.TaskScheduler; @@ -122,14 +123,15 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP private volatile boolean expireGroupsUponTimeout = true; public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store, - CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) { + CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) { Assert.notNull(processor); Assert.notNull(store); setMessageStore(store); this.outputProcessor = processor; - this.correlationStrategy = correlationStrategy == null ? - new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID) : correlationStrategy; + this.correlationStrategy = (correlationStrategy == null + ? new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID) + : correlationStrategy); this.releaseStrategy = releaseStrategy == null ? new SequenceSizeReleaseStrategy() : releaseStrategy; setSendTimeout(DEFAULT_SEND_TIMEOUT); sequenceAware = this.releaseStrategy instanceof SequenceSizeReleaseStrategy; @@ -214,9 +216,9 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP if (this.releasePartialSequences) { Assert.isInstanceOf(SequenceSizeReleaseStrategy.class, this.releaseStrategy, - "Release strategy of type [" + this.releaseStrategy.getClass().getSimpleName() - + "] cannot release partial sequences. Use the default SequenceSizeReleaseStrategy instead."); - ((SequenceSizeReleaseStrategy)this.releaseStrategy).setReleasePartialSequences(releasePartialSequences); + "Release strategy of type [" + this.releaseStrategy.getClass().getSimpleName() + + "] cannot release partial sequences. Use the default SequenceSizeReleaseStrategy instead."); + ((SequenceSizeReleaseStrategy) this.releaseStrategy).setReleasePartialSequences(releasePartialSequences); } /* @@ -339,7 +341,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP @Override protected void handleMessageInternal(Message message) throws Exception { Object correlationKey = correlationStrategy.getCorrelationKey(message); - Assert.state(correlationKey!=null, "Null correlation not allowed. Maybe the CorrelationStrategy is failing?"); + Assert.state(correlationKey != null, "Null correlation not allowed. Maybe the CorrelationStrategy is failing?"); if (logger.isDebugEnabled()) { logger.debug("Handling message with correlationKey [" + correlationKey + "]: " + message); @@ -354,11 +356,12 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP if (scheduledFuture != null) { boolean canceled = scheduledFuture.cancel(true); if (canceled && logger.isDebugEnabled()) { - logger.debug("Cancel 'forceComplete' scheduling for MessageGroup with Correlation Key [ " + correlationKey + "]."); + logger.debug("Cancel 'forceComplete' scheduling for MessageGroup with Correlation Key [ " + + correlationKey + "]."); } } MessageGroup messageGroup = messageStore.getMessageGroup(correlationKey); - if (this.sequenceAware){ + if (this.sequenceAware) { messageGroup = new SequenceAwareMessageGroup(messageGroup); } @@ -380,29 +383,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } } else { - Long groupTimeout = this.obtainGroupTimeout(messageGroup); - if (groupTimeout != null && groupTimeout >= 0) { - if (groupTimeout > 0) { - final MessageGroup messageGroupToSchedule = messageGroup; - - scheduledFuture = this.getTaskScheduler() - .schedule(new Runnable() { - - @Override - public void run() { - AbstractCorrelatingMessageHandler.this.forceComplete(messageGroupToSchedule); - } - }, new Date(System.currentTimeMillis() + groupTimeout)); - - if (logger.isDebugEnabled()) { - logger.debug("Schedule MessageGroup [ " + messageGroup + "] to 'forceComplete'."); - } - this.expireGroupScheduledFutures.put(groupIdUuid, scheduledFuture); - } - else { - this.forceComplete(messageGroup); - } - } + scheduleGroupToForceComplete(messageGroup); } } else { @@ -414,6 +395,43 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } } + private void scheduleGroupToForceComplete(final MessageGroup messageGroup) { + final Long groupTimeout = this.obtainGroupTimeout(messageGroup); + /* + * When 'groupTimeout' is evaluated to 'null' we do nothing. + * The 'MessageGroupStoreReaper' can be used to 'forceComplete' message groups. + */ + if (groupTimeout != null && groupTimeout >= 0) { + if (groupTimeout > 0) { + ScheduledFuture scheduledFuture = this.getTaskScheduler() + .schedule(new Runnable() { + + @Override + public void run() { + try { + AbstractCorrelatingMessageHandler.this.forceComplete(messageGroup); + } + catch (MessageDeliveryException e) { + if (logger.isDebugEnabled()) { + logger.debug("The MessageGroup [ " + messageGroup + + "] is rescheduled by the reason: " + e.getMessage()); + } + scheduleGroupToForceComplete(messageGroup); + } + } + }, new Date(System.currentTimeMillis() + groupTimeout)); + + if (logger.isDebugEnabled()) { + logger.debug("Schedule MessageGroup [ " + messageGroup + "] to 'forceComplete'."); + } + this.expireGroupScheduledFutures.put(UUIDConverter.getUUID(messageGroup.getGroupId()), scheduledFuture); + } + else { + forceComplete(messageGroup); + } + } + } + private void discardMessage(Message message) { if (this.discardChannelName != null) { synchronized (this) { @@ -429,7 +447,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } } } - this.discardChannel.send(message); + this.messagingTemplate.send(this.discardChannel, message); } /** @@ -448,7 +466,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP try { lock.lockInterruptibly(); try { - ScheduledFuture scheduledFuture = this.expireGroupScheduledFutures.remove(UUIDConverter.getUUID(correlationKey)); + ScheduledFuture scheduledFuture = + this.expireGroupScheduledFutures.remove(UUIDConverter.getUUID(correlationKey)); if (scheduledFuture != null) { boolean canceled = scheduledFuture.cancel(false); if (canceled && logger.isDebugEnabled()) { @@ -512,6 +531,15 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } } } + catch (MessageDeliveryException e) { + removeGroup = false; + if (logger.isDebugEnabled()) { + logger.debug("Group expiry candidate (" + correlationKey + + ") has been affected by MessageDeliveryException - " + + "it may be reconsidered for a future expiration one more time"); + } + throw e; + } finally { try { if (removeGroup) { @@ -534,7 +562,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP messageStore.removeMessageGroup(correlationKey); } - protected int findLastReleasedSequenceNumber(Object groupId, Collection> partialSequence){ + protected int findLastReleasedSequenceNumber(Object groupId, Collection> partialSequence) { List> sorted = new ArrayList>(partialSequence); Collections.sort(sorted, new SequenceNumberComparator()); @@ -570,7 +598,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } if (this.applicationEventPublisher != null) { this.applicationEventPublisher.publishEvent(new MessageGroupExpiredEvent(this, correlationKey, group - .size(), new Date(group.getLastModified()) , new Date(), !sendPartialResultOnExpiry)); + .size(), new Date(group.getLastModified()), new Date(), !sendPartialResultOnExpiry)); } } @@ -598,9 +626,10 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP return partialSequence; } - protected void verifyResultCollectionConsistsOfMessages(Collection elements){ + protected void verifyResultCollectionConsistsOfMessages(Collection elements) { Class commonElementType = CollectionUtils.findCommonElementType(elements); - Assert.isAssignable(Message.class, commonElementType, "The expected collection of Messages contains non-Message element: " + commonElementType); + Assert.isAssignable(Message.class, commonElementType, + "The expected collection of Messages contains non-Message element: " + commonElementType); } @SuppressWarnings("rawtypes") @@ -619,7 +648,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP for (Object next : (Iterable) processorResult) { this.sendReplyMessage(next, replyChannel); } - } else { + } + else { this.sendReplyMessage(processorResult, replyChannel); } } @@ -628,16 +658,20 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP if (replyChannel instanceof MessageChannel) { if (reply instanceof Message) { this.messagingTemplate.send((MessageChannel) replyChannel, (Message) reply); - } else { + } + else { this.messagingTemplate.convertAndSend((MessageChannel) replyChannel, reply); } - } else if (replyChannel instanceof String) { + } + else if (replyChannel instanceof String) { if (reply instanceof Message) { this.messagingTemplate.send((String) replyChannel, (Message) reply); - } else { + } + else { this.messagingTemplate.convertAndSend((String) replyChannel, reply); } - } else { + } + else { throw new MessagingException("replyChannel must be a MessageChannel or String"); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouterManagement.java b/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouterManagement.java index d227c07a1a..c231d364fc 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouterManagement.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouterManagement.java @@ -53,6 +53,7 @@ public interface RecipientListRouterManagement { /** * Remove all recipients that match the channelName. * @param channelName The channel name. + * @return The number of recipients removed. */ @ManagedOperation int removeRecipient(String channelName); @@ -61,6 +62,7 @@ public interface RecipientListRouterManagement { * Remove all recipients that match the channelName and expression. * @param channelName The channel name. * @param selectorExpression The expression to filter the incoming message + * @return The number of recipients removed. */ @ManagedOperation int removeRecipient(String channelName, String selectorExpression); diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd index 918e762f64..185f33d817 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd @@ -4295,8 +4295,10 @@ The list of component name patterns you want to track (e.g., tracked-components Specify the maximum amount of time in milliseconds to wait when sending a reply - Message to the - output channel. By default the send will block for one second. + Message to the output channel. By default the send will block for one second. + It is applied only if the output channel has some 'sending' limitations, e.g. QueueChannel with + fixed a 'capacity'. In this case a MessageDeliveryException is thrown. The 'send-timeout' + is ignored in case of AbstractSubscribableChannel implementations. diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTests.java index f0939566ba..280d137cb0 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTests.java @@ -16,6 +16,7 @@ package org.springframework.integration.aggregator; +import static org.mockito.Matchers.any; import static org.mockito.Matchers.isA; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -63,6 +64,7 @@ public class CorrelatingMessageHandlerIntegrationTests { public void completesAfterThreshold() throws Exception { defaultHandler.setReleaseStrategy(new MessageCountReleaseStrategy()); MessageChannel discardChannel = mock(MessageChannel.class); + when(discardChannel.send(any(Message.class))).thenReturn(true); defaultHandler.setDiscardChannel(discardChannel); Message message1 = correlatedMessage(1, 2, 1); Message message2 = correlatedMessage(1, 2, 2); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml index 6a94a2714f..0492fd04c1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml @@ -37,6 +37,7 @@ + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java index d3db14d031..51a5f54a81 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java @@ -16,15 +16,19 @@ package org.springframework.integration.aggregator.integration; +import static org.hamcrest.Matchers.containsString; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import java.util.Collection; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Queue; import org.junit.Test; import org.junit.runner.RunWith; @@ -32,11 +36,13 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -71,11 +77,17 @@ public class AggregatorIntegrationTests { private MessageChannel groupTimeoutExpressionAggregatorInput; @Autowired - private PollableChannel output; + private MessageChannel zeroGroupTimeoutExpressionAggregatorInput; + + @Autowired + private QueueChannel output; @Autowired private PollableChannel discard; + @Autowired + private QueueChannel errors; + @Test//(timeout=5000) public void testVanillaAggregation() throws Exception { for (int i = 0; i < 5; i++) { @@ -147,6 +159,31 @@ public class AggregatorIntegrationTests { } } + @Test + public void testGroupTimeoutReschedulingOnMessageDeliveryException() throws Exception { + for (int i = 0; i < 5; i++) { + this.output.send(new GenericMessage("fake message")); + } + + Map headers = stubHeaders(1, 2, 1); + this.groupTimeoutAggregatorInput.send(new GenericMessage(1, headers)); + + //Wait until 'group-timeout' does its stuff. + MessageGroupStore mgs = TestUtils.getPropertyValue(this.context.getBean("gta.handler"), "messageStore", + MessageGroupStore.class); + int n = 0; + while (n++ < 100 && mgs.getMessageGroupCount() > 0) { + Thread.sleep(100); + if (n == 10) { + TestUtils.getPropertyValue(this.output, "queue", Queue.class).clear(); + } + } + assertTrue("Group did not complete", n < 100); + Message receive = this.output.receive(1000); + assertNotNull(receive); + assertEquals(Collections.singletonList(1), receive.getPayload()); + assertNull(this.discard.receive(0)); + } @Test public void testGroupTimeoutExpressionScheduling() throws Exception { @@ -187,6 +224,28 @@ public class AggregatorIntegrationTests { assertNull(this.discard.receive(0)); } + @Test + public void testZeroGroupTimeoutExpressionScheduling() throws Exception { + try { + this.output.purge(null); + this.errors.purge(null); + GenericMessage message = new GenericMessage("foo"); + this.output.send(message); + this.output.send(message); + this.output.send(message); + this.output.send(message); + this.output.send(message); + this.zeroGroupTimeoutExpressionAggregatorInput.send(new GenericMessage(1, stubHeaders(1, 2, 1))); + ErrorMessage em = (ErrorMessage) this.errors.receive(10000); + assertNotNull(em); + assertThat(em.getPayload().getMessage(), + containsString("failed to send message to channel 'output' within timeout: 10")); + } + finally { + this.output.purge(null); + this.errors.purge(null); + } + } // configured in context associated with this test public static class SummingAggregator { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/json/ObjectToJsonTransformerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/json/ObjectToJsonTransformerTests.java index 78d21d3743..057422422c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/json/ObjectToJsonTransformerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/json/ObjectToJsonTransformerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 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. @@ -53,21 +53,21 @@ public class ObjectToJsonTransformerTests { @Test public void simpleStringPayload() throws Exception { - ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(); + ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(); String result = (String) transformer.transform(new GenericMessage("foo")).getPayload(); assertEquals("\"foo\"", result); } @Test public void withDefaultContentType() throws Exception { - ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(); + ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(); Message result = transformer.transform(new GenericMessage("foo")); assertEquals(ObjectToJsonTransformer.JSON_CONTENT_TYPE, result.getHeaders().get(MessageHeaders.CONTENT_TYPE)); } @Test public void withProvidedContentType() throws Exception { - ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(); + ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(); Message message = MessageBuilder.withPayload("foo").setHeader(MessageHeaders.CONTENT_TYPE, "text/xml").build(); Message result = transformer.transform(message); assertEquals("text/xml", result.getHeaders().get(MessageHeaders.CONTENT_TYPE)); @@ -75,7 +75,7 @@ public class ObjectToJsonTransformerTests { @Test public void withProvidedContentTypeWithOverride() throws Exception { - ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(); + ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(); transformer.setContentType(ObjectToJsonTransformer.JSON_CONTENT_TYPE); Message message = MessageBuilder.withPayload("foo").setHeader(MessageHeaders.CONTENT_TYPE, "text/xml").build(); Message result = transformer.transform(message); @@ -84,7 +84,7 @@ public class ObjectToJsonTransformerTests { @Test public void withProvidedContentTypeAsEmptyString() throws Exception { - ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(); + ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(); transformer.setContentType(""); Message message = MessageBuilder.withPayload("foo").build(); Message result = transformer.transform(message); @@ -93,29 +93,29 @@ public class ObjectToJsonTransformerTests { @Test public void withProvidedContentTypeAsEmptyStringDoesNotOverride() throws Exception { - ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(); + ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(); transformer.setContentType(""); Message message = MessageBuilder.withPayload("foo").setHeader(MessageHeaders.CONTENT_TYPE, "text/xml").build(); Message result = transformer.transform(message); assertEquals("text/xml", result.getHeaders().get(MessageHeaders.CONTENT_TYPE)); } - @Test(expected=IllegalArgumentException.class) + @Test(expected = IllegalArgumentException.class) public void withProvidedContentTypeAsNull() throws Exception { - ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(); + ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(); transformer.setContentType(null); } @Test public void simpleIntegerPayload() throws Exception { - ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(); + ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(); String result = (String) transformer.transform(new GenericMessage(123)).getPayload(); assertEquals("123", result); } @Test public void objectPayload() throws Exception { - ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(); + ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(); TestAddress address = new TestAddress(123, "Main Street"); TestPerson person = new TestPerson("John", "Doe", 42); person.setAddress(address); @@ -135,7 +135,7 @@ public class ObjectToJsonTransformerTests { public void objectPayloadWithCustomObjectMapper() throws Exception { ObjectMapper customMapper = new ObjectMapper(); customMapper.configure(JsonGenerator.Feature.QUOTE_FIELD_NAMES, Boolean.FALSE); - ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(new Jackson2JsonObjectMapper(customMapper)); + ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(new Jackson2JsonObjectMapper(customMapper)); TestPerson person = new TestPerson("John", "Doe", 42); person.setAddress(new TestAddress(123, "Main Street")); String result = (String) transformer.transform(new GenericMessage(person)).getPayload(); @@ -152,7 +152,7 @@ public class ObjectToJsonTransformerTests { @Test public void testBoonJsonObjectMapper() throws Exception { - ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(new BoonJsonObjectMapper()); + ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(new BoonJsonObjectMapper()); TestPerson person = new TestPerson("John", "Doe", 42); person.setAddress(new TestAddress(123, "Main Street")); String result = (String) transformer.transform(new GenericMessage(person)).getPayload(); diff --git a/src/reference/docbook/aggregator.xml b/src/reference/docbook/aggregator.xml index 4454321a51..787199e422 100644 --- a/src/reference/docbook/aggregator.xml +++ b/src/reference/docbook/aggregator.xml @@ -452,8 +452,15 @@ then you should simply provide an implementation of the ReleaseStrate - The timeout interval for sending the aggregated messages to the output or - reply channel. Optional. + The timeout interval to wait when sending a reply + Message to the output-channel or discard-channel. + By default the send will block for one second. + It is applied only if the output channel has some 'sending' limitations, e.g. QueueChannel + with a fixed 'capacity'. In this case a MessageDeliveryException is thrown. + The send-timeout is ignored in case of AbstractSubscribableChannel implementations. + In case of group-timeout(-expression) the MessageDeliveryException + from the scheduled expire task leads this task to be rescheduled. + Optional. diff --git a/src/reference/docbook/resequencer.xml b/src/reference/docbook/resequencer.xml index 01529a7ed6..01e849459c 100644 --- a/src/reference/docbook/resequencer.xml +++ b/src/reference/docbook/resequencer.xml @@ -102,8 +102,15 @@ - The timeout for sending out messages. - Optional. + The timeout interval to wait when sending a reply + Message to the output-channel or discard-channel. + By default the send will block for one second. + It is applied only if the output channel has some 'sending' limitations, e.g. QueueChannel + with a fixed 'capacity'. In this case a MessageDeliveryException is thrown. + The send-timeout is ignored in case of AbstractSubscribableChannel implementations. + In case of group-timeout(-expression) the MessageDeliveryException + from the scheduled expire task leads this task to be rescheduled. + Optional.