The <channel-adapter/> now actually creates a channel instance rather than requiring another distinct channel object. Instead of configuring the poller on the channel-adapter, it is currently to be configured on the consuming endpoint just as if the <channel-adapter/> were any other pollable channel (e.g. <queue-channel/>).

org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractChannelAdapter.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2002-2008 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.channel;
+
+import org.springframework.integration.message.BlockingTarget;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageTarget;
+import org.springframework.util.Assert;
+
+/**
+ * The base class for Channel Adapters.
+ * 
+ * @author Mark Fisher
+ */
+public class AbstractChannelAdapter extends AbstractMessageChannel {
+
+	private final MessageTarget target;
+
+
+	public AbstractChannelAdapter(String name, MessageTarget target) {
+		Assert.notNull(name, "name must not be null");
+		this.setName(name);
+		this.target = target;
+	}
+
+
+	@Override
+	protected boolean doSend(Message<?> message, long timeout) {
+		if (this.target == null) {
+			return false;
+		}
+		return (timeout >= 0 && this.target instanceof BlockingTarget)
+				? ((BlockingTarget) this.target).send(message, timeout)
+				: this.target.send(message);
+	}
+
+}

org.springframework.integration/src/main/java/org/springframework/integration/channel/PollableChannelAdapter.java

@@ -0,0 +1,84 @@
+/*
+ * Copyright 2002-2008 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.channel;
+
+import java.util.List;
+
+import org.springframework.integration.message.BlockingSource;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageDeliveryAware;
+import org.springframework.integration.message.MessageTarget;
+import org.springframework.integration.message.MessagingException;
+import org.springframework.integration.message.PollableSource;
+import org.springframework.integration.message.selector.MessageSelector;
+
+/**
+ * Channel Adapter implementation for a {@link PollableSource}.
+ * 
+ * @author Mark Fisher
+ */
+public class PollableChannelAdapter extends AbstractChannelAdapter implements PollableChannel, MessageDeliveryAware {
+
+	private final PollableSource<?> source;
+
+
+	public PollableChannelAdapter(String name, PollableSource<?> source, MessageTarget target) {
+		super(name, target);
+		this.source = source;
+	}
+
+
+	public Message<?> receive() {
+		return this.receive(-1);
+	}
+
+	public Message<?> receive(long timeout) {
+		if (this.source != null) {
+			return (timeout >= 0 && this.source instanceof BlockingSource)
+					? ((BlockingSource<?>) this.source).receive(timeout)
+					: this.source.receive();
+		}
+		return null;
+	}
+
+	public List<Message<?>> clear() {
+		if (this.source != null && this.source instanceof PollableChannel) {
+			return ((PollableChannel) this.source).clear();
+		}
+		return null;
+	}
+
+	public List<Message<?>> purge(MessageSelector selector) {
+		if (this.source != null && this.source instanceof PollableChannel) {
+			return ((PollableChannel) this.source).purge(selector);
+		}
+		return null;
+	}
+
+	public void onSend(Message<?> sentMessage) {
+		if (this.source != null && this.source instanceof MessageDeliveryAware) {
+			((MessageDeliveryAware) this.source).onSend(sentMessage);
+		}
+	}
+
+	public void onFailure(MessagingException exception) {
+		if (this.source != null && this.source instanceof MessageDeliveryAware) {
+			((MessageDeliveryAware) this.source).onFailure(exception);
+		}
+	}
+
+}

org.springframework.integration/src/main/java/org/springframework/integration/channel/SubscribableChannelAdapter.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2002-2008 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.channel;
+
+import org.springframework.integration.message.MessageTarget;
+import org.springframework.integration.message.SubscribableSource;
+import org.springframework.util.Assert;
+
+/**
+ * Channel Adapter implementation for a {@link SubscribableSource}.
+ * 
+ * @author Mark Fisher
+ */
+public class SubscribableChannelAdapter extends AbstractChannelAdapter implements SubscribableSource {
+
+	private final SubscribableSource source;
+
+
+	public SubscribableChannelAdapter(String name, SubscribableSource source, MessageTarget target) {
+		super(name, target);
+		Assert.notNull(source, "source must not be null");
+		this.source = source;
+	}
+
+
+	public boolean subscribe(MessageTarget target) {
+		return this.source.subscribe(target);
+	}
+
+	public boolean unsubscribe(MessageTarget target) {
+		return this.source.unsubscribe(target);
+	}
+
+}

org.springframework.integration/src/main/java/org/springframework/integration/channel/config/AbstractChannelParser.java

@@ -28,6 +28,7 @@ import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.channel.interceptor.MessageSelectingInterceptor;
+import org.springframework.integration.config.ChannelInterceptorParser;
 import org.springframework.integration.message.selector.PayloadTypeSelector;
 import org.springframework.util.StringUtils;
 
@@ -38,13 +39,6 @@ import org.springframework.util.StringUtils;
  */
 public abstract class AbstractChannelParser extends AbstractSingleBeanDefinitionParser {
 
-	private static final String DATATYPE_ATTRIBUTE = "datatype";
-
-	private static final String INTERCEPTOR_ELEMENT = "interceptor";
-
-	private static final String INTERCEPTORS_PROPERTY = "interceptors";
-
-
 	@Override
 	protected boolean shouldGenerateId() {
 		return false;
@@ -58,22 +52,25 @@ public abstract class AbstractChannelParser extends AbstractSingleBeanDefinition
 	@Override
 	protected abstract Class<?> getBeanClass(Element element);
 
-	protected void configureConstructorArgs(BeanDefinitionBuilder builder, Element element) {
+	protected void postProcess(BeanDefinitionBuilder builder, Element element) {
 	}
 
 	@Override
 	@SuppressWarnings("unchecked")
 	protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
-		ManagedList interceptors = new ManagedList();
+		ManagedList interceptors = null;
 		NodeList childNodes = element.getChildNodes();
 		for (int i = 0; i < childNodes.getLength(); i++) {
 			Node child = childNodes.item(i);
-			if (child.getNodeType() == Node.ELEMENT_NODE && child.getLocalName().equals(INTERCEPTOR_ELEMENT)) {
-				String ref = ((Element) child).getAttribute("ref");
-				interceptors.add(new RuntimeBeanReference(ref));
+			if (child.getNodeType() == Node.ELEMENT_NODE && child.getLocalName().equals("interceptors")) {
+				ChannelInterceptorParser interceptorParser = new ChannelInterceptorParser();
+				interceptors = interceptorParser.parseInterceptors((Element) child, parserContext);
 			}
 		}
-		String datatypeAttr = element.getAttribute(DATATYPE_ATTRIBUTE);
+		if (interceptors == null) {
+			interceptors = new ManagedList();
+		}
+		String datatypeAttr = element.getAttribute("datatype");
 		if (StringUtils.hasText(datatypeAttr)) {
 			String[] datatypes = StringUtils.commaDelimitedListToStringArray(datatypeAttr);
 			RootBeanDefinition selectorDef = new RootBeanDefinition(PayloadTypeSelector.class);
@@ -88,8 +85,8 @@ public abstract class AbstractChannelParser extends AbstractSingleBeanDefinition
 			parserContext.registerBeanComponent(interceptorComponent);
 			interceptors.add(new RuntimeBeanReference(interceptorBeanName));
 		}
-		builder.addPropertyValue(INTERCEPTORS_PROPERTY, interceptors);
-		this.configureConstructorArgs(builder, element);
+		builder.addPropertyValue("interceptors", interceptors);
+		this.postProcess(builder, element);
 	}
 
 }

org.springframework.integration/src/main/java/org/springframework/integration/channel/config/ChannelAdapterFactoryBean.java

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2002-2008 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.channel.config;
+
+import java.util.List;
+
+import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.beans.factory.FactoryBean;
+import org.springframework.integration.ConfigurationException;
+import org.springframework.integration.channel.AbstractMessageChannel;
+import org.springframework.integration.channel.ChannelInterceptor;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.channel.PollableChannelAdapter;
+import org.springframework.integration.channel.SubscribableChannelAdapter;
+import org.springframework.integration.message.MessageSource;
+import org.springframework.integration.message.MessageTarget;
+import org.springframework.integration.message.PollableSource;
+import org.springframework.integration.message.SubscribableSource;
+
+/**
+ * @author Mark Fisher
+ */
+public class ChannelAdapterFactoryBean implements FactoryBean, BeanNameAware {
+
+	private volatile String name;
+
+	private volatile MessageSource<?> source;
+
+	private volatile MessageTarget target;
+
+	private volatile AbstractMessageChannel channel;
+
+	private List<ChannelInterceptor> interceptors;
+
+	private volatile boolean initialized;
+
+	private final Object initializationMonitor = new Object();
+
+
+	public void setBeanName(String name) {
+		this.name = name;
+	}
+
+	public void setSource(MessageSource<?> source) {
+		this.source = source;
+	}
+
+	public void setTarget(MessageTarget target) {
+		this.target = target;
+	}
+
+	public void setInterceptors(List<ChannelInterceptor> interceptors) {
+		this.interceptors = interceptors;
+	}
+
+	public Object getObject() throws Exception {
+		if (!this.initialized) {
+			this.initializeChannel();
+		}
+		return this.channel;
+	}
+
+	public Class<?> getObjectType() {
+		if (!this.initialized) {
+			return MessageChannel.class;
+		}
+		return this.channel.getClass();
+	}
+
+	public boolean isSingleton() {
+		return true;
+	}
+
+	private void initializeChannel() {
+		synchronized (this.initializationMonitor) {
+			if (this.initialized) {
+				return;
+			}
+			if (this.source == null || source instanceof PollableSource) {
+				this.channel = new PollableChannelAdapter(
+						this.name, (PollableSource<?>) this.source, this.target);
+			}
+			else if (this.source instanceof SubscribableSource) {
+				this.channel = new SubscribableChannelAdapter(
+						this.name, (SubscribableSource) this.source, this.target);
+			}
+			else {
+				throw new ConfigurationException("source must be either a PollableSource or SubscribableSource");
+			}
+			if (this.interceptors != null) {
+				this.channel.setInterceptors(this.interceptors);
+			}
+		}
+	}
+
+}

org.springframework.integration/src/main/java/org/springframework/integration/channel/config/PriorityChannelParser.java

@@ -35,8 +35,8 @@ public class PriorityChannelParser extends QueueChannelParser {
 	}
 
 	@Override
-	protected void configureConstructorArgs(BeanDefinitionBuilder builder, Element element) {
-		super.configureConstructorArgs(builder, element);
+	protected void postProcess(BeanDefinitionBuilder builder, Element element) {
+		super.postProcess(builder, element);
 		String comparator = element.getAttribute("comparator");
 		if (StringUtils.hasText(comparator)) {
 			builder.addConstructorArgReference(comparator);

org.springframework.integration/src/main/java/org/springframework/integration/channel/config/QueueChannelParser.java

@@ -35,7 +35,7 @@ public class QueueChannelParser extends AbstractChannelParser {
 	}
 
 	@Override
-	protected void configureConstructorArgs(BeanDefinitionBuilder builder, Element element) {
+	protected void postProcess(BeanDefinitionBuilder builder, Element element) {
 		String capacityAttribute = element.getAttribute("capacity");
 		int capacity = (StringUtils.hasText(capacityAttribute)) ?
 				Integer.parseInt(capacityAttribute) : QueueChannel.DEFAULT_CAPACITY;

org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractHandlerEndpointParser.java

@@ -112,7 +112,7 @@ public abstract class AbstractHandlerEndpointParser extends AbstractSingleBeanDe
 				}
 				else if (INTERCEPTORS_ELEMENT.equals(localName)) {
 					EndpointInterceptorParser parser = new EndpointInterceptorParser();
-					ManagedList interceptors = parser.parseEndpointInterceptors(childElement, parserContext);
+					ManagedList interceptors = parser.parseInterceptors(childElement, parserContext);
 					builder.addPropertyValue("interceptors", interceptors);
 				}
 			}

org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractInterceptorParser.java

@@ -0,0 +1,109 @@
+/*
+ * Copyright 2002-2008 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.config;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.w3c.dom.Element;
+import org.w3c.dom.Node;
+import org.w3c.dom.NodeList;
+
+import org.springframework.beans.factory.config.RuntimeBeanReference;
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.ManagedList;
+import org.springframework.beans.factory.xml.NamespaceHandler;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.util.Assert;
+
+/**
+ * A helper class for parsing the sub-elements of an endpoint
+ * or channel-adapter's <em>interceptors</em> element.
+ * 
+ * @author Mark Fisher
+ */
+public abstract class AbstractInterceptorParser {
+
+	private final Map<String, BeanDefinitionRegisteringParser> parsers = new HashMap<String, BeanDefinitionRegisteringParser>();
+
+	private volatile boolean initialized;
+
+	private final Object initializationMonitor = new Object();
+
+
+	protected abstract Map<String, BeanDefinitionRegisteringParser> getParserMap();
+
+	private void initializeParserMap() {
+		synchronized (this.initializationMonitor) {
+			if (!this.initialized) {
+				Map<String, BeanDefinitionRegisteringParser> parserMap = this.getParserMap();
+				if (parserMap != null) {
+					this.parsers.putAll(parserMap);
+				}
+				this.initialized = true;
+			}
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	public ManagedList parseInterceptors(Element element, ParserContext parserContext) {
+		if (!initialized) {
+			this.initializeParserMap();
+		}
+		ManagedList interceptors = new ManagedList();
+		NodeList childNodes = element.getChildNodes();
+		for (int i = 0; i < childNodes.getLength(); i++) {
+			Node child = childNodes.item(i);
+			if (child.getNodeType() == Node.ELEMENT_NODE) {
+				Element childElement = (Element) child;
+				String localName = child.getLocalName();
+				if ("bean".equals(localName)) {
+					interceptors.add(new RuntimeBeanReference(
+							IntegrationNamespaceUtils.parseBeanDefinitionElement(childElement, parserContext)));
+				}
+				else if ("ref".equals(localName)) {
+					String ref = childElement.getAttribute("bean");
+					interceptors.add(new RuntimeBeanReference(ref));
+				}
+				else {
+					BeanDefinitionRegisteringParser parser = this.parsers.get(localName);
+					String interceptorBeanName = null;
+					if (parser == null) {
+						interceptorBeanName = handleNonstandardInterceptor(childElement, parserContext);
+					}
+					else {
+						interceptorBeanName = parser.parse(childElement, parserContext);
+					}
+					interceptors.add(new RuntimeBeanReference(interceptorBeanName));
+				}
+			}
+		}
+		return interceptors;
+	}
+
+	protected String handleNonstandardInterceptor(Element childElement, ParserContext parserContext) {
+		NamespaceHandler handler = parserContext.getReaderContext().getNamespaceHandlerResolver()
+				.resolve(childElement.getNamespaceURI());
+		AbstractBeanDefinition interceptorDefinition =
+				((AbstractBeanDefinition) handler.parse(childElement, parserContext));
+		String beanName = (String) interceptorDefinition.getMetadataAttribute("interceptorName").getValue();
+		Assert.hasText("No value for interceptorName provided by namespace handler for element '"
+				+ childElement.getNodeName() + "'");
+		return beanName;
+	}
+
+}

org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java

@@ -18,115 +18,25 @@ package org.springframework.integration.config;
 
 import org.w3c.dom.Element;
 
-import org.springframework.beans.factory.config.BeanDefinition;
-import org.springframework.beans.factory.config.RuntimeBeanReference;
-import org.springframework.beans.factory.parsing.BeanComponentDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
-import org.springframework.beans.factory.support.ManagedList;
-import org.springframework.beans.factory.support.RootBeanDefinition;
-import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser;
-import org.springframework.beans.factory.xml.ParserContext;
-import org.springframework.integration.ConfigurationException;
-import org.springframework.integration.endpoint.SourceEndpoint;
-import org.springframework.integration.endpoint.TargetEndpoint;
-import org.springframework.integration.handler.MethodInvokingTarget;
-import org.springframework.integration.message.MethodInvokingSource;
-import org.springframework.integration.scheduling.PollingSchedule;
-import org.springframework.integration.scheduling.Schedule;
-import org.springframework.util.StringUtils;
-import org.springframework.util.xml.DomUtils;
+import org.springframework.integration.channel.config.AbstractChannelParser;
+import org.springframework.integration.channel.config.ChannelAdapterFactoryBean;
 
 /**
  * Parser for the <channel-adapter/> element.
  * 
  * @author Mark Fisher
  */
-public class ChannelAdapterParser extends AbstractSimpleBeanDefinitionParser {
+public class ChannelAdapterParser extends AbstractChannelParser {
 
 	protected final Class<?> getBeanClass(Element element) {
-		boolean hasSource = StringUtils.hasText(element.getAttribute("source"));
-		boolean hasTarget = StringUtils.hasText(element.getAttribute("target"));
-		if (!(hasSource ^ hasTarget)) {
-			throw new ConfigurationException("exactly one of 'source' or 'target' is required");
-		}
-		return hasSource ? SourceEndpoint.class : TargetEndpoint.class;
-	}
-
-	protected boolean shouldGenerateId() {
-		return false;
-	}
-
-	protected boolean shouldGenerateIdAsFallback() {
-		return true;
-	}
-
-	protected boolean isEligibleAttribute(String name) {
-		return (!"source".equals(name)
-				&& !"target".equals(name)
-				&& !"channel".equals(name)
-				&& super.isEligibleAttribute(name));
+		return ChannelAdapterFactoryBean.class;
 	}
 
 	@Override
-	protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
-		String source = element.getAttribute("source");
-		String target = element.getAttribute("target");
-		String channel = element.getAttribute("channel");
-		if (!StringUtils.hasText(channel)) {
-			throw new ConfigurationException("'channel' is required");
-		}
-		boolean isSource = StringUtils.hasText(source); 
-		if (isSource) {
-			builder.addConstructorArgReference(this.resolveConstructorArgument(
-					source, MethodInvokingSource.class, element, parserContext));
-			builder.addPropertyValue("outputChannelName", channel);
-		}
-		else {
-			builder.addConstructorArgReference(this.resolveConstructorArgument(
-					target, MethodInvokingTarget.class, element, parserContext));
-			builder.addPropertyValue("inputChannelName", channel);
-		}
-		Element scheduleElement = DomUtils.getChildElementByTagName(element, "schedule");
-		if (scheduleElement != null) {
-			builder.addPropertyValue("schedule", this.parseSchedule(scheduleElement));
-		}
-		Element pollerElement = DomUtils.getChildElementByTagName(element, "poller");
-		if (pollerElement != null) {
-			builder.addPropertyReference("poller",
-					IntegrationNamespaceUtils.parsePoller(pollerElement, parserContext));
-		}
-		Element interceptorsElement = DomUtils.getChildElementByTagName(element, "interceptors");
-		if (interceptorsElement != null) {
-			EndpointInterceptorParser parser = new EndpointInterceptorParser();
-			ManagedList interceptors = parser.parseEndpointInterceptors(interceptorsElement, parserContext);
-			builder.addPropertyValue("interceptors", interceptors);
-		}
-	}
-
-	private String resolveConstructorArgument(String ref, Class<?> targetClass, Element element, ParserContext parserContext) {
-		String method = element.getAttribute("method");
-		if (StringUtils.hasText(method)) {
-			BeanDefinition adapterDef = new RootBeanDefinition(targetClass);
-			adapterDef.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(ref));
-			adapterDef.getPropertyValues().addPropertyValue("methodName", method);
-			String adapterBeanName = parserContext.getReaderContext().generateBeanName(adapterDef);
-			parserContext.registerBeanComponent(new BeanComponentDefinition(adapterDef, adapterBeanName));
-			return adapterBeanName;
-		}
-		return ref;
-	}
-
-	/**
-	 * Subclasses may override this method to control the creation of the {@link Schedule}. The default
-	 * implementation creates a {@link PollingSchedule} instance based on the provided "period" attribute. 
-	 */
-	protected Schedule parseSchedule(Element element) {
-		String period = element.getAttribute("period");
-		if (!StringUtils.hasText(period)) {
-			throw new ConfigurationException("The 'period' attribute is required for the 'schedule' element.");
-		}
-		PollingSchedule schedule = new PollingSchedule(Long.valueOf(period));
-		return schedule;
+	protected void postProcess(BeanDefinitionBuilder builder, Element element) {
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "source");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "target");
 	}
 
 }

org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelInterceptorParser.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2002-2008 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.config;
+
+import java.util.Map;
+
+/**
+ * @author Mark Fisher
+ */
+public class ChannelInterceptorParser extends AbstractInterceptorParser {
+
+	@Override
+	protected Map<String, BeanDefinitionRegisteringParser> getParserMap() {
+		return null;
+	}
+
+}

org.springframework.integration/src/main/java/org/springframework/integration/config/PublishSubscribeChannelParser.java

@@ -36,7 +36,7 @@ public class PublishSubscribeChannelParser extends AbstractChannelParser {
 	}
 
 	@Override
-	protected void configureConstructorArgs(BeanDefinitionBuilder builder, Element element) {
+	protected void postProcess(BeanDefinitionBuilder builder, Element element) {
 		String taskExecutorRef = element.getAttribute("task-executor");
 		if (StringUtils.hasText(taskExecutorRef)) {
 			builder.addConstructorArgReference(taskExecutorRef);

org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd

@@ -25,7 +25,11 @@
 				</xsd:documentation>
 			</xsd:annotation>
 			<xsd:sequence>
-				<xsd:element ref="interceptor" minOccurs="0" maxOccurs="unbounded"/>
+				<xsd:element name="interceptor" minOccurs="0" maxOccurs="unbounded">
+					<xsd:complexType>
+						<xsd:attribute name="ref" type="xsd:string" use="required"/>
+					</xsd:complexType>
+				</xsd:element>
 			</xsd:sequence>
 			<xsd:attribute name="auto-startup" type="xsd:boolean"/>
 			<xsd:attribute name="auto-create-channels" type="xsd:boolean"/>
@@ -126,23 +130,12 @@
 			</xsd:documentation>
 		</xsd:annotation>
 		<xsd:sequence>
-			<xsd:element ref="interceptor" minOccurs="0" maxOccurs="unbounded"/>
+			<xsd:element name="interceptors" type="channelInterceptorsType" minOccurs="0" maxOccurs="1"/>
 		</xsd:sequence>
 		<xsd:attribute name="id" type="xsd:ID" use="required"/>
 		<xsd:attribute name="datatype" type="xsd:string"/>
 	</xsd:complexType>
 
-	<xsd:element name="interceptor">
-		<xsd:complexType>
-			<xsd:annotation>
-				<xsd:documentation>
-	Provides a channel interceptor reference.
-				</xsd:documentation>
-			</xsd:annotation>
-			<xsd:attribute name="ref" type="xsd:string" use="required"/>
-		</xsd:complexType>
-	</xsd:element>
-
 	<xsd:element name="gateway">
 		<xsd:complexType>
 			<xsd:annotation>
@@ -171,16 +164,9 @@
 				</xsd:documentation>
 			</xsd:annotation>
 			<xsd:complexContent>
-				<xsd:extension base="beans:identifiedType">
-					<xsd:all>
-						<xsd:element ref="schedule" minOccurs="0" maxOccurs="1"/>
-						<xsd:element ref="poller" minOccurs="0" maxOccurs="1"/>
-						<xsd:element name="interceptors" type="interceptorsType" minOccurs="0" maxOccurs="1"/>
-					</xsd:all>
+				<xsd:extension base="channelType">
 					<xsd:attribute name="source" type="xsd:string"/>
 					<xsd:attribute name="target" type="xsd:string"/>
-					<xsd:attribute name="method" type="xsd:string"/>
-					<xsd:attribute name="channel" type="xsd:string" use="required"/>
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -223,7 +209,7 @@
 				<xsd:all>
 					<xsd:element ref="schedule" minOccurs="0" maxOccurs="1"/>
 					<xsd:element ref="poller" minOccurs="0" maxOccurs="1"/>
-					<xsd:element name="interceptors" type="interceptorsType" minOccurs="0" maxOccurs="1"/>
+					<xsd:element name="interceptors" type="endpointInterceptorsType" minOccurs="0" maxOccurs="1"/>
 				</xsd:all>
 				<xsd:attribute name="ref" type="xsd:string" use="required"/>
 				<xsd:attribute name="method" type="xsd:string"/>
@@ -428,10 +414,28 @@
 		</xsd:complexType>
 	</xsd:element>
 
-	<xsd:complexType name="interceptorsType">
+	<xsd:complexType name="channelInterceptorsType">
 		<xsd:annotation>
 			<xsd:documentation>
-	Defines a list of interceptors. Each element may be an EndpointInterceptor or any Advice instance.
+	Defines a list of interceptors. Each element may be a ChannelInterceptor, ref, or inner-bean.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:sequence>
+			<xsd:choice minOccurs="0" maxOccurs="unbounded">
+				<xsd:element name="ref" minOccurs="0" maxOccurs="unbounded">
+					<xsd:complexType>
+						<xsd:attribute name="bean" use="required"/>
+					</xsd:complexType>
+				</xsd:element>
+				<xsd:any namespace="##other" processContents="strict" minOccurs="0" maxOccurs="unbounded"/>
+			</xsd:choice>
+		</xsd:sequence>
+	</xsd:complexType>
+
+	<xsd:complexType name="endpointInterceptorsType">
+		<xsd:annotation>
+			<xsd:documentation>
+	Defines a list of interceptors. Each element may be an EndpointInterceptor, ref, or inner-bean.
 			</xsd:documentation>
 		</xsd:annotation>
 		<xsd:sequence>
@@ -443,7 +447,7 @@
 				</xsd:element>
 				<xsd:element name="transaction-interceptor" type="transactionalType" minOccurs="0" maxOccurs="1"/>
 				<xsd:element name="concurrency-interceptor" type="concurrencyInterceptorType" minOccurs="0" maxOccurs="1"/>
-				<xsd:any namespace="##other" processContents="strict" minOccurs="0" maxOccurs="unbounded"/>				
+				<xsd:any namespace="##other" processContents="strict" minOccurs="0" maxOccurs="unbounded"/>
 			</xsd:choice>
 		</xsd:sequence>
 	</xsd:complexType>