INT-2371 DefaultHeaderChannelRegistry Improvements

JIRA: https://jira.spring.io/browse/INT-3471

- Make final fields protected so they are available to subclasses
- Add property `removeOnGet` allowing the map entry to be removed immediately when it is used
- Add `time-to-live-expression` allowing override of the reaper delay

Polishing

Conflicts:
	src/reference/docbook/whats-new.xml

spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultHeaderChannelRegistry.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013 the original author or authors.
+ * Copyright 2013-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.
@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.integration.channel;
 
 import java.util.Date;
@@ -48,11 +49,13 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport
 
 	private static final int DEFAULT_REAPER_DELAY = 60000;
 
-	private final Map<String, MessageChannelWrapper> channels = new ConcurrentHashMap<String, DefaultHeaderChannelRegistry.MessageChannelWrapper>();
+	protected final Map<String, MessageChannelWrapper> channels = new ConcurrentHashMap<String, DefaultHeaderChannelRegistry.MessageChannelWrapper>();
 
-	private static final AtomicLong id = new AtomicLong();
+	protected static final AtomicLong id = new AtomicLong();
 
-	private final String uuid = UUID.randomUUID().toString() + ":";
+	protected final String uuid = UUID.randomUUID().toString() + ":";
+
+	private volatile boolean removeOnGet;
 
 	private volatile long reaperDelay;
 
@@ -74,7 +77,6 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport
 	/**
 	 * Constructs a registry with the provided delay (milliseconds) for
 	 * channel expiry.
-	 *
 	 * @param reaperDelay the delay in milliseconds.
 	 */
 	public DefaultHeaderChannelRegistry(long reaperDelay) {
@@ -83,7 +85,6 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport
 
 	/**
 	 * Set the reaper delay.
-	 *
 	 * @param reaperDelay the delay in milliseconds.
 	 */
 	public final void setReaperDelay(long reaperDelay) {
@@ -95,6 +96,16 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport
 		return reaperDelay;
 	}
 
+	/**
+	 * Set to true to immediately remove the channel mapping when
+	 * {@link #channelNameToChannel(String)} is invoked.
+	 * @param removeOnGet true to remove immediately, default false.
+	 * @since 4.1
+	 */
+	public void setRemoveOnGet(boolean removeOnGet) {
+		this.removeOnGet = removeOnGet;
+	}
+
 	@Override
 	public void setTaskScheduler(TaskScheduler taskScheduler) {
 		super.setTaskScheduler(taskScheduler);
@@ -160,9 +171,15 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport
 
 	@Override
 	public Object channelToChannelName(Object channel) {
+		return channelToChannelName(channel, this.reaperDelay);
+	}
+
+	@Override
+	public Object channelToChannelName(Object channel, long timeToLive) {
 		if (channel != null && channel instanceof MessageChannel) {
 			String name = this.uuid + DefaultHeaderChannelRegistry.id.incrementAndGet();
-			channels.put(name, new MessageChannelWrapper((MessageChannel) channel));
+			channels.put(name, new MessageChannelWrapper((MessageChannel) channel,
+					System.currentTimeMillis() + timeToLive));
 			if (logger.isDebugEnabled()) {
 				logger.debug("Registered " + channel + " as " + name);
 			}
@@ -176,7 +193,13 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport
 	@Override
 	public MessageChannel channelNameToChannel(String name) {
 		if (name != null) {
-			MessageChannelWrapper messageChannelWrapper = this.channels.get(name);
+			MessageChannelWrapper messageChannelWrapper;
+			if (this.removeOnGet) {
+				messageChannelWrapper = this.channels.remove(name);
+			}
+			else {
+				messageChannelWrapper = this.channels.get(name);
+			}
 			if (logger.isDebugEnabled() && messageChannelWrapper != null) {
 				logger.debug("Retrieved " + messageChannelWrapper.getChannel() + " with " + name);
 			}
@@ -204,10 +227,10 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport
 			logger.trace("Reaper started; channels size=" + this.channels.size());
 		}
 		Iterator<Entry<String, MessageChannelWrapper>> iterator = this.channels.entrySet().iterator();
-		long threshold = System.currentTimeMillis() - this.reaperDelay;
+		long now = System.currentTimeMillis();
 		while (iterator.hasNext()) {
 			Entry<String, MessageChannelWrapper> entry = iterator.next();
-			if (entry.getValue().getCreated() < threshold) {
+			if (entry.getValue().getExpireAt() < now) {
 				if (logger.isDebugEnabled()) {
 					logger.debug("Expiring " + entry.getKey() + " (" + entry.getValue().getChannel() + ")");
 				}
@@ -230,15 +253,15 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport
 
 		private final MessageChannel channel;
 
-		private final long created;
+		private final long expireAt;
 
-		private MessageChannelWrapper(MessageChannel channel) {
+		private MessageChannelWrapper(MessageChannel channel, long expireAt) {
 			this.channel = channel;
-			this.created = System.currentTimeMillis();
+			this.expireAt = expireAt;
 		}
 
-		public final long getCreated() {
-			return created;
+		public final long getExpireAt() {
+			return expireAt;
 		}
 
 		public final MessageChannel getChannel() {

spring-integration-core/src/main/java/org/springframework/integration/config/xml/HeaderEnricherParserSupport.java

@@ -59,9 +59,9 @@ public abstract class HeaderEnricherParserSupport extends AbstractTransformerPar
 	static {
 		cannedHeaderElementExpressions.put("header-channels-to-string", new String[][] {
 				{"replyChannel", "@" + IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME
-						+ ".channelToChannelName(headers.replyChannel)" },
+						+ ".channelToChannelName(headers.replyChannel, ####)" },
 				{"errorChannel", "@" + IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME
-						+ ".channelToChannelName(headers.errorChannel)" },
+						+ ".channelToChannelName(headers.errorChannel, ####)" },
 		});
 	}
 
@@ -132,10 +132,17 @@ public abstract class HeaderEnricherParserSupport extends AbstractTransformerPar
 					}
 				}
 				if (headerName == null) {
+					String ttlExpression = headerElement.getAttribute("time-to-live-expression");
 					if (cannedHeaderElementExpressions.containsKey(elementName)) {
 						for (int j = 0; j < cannedHeaderElementExpressions.get(elementName).length; j++) {
 							headerName = cannedHeaderElementExpressions.get(elementName)[j][0];
 							expression = cannedHeaderElementExpressions.get(elementName)[j][1];
+							if (StringUtils.hasText(ttlExpression)) {
+								expression = expression.replace("####", ttlExpression);
+							}
+							else {
+								expression = expression.replace(", ####", "");
+							}
 							overwrite = "true";
 							this.addHeader(element, headers, parserContext, headerName, headerElement, headerType,
 									expression, overwrite);
@@ -143,7 +150,7 @@ public abstract class HeaderEnricherParserSupport extends AbstractTransformerPar
 					}
 				}
 				else {
-					this.addHeader(element, headers, parserContext, headerName, headerElement, headerType, expression,
+					this.addHeader(element, headers, parserContext, headerName, headerElement, headerType, null,
 							overwrite);
 				}
 			}
@@ -178,11 +185,13 @@ public abstract class HeaderEnricherParserSupport extends AbstractTransformerPar
 				expressionElement = subElement;
 			}
 			if (beanElement == null && scriptElement == null && expressionElement == null) {
-				parserContext.getReaderContext().error("Only 'bean', 'script' or 'expression' can be defined as a sub-element", element);
+				parserContext.getReaderContext()
+						.error("Only 'bean', 'script' or 'expression' can be defined as a sub-element", element);
 			}
 		}
 		if (StringUtils.hasText(expression) && expressionElement != null) {
-			parserContext.getReaderContext().error("The 'expression' attribute and sub-element are mutually exclusive", element);
+			parserContext.getReaderContext()
+					.error("The 'expression' attribute and sub-element are mutually exclusive", element);
 		}
 
 		boolean isValue = StringUtils.hasText(value);
@@ -194,7 +203,9 @@ public abstract class HeaderEnricherParserSupport extends AbstractTransformerPar
 		BeanDefinition innerComponentDefinition = null;
 
 		if (beanElement != null) {
-			innerComponentDefinition = parserContext.getDelegate().parseBeanDefinitionElement(beanElement).getBeanDefinition();
+			innerComponentDefinition = parserContext.getDelegate()
+					.parseBeanDefinitionElement(beanElement)
+					.getBeanDefinition();
 		}
 		else if (isScript) {
 			innerComponentDefinition = parserContext.getDelegate().parseCustomElement(scriptElement);
@@ -203,7 +214,8 @@ public abstract class HeaderEnricherParserSupport extends AbstractTransformerPar
 		boolean isCustomBean = innerComponentDefinition != null;
 
 		if (hasMethod && isScript) {
-			parserContext.getReaderContext().error("The 'method' attribute cannot be used when a 'script' sub-element is defined", element);
+			parserContext.getReaderContext()
+					.error("The 'method' attribute cannot be used when a 'script' sub-element is defined", element);
 		}
 
 		if (!(isValue ^ (isRef ^ (isExpression ^ isCustomBean)))) {
@@ -218,17 +230,19 @@ public abstract class HeaderEnricherParserSupport extends AbstractTransformerPar
 			}
 			Object headerValue = (headerType != null) ?
 					new TypedStringValue(value, headerType) : value;
-			valueProcessorBuilder = BeanDefinitionBuilder.genericBeanDefinition(StaticHeaderValueMessageProcessor.class);
-			valueProcessorBuilder.addConstructorArgValue(headerValue);
+			valueProcessorBuilder = BeanDefinitionBuilder.genericBeanDefinition(StaticHeaderValueMessageProcessor.class)
+					.addConstructorArgValue(headerValue);
 		}
 		else if (isExpression) {
 			if (hasMethod) {
 				parserContext.getReaderContext().error(
 						"The 'method' attribute cannot be used with the 'expression' attribute.", element);
 			}
-			valueProcessorBuilder = BeanDefinitionBuilder.genericBeanDefinition(ExpressionEvaluatingHeaderValueMessageProcessor.class);
+			valueProcessorBuilder =
+					BeanDefinitionBuilder.genericBeanDefinition(ExpressionEvaluatingHeaderValueMessageProcessor.class);
 			if (expressionElement != null) {
-				BeanDefinitionBuilder dynamicExpressionBuilder = BeanDefinitionBuilder.genericBeanDefinition(DynamicExpression.class);
+				BeanDefinitionBuilder dynamicExpressionBuilder =
+						BeanDefinitionBuilder.genericBeanDefinition(DynamicExpression.class);
 				dynamicExpressionBuilder.addConstructorArgValue(expressionElement.getAttribute("key"));
 				dynamicExpressionBuilder.addConstructorArgReference(expressionElement.getAttribute("source"));
 				valueProcessorBuilder.addConstructorArgValue(dynamicExpressionBuilder.getBeanDefinition());
@@ -244,8 +258,9 @@ public abstract class HeaderEnricherParserSupport extends AbstractTransformerPar
 						"The 'type' attribute cannot be used with an inner bean.", element);
 			}
 			if (hasMethod || isScript) {
-				valueProcessorBuilder = BeanDefinitionBuilder.genericBeanDefinition(MessageProcessingHeaderValueMessageProcessor.class);
-				valueProcessorBuilder.addConstructorArgValue(innerComponentDefinition);
+				valueProcessorBuilder =
+						BeanDefinitionBuilder.genericBeanDefinition(MessageProcessingHeaderValueMessageProcessor.class)
+								.addConstructorArgValue(innerComponentDefinition);
 				if (hasMethod) {
 					valueProcessorBuilder.addConstructorArgValue(method);
 				}
@@ -261,13 +276,14 @@ public abstract class HeaderEnricherParserSupport extends AbstractTransformerPar
 						"The 'type' attribute cannot be used with the 'ref' attribute.", element);
 			}
 			if (hasMethod) {
-				valueProcessorBuilder = BeanDefinitionBuilder.genericBeanDefinition(MessageProcessingHeaderValueMessageProcessor.class);
-				valueProcessorBuilder.addConstructorArgReference(ref);
-				valueProcessorBuilder.addConstructorArgValue(method);
+				valueProcessorBuilder =
+						BeanDefinitionBuilder.genericBeanDefinition(MessageProcessingHeaderValueMessageProcessor.class)
+								.addConstructorArgReference(ref)
+								.addConstructorArgValue(method);
 			}
 			else {
-				valueProcessorBuilder = BeanDefinitionBuilder.genericBeanDefinition(StaticHeaderValueMessageProcessor.class);
-				valueProcessorBuilder.addConstructorArgReference(ref);
+				valueProcessorBuilder = BeanDefinitionBuilder.genericBeanDefinition(StaticHeaderValueMessageProcessor.class)
+						.addConstructorArgReference(ref);
 			}
 		}
 		if (StringUtils.hasText(overwrite)) {
@@ -278,7 +294,6 @@ public abstract class HeaderEnricherParserSupport extends AbstractTransformerPar
 
 	/**
 	 * Subclasses may override this method to provide any additional processing.
-	 *
 	 * @param builder The builder.
 	 * @param element The element.
 	 * @param parserContext The parser context.

spring-integration-core/src/main/java/org/springframework/integration/support/channel/HeaderChannelRegistry.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013 the original author or authors.
+ * Copyright 2013-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.
@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.integration.support.channel;
 
 import org.springframework.jmx.export.annotation.ManagedAttribute;
@@ -33,11 +34,21 @@ public interface HeaderChannelRegistry {
 	/**
 	 * Converts the channel to a name (String). If the channel is not a
 	 * {@link MessageChannel}, it is returned unchanged.
-	 *
 	 * @param channel The channel.
 	 * @return The channel name, or the channel if it is not a MessageChannel.
 	 */
-	public abstract Object channelToChannelName(Object channel);
+	Object channelToChannelName(Object channel);
+
+	/**
+	 * Converts the channel to a name (String). If the channel is not a
+	 * {@link MessageChannel}, it is returned unchanged.
+	 * @param channel The channel.
+	 * @param timeToLive How long (ms) at a minimum, the channel mapping should
+	 * remain in the registry.
+	 * @return The channel name, or the channel if it is not a MessageChannel.
+	 * @since 4.1
+	 */
+	Object channelToChannelName(Object channel, long timeToLive);
 
 	/**
 	 * Converts the channel name back to a {@link MessageChannel} (if it is
@@ -45,18 +56,18 @@ public interface HeaderChannelRegistry {
 	 * @param name The name of the channel.
 	 * @return The channel, or null if there is no channel registered with the name.
 	 */
-	public abstract MessageChannel channelNameToChannel(String name);
+	MessageChannel channelNameToChannel(String name);
 
 	/**
 	 * @return the current size of the registry
 	 */
 	@ManagedAttribute
-	public abstract int size();
+	int size();
 
 	/**
 	 * Cancel the scheduled reap task and run immediately; then reschedule.
 	 */
 	@ManagedOperation(description = "Cancel the scheduled reap task and run immediately; then reschedule.")
-	public abstract void runReaper();
+	void runReaper();
 
 }

spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd

@@ -1917,6 +1917,16 @@
 					if the header does not exist, or if the header does not currently reference a MessageChannel.
 					</xsd:documentation>
 				</xsd:annotation>
+				<xsd:complexType>
+					<xsd:attribute name="time-to-live-expression" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+							Overrides the header channel registry default 'reaperDelay' - specifies the minimum time
+							that the mapping will remain in the registry, in milliseconds.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+				</xsd:complexType>
 			</xsd:element>
 			<xsd:element name="error-channel" type="referenceOrValueHeaderType">
 				<xsd:annotation>

spring-integration-core/src/test/java/org/springframework/integration/channel/registry/HeaderChannelRegistryTests-context.xml

@@ -11,6 +11,14 @@
 		<int:header-channels-to-string />
 	</int:header-enricher>
 
+	<int:header-enricher input-channel="inputTtl" output-channel="next">
+		<int:header-channels-to-string time-to-live-expression="120000" />
+	</int:header-enricher>
+
+	<int:header-enricher input-channel="inputCustomTtl" output-channel="next">
+		<int:header-channels-to-string time-to-live-expression="headers['channelTTL'] ?: 240000" />
+	</int:header-enricher>
+
 	<int:transformer input-channel="next">
 		<bean class="org.springframework.integration.channel.registry.HeaderChannelRegistryTests$Foo" />
 	</int:transformer>

spring-integration-core/src/test/java/org/springframework/integration/channel/registry/HeaderChannelRegistryTests.java

@@ -15,10 +15,14 @@
  */
 package org.springframework.integration.channel.registry;
 
+import static org.hamcrest.Matchers.allOf;
+import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.lessThan;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
@@ -26,6 +30,8 @@ import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import java.util.Map;
+
 import org.hamcrest.Matchers;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -45,6 +51,7 @@ import org.springframework.integration.handler.AbstractReplyProducingMessageHand
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
 import org.springframework.integration.support.channel.HeaderChannelRegistry;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.core.DestinationResolutionException;
@@ -68,6 +75,12 @@ public class HeaderChannelRegistryTests {
 	@Autowired
 	MessageChannel input;
 
+	@Autowired
+	MessageChannel inputTtl;
+
+	@Autowired
+	MessageChannel inputCustomTtl;
+
 	@Autowired
 	MessageChannel inputPolled;
 
@@ -83,6 +96,9 @@ public class HeaderChannelRegistryTests {
 	@Autowired
 	Gateway gatewayExplicitReplyChannel;
 
+	@Autowired
+	DefaultHeaderChannelRegistry registry;
+
 	@Test
 	public void testReplace() {
 		MessagingTemplate template = new MessagingTemplate();
@@ -90,6 +106,51 @@ public class HeaderChannelRegistryTests {
 		Message<?> reply = template.sendAndReceive(new GenericMessage<String>("foo"));
 		assertNotNull(reply);
 		assertEquals("echo:foo", reply.getPayload());
+		String stringReplyChannel = reply.getHeaders().get("stringReplyChannel", String.class);
+		assertThat(TestUtils.getPropertyValue(
+				TestUtils.getPropertyValue(registry, "channels", Map.class)
+					.get(stringReplyChannel), "expireAt", Long.class) - System.currentTimeMillis(),
+						lessThan(61000L));
+	}
+
+	@Test
+	public void testReplaceTtl() {
+		MessagingTemplate template = new MessagingTemplate();
+		template.setDefaultDestination(this.inputTtl);
+		Message<?> reply = template.sendAndReceive(new GenericMessage<String>("ttl"));
+		assertNotNull(reply);
+		assertEquals("echo:ttl", reply.getPayload());
+		String stringReplyChannel = reply.getHeaders().get("stringReplyChannel", String.class);
+		assertThat(TestUtils.getPropertyValue(
+				TestUtils.getPropertyValue(registry, "channels", Map.class)
+					.get(stringReplyChannel), "expireAt", Long.class) - System.currentTimeMillis(),
+						greaterThan(100000L));
+	}
+
+	@Test
+	public void testReplaceCustomTtl() {
+		MessagingTemplate template = new MessagingTemplate();
+		template.setDefaultDestination(this.inputCustomTtl);
+		Message<String> requestMessage = MessageBuilder.withPayload("ttl")
+				.setHeader("channelTTL", 180000)
+				.build();
+		Message<?> reply = template.sendAndReceive(requestMessage);
+		assertNotNull(reply);
+		assertEquals("echo:ttl", reply.getPayload());
+		String stringReplyChannel = reply.getHeaders().get("stringReplyChannel", String.class);
+		assertThat(TestUtils.getPropertyValue(
+				TestUtils.getPropertyValue(registry, "channels", Map.class)
+					.get(stringReplyChannel), "expireAt", Long.class) - System.currentTimeMillis(),
+						allOf(greaterThan(160000L), lessThan(181000L)));
+		// Now for Elvis...
+		reply = template.sendAndReceive(new GenericMessage<String>("ttl"));
+		assertNotNull(reply);
+		assertEquals("echo:ttl", reply.getPayload());
+		stringReplyChannel = reply.getHeaders().get("stringReplyChannel", String.class);
+		assertThat(TestUtils.getPropertyValue(
+				TestUtils.getPropertyValue(registry, "channels", Map.class)
+					.get(stringReplyChannel), "expireAt", Long.class) - System.currentTimeMillis(),
+						greaterThan(220000L));
 	}
 
 	@Test
@@ -205,6 +266,21 @@ public class HeaderChannelRegistryTests {
 		}
 	}
 
+	@Test
+	public void testRemoveOnGet() {
+		DefaultHeaderChannelRegistry registry = new DefaultHeaderChannelRegistry();
+		MessageChannel channel = new DirectChannel();
+		String foo = (String) registry.channelToChannelName(channel);
+		Map<?, ?> map = TestUtils.getPropertyValue(registry, "channels", Map.class);
+		assertEquals(1, map.size());
+		assertSame(channel, registry.channelNameToChannel(foo));
+		assertEquals(1, map.size());
+		registry.setRemoveOnGet(true);
+		assertSame(channel, registry.channelNameToChannel(foo));
+		assertEquals(0, map.size());
+	}
+
+
 	public static class Foo extends AbstractReplyProducingMessageHandler {
 
 		@Override
@@ -216,7 +292,9 @@ public class HeaderChannelRegistryTests {
 			if (requestMessage.getPayload().equals("bar")) {
 				throw new RuntimeException("intentional");
 			}
-			return "echo:" + requestMessage.getPayload();
+			return MessageBuilder.withPayload("echo:" + requestMessage.getPayload())
+					.setHeader("stringReplyChannel", requestMessage.getHeaders().getReplyChannel())
+					.build();
 		}
 
 	}

src/reference/docbook/content-enrichment.xml

@@ -148,7 +148,8 @@
 			the incoming Message.
 	    </para>
 
-		<para><emphasis role="bold">Header Channel Registry</emphasis></para>
+	  <section id="header-channel-registry">
+		<title>Header Channel Registry</title>
 
 		<para>
 			Starting with <emphasis>Spring Integration 3.0</emphasis>, a new sub-element
@@ -169,9 +170,16 @@
 			bean. By default, the framework creates a <classname>DefaultHeaderChannelRegistry</classname>
 			with the default expiry (60 seconds). Channels
 			are removed from the registry after this time. To change this, simply define a bean
-			with id <code>integrationHeaderChannelRegistry</code> and configure the required delay using
+			with id <code>integrationHeaderChannelRegistry</code> and configure the required default delay using
 			a constructor argument (milliseconds).
 		</para>
+		
+		<para>
+			Since <emphasis>version 4.1</emphasis>, you can set a property <code>removeOnGet</code>
+			to <code>true</code> on the <code>&lt;bean/&gt;</code> definition, and the mapping entry will be removed
+			immediately on first use. This might be useful in a high-volume environment and when
+			the channel is only used once, rather than waiting for the reaper to remove it.
+		</para>
 
 		<para>
 			The <classname>HeaderChannelRegistry</classname> has a <code>size()</code> method to
@@ -188,15 +196,32 @@
 			This sub-element is a convenience only, and is the equivalent of specifying:
 		</para>
 		<programlisting language="xml"><![CDATA[<int:reply-channel
-	expression="@integrationHeaderChannelRegistry.channelToChannelName(headers.replyChannel)"/>
+	expression="@integrationHeaderChannelRegistry.channelToChannelName(headers.replyChannel)"
+	overwrite="true" />
 <int:error-channel
-	expression="@integrationHeaderChannelRegistry.channelToChannelName(headers.errorChannel)"/>]]></programlisting>
+	expression="@integrationHeaderChannelRegistry.channelToChannelName(headers.errorChannel)"
+	overwrite="true" />]]></programlisting>
 
-		<tip>
-			For more examples for configuring header enrichers, see
-			<ulink url="https://github.com/SpringSource/spring-integration/wiki/Header-Enricher-Advanced-Configuration">
-			Header Enricher Advanced Configuration</ulink>.
-		</tip>
+		<para>
+			Starting with <emphasis>version 4.1</emphasis>, you can now override the registry's configured
+			reaper delay, so the the channel mapping is retained for at least the specified time, regardless
+			of the reaper delay:
+		</para>
+		<programlisting language="xml"><![CDATA[<int:header-enricher input-channel="inputTtl" output-channel="next">
+	<int:header-channels-to-string time-to-live-expression="120000" />
+</int:header-enricher>
+
+<int:header-enricher input-channel="inputCustomTtl" output-channel="next">
+	<int:header-channels-to-string 
+		time-to-live-expression="headers['channelTTL'] ?: 120000" />
+</int:header-enricher>]]></programlisting>
+
+		<para>
+			In the first case, the time to live for every header channel mapping will be 2 minutes; in the second case,
+			the time to live is specified in the message header and uses an elvis operator to use
+			2 minutes if there is no header.
+		</para>
+	  </section>
     </section>
 
     <section id="payload-enricher">

src/reference/docbook/whats-new.xml

@@ -97,5 +97,14 @@
 				See <xref linkend="content-enricher"/> for more information.
 			</para>
 		</section>
+		<section id="4.1-header-channel-registry">
+			<title>Header Channel Registry</title>
+			<para>
+				The <code>&lt;header-enricher/&gt;</code>'s <code>&lt;header-channels-to-string/&gt;</code>
+				element can now override the header channel registry's default time for retaining channel
+				mappings.
+				See <xref linkend="header-channel-registry"/> for more information.
+			</para>
+		</section>
 	</section>
 </chapter>