diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/jms/config/jmsSourceEndpoint.xml b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/jms/config/jmsSourceEndpoint.xml index 5d22dd696b..89be9fd48c 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/jms/config/jmsSourceEndpoint.xml +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/jms/config/jmsSourceEndpoint.xml @@ -9,11 +9,7 @@ - - - - - + diff --git a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/filecopy/fileCopyDemo-common.xml b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/filecopy/fileCopyDemo-common.xml index 8e5ca6b3b5..f7aaf1918c 100644 --- a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/filecopy/fileCopyDemo-common.xml +++ b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/filecopy/fileCopyDemo-common.xml @@ -11,18 +11,15 @@ - + - + + + - - - - - - - + diff --git a/org.springframework.integration.security/src/main/java/org/springframework/integration/security/channel/SecurityEnforcingChannelInterceptor.java b/org.springframework.integration.security/src/main/java/org/springframework/integration/security/channel/SecurityEnforcingChannelInterceptor.java index e006916805..eb84abb4ae 100644 --- a/org.springframework.integration.security/src/main/java/org/springframework/integration/security/channel/SecurityEnforcingChannelInterceptor.java +++ b/org.springframework.integration.security/src/main/java/org/springframework/integration/security/channel/SecurityEnforcingChannelInterceptor.java @@ -20,6 +20,8 @@ import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter; import org.springframework.integration.message.Message; import org.springframework.security.AccessDecisionManager; +import org.springframework.security.Authentication; +import org.springframework.security.AuthenticationCredentialsNotFoundException; import org.springframework.security.ConfigAttributeDefinition; import org.springframework.security.context.SecurityContextHolder; import org.springframework.util.Assert; @@ -29,6 +31,7 @@ import org.springframework.util.Assert; * enforce the security on the send and receive calls of the {@link MessageChannel}. * * @author Jonas Partner + * @author Mark Fisher */ public class SecurityEnforcingChannelInterceptor extends ChannelInterceptorAdapter { @@ -83,8 +86,12 @@ public class SecurityEnforcingChannelInterceptor extends ChannelInterceptorAdapt private void checkPermission(MessageChannel messageChannel, ConfigAttributeDefinition securityAttributes) { if (securityAttributes != null) { - this.accessDecisionManger.decide(SecurityContextHolder.getContext().getAuthentication(), messageChannel, - securityAttributes); + Authentication authentication = SecurityContextHolder.getContext().getAuthentication(); + if (authentication == null) { + throw new AuthenticationCredentialsNotFoundException( + "No Authentication object available. Consider enabling the SecurityPropagatingBeanPostProcessor."); + } + this.accessDecisionManger.decide(authentication, messageChannel, securityAttributes); } } diff --git a/org.springframework.integration.security/src/test/java/org/springframework/integration/security/channel/ChannelAdapterSecurityIntegrationTests-context.xml b/org.springframework.integration.security/src/test/java/org/springframework/integration/security/channel/ChannelAdapterSecurityIntegrationTests-context.xml new file mode 100644 index 0000000000..57a73f4514 --- /dev/null +++ b/org.springframework.integration.security/src/test/java/org/springframework/integration/security/channel/ChannelAdapterSecurityIntegrationTests-context.xml @@ -0,0 +1,29 @@ + + + + + + + + + secured.* + + + + + + + + + \ No newline at end of file diff --git a/org.springframework.integration.security/src/test/java/org/springframework/integration/security/channel/ChannelAdapterSecurityIntegrationTests.java b/org.springframework.integration.security/src/test/java/org/springframework/integration/security/channel/ChannelAdapterSecurityIntegrationTests.java new file mode 100644 index 0000000000..3e77c332aa --- /dev/null +++ b/org.springframework.integration.security/src/test/java/org/springframework/integration/security/channel/ChannelAdapterSecurityIntegrationTests.java @@ -0,0 +1,112 @@ +/* + * 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.security.channel; + +import static org.junit.Assert.assertEquals; + +import org.junit.After; +import org.junit.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.StringMessage; +import org.springframework.integration.security.SecurityTestUtil; +import org.springframework.integration.security.endpoint.TestTarget; +import org.springframework.security.AccessDeniedException; +import org.springframework.security.AuthenticationException; +import org.springframework.security.context.SecurityContext; +import org.springframework.security.context.SecurityContextHolder; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests; + +/** + * @author Mark Fisher + */ +@ContextConfiguration +public class ChannelAdapterSecurityIntegrationTests extends AbstractJUnit4SpringContextTests { + + @Autowired + @Qualifier("securedChannelAdapter") + MessageChannel securedChannelAdapter; + + @Autowired + @Qualifier("unsecuredChannelAdapter") + MessageChannel unsecuredChannelAdapter; + + @Autowired + TestTarget testTarget; + + + @After + public void tearDown() { + SecurityContextHolder.clearContext(); + } + + + @Test + @DirtiesContext + public void testSecuredWithPermission() { + login("bob", "bobspassword", "ROLE_ADMIN"); + securedChannelAdapter.send(new StringMessage("test")); + assertEquals("Wrong size of message list in target", 1, testTarget.sentMessages.size()); + } + + @Test(expected = AccessDeniedException.class) + @DirtiesContext + public void testSecuredWithoutPermision() { + login("bob", "bobspassword", "ROLE_USER"); + securedChannelAdapter.send(new StringMessage("test")); + } + + @Test(expected = AuthenticationException.class) + @DirtiesContext + public void testSecuredWithoutAuthenticating() { + securedChannelAdapter.send(new StringMessage("test")); + } + + @Test + @DirtiesContext + public void testUnsecuredAsAdmin() { + login("bob", "bobspassword", "ROLE_ADMIN"); + unsecuredChannelAdapter.send(new StringMessage("test")); + assertEquals("Wrong size of message list in target", 1, testTarget.sentMessages.size()); + } + + @Test + @DirtiesContext + public void testUnsecuredAsUser() { + login("bob", "bobspassword", "ROLE_USER"); + unsecuredChannelAdapter.send(new StringMessage("test")); + assertEquals("Wrong size of message list in target", 1, testTarget.sentMessages.size()); + } + + @Test + @DirtiesContext + public void testUnsecuredWithoutAuthenticating() { + unsecuredChannelAdapter.send(new StringMessage("test")); + assertEquals("Wrong size of message list in target", 1, testTarget.sentMessages.size()); + } + + + private void login(String username, String password, String... roles) { + SecurityContext context = SecurityTestUtil.createContext(username, password, roles); + SecurityContextHolder.setContext(context); + } + +} diff --git a/org.springframework.integration.security/src/test/java/org/springframework/integration/security/channel/SecurityEnforcingChannelInterceptorTests.java b/org.springframework.integration.security/src/test/java/org/springframework/integration/security/channel/SecurityEnforcingChannelInterceptorTests.java index d3d8885e00..48a3c94f4a 100644 --- a/org.springframework.integration.security/src/test/java/org/springframework/integration/security/channel/SecurityEnforcingChannelInterceptorTests.java +++ b/org.springframework.integration.security/src/test/java/org/springframework/integration/security/channel/SecurityEnforcingChannelInterceptorTests.java @@ -30,8 +30,10 @@ import org.springframework.security.AccessDeniedException; import org.springframework.security.Authentication; import org.springframework.security.ConfigAttribute; import org.springframework.security.ConfigAttributeDefinition; +import org.springframework.security.GrantedAuthorityImpl; import org.springframework.security.InsufficientAuthenticationException; import org.springframework.security.context.SecurityContextHolder; +import org.springframework.security.providers.TestingAuthenticationToken; /** * @author Jonas Partner @@ -46,8 +48,10 @@ public class SecurityEnforcingChannelInterceptorTests { @Before public void setUp() { channel = new QueueChannel(); + SecurityContextHolder.getContext().setAuthentication(new TestingAuthenticationToken( + "stub", "passwd", new GrantedAuthorityImpl[] {})); } - + @After public void clearSecurityContext(){ SecurityContextHolder.clearContext(); diff --git a/org.springframework.integration.security/src/test/java/org/springframework/integration/security/config/SecuredChannelsParserTests-context.xml b/org.springframework.integration.security/src/test/java/org/springframework/integration/security/config/SecuredChannelsParserTests-context.xml index d527711e00..ab678e83c0 100644 --- a/org.springframework.integration.security/src/test/java/org/springframework/integration/security/config/SecuredChannelsParserTests-context.xml +++ b/org.springframework.integration.security/src/test/java/org/springframework/integration/security/config/SecuredChannelsParserTests-context.xml @@ -11,33 +11,26 @@ http://www.springframework.org/schema/integration-security http://www.springframework.org/schema/integration/spring-integration-security-1.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd"> - adminRequiredForSend - + adminOrUserRequiredForSend - + adminRequiredForReceive - + adminOrUserRequiredForReceive - + adminForSendAndReceive - - - - - - - + \ No newline at end of file diff --git a/org.springframework.integration.security/src/test/java/org/springframework/integration/security/endpoint/EndpointSecurityIntegrationTest-context.xml b/org.springframework.integration.security/src/test/java/org/springframework/integration/security/endpoint/EndpointSecurityIntegrationTest-context.xml index 8ea896d13f..384bd48ccd 100644 --- a/org.springframework.integration.security/src/test/java/org/springframework/integration/security/endpoint/EndpointSecurityIntegrationTest-context.xml +++ b/org.springframework.integration.security/src/test/java/org/springframework/integration/security/endpoint/EndpointSecurityIntegrationTest-context.xml @@ -13,17 +13,24 @@ - + - - - + - + + + + + + diff --git a/org.springframework.integration.security/src/test/java/org/springframework/integration/security/endpoint/EndpointSecurityIntegrationTest.java b/org.springframework.integration.security/src/test/java/org/springframework/integration/security/endpoint/EndpointSecurityIntegrationTest.java index 22c7c694a9..654df01168 100644 --- a/org.springframework.integration.security/src/test/java/org/springframework/integration/security/endpoint/EndpointSecurityIntegrationTest.java +++ b/org.springframework.integration.security/src/test/java/org/springframework/integration/security/endpoint/EndpointSecurityIntegrationTest.java @@ -13,13 +13,16 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.security.endpoint; import static org.junit.Assert.assertEquals; import org.junit.After; import org.junit.Test; + import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.StringMessage; import org.springframework.integration.security.SecurityTestUtil; @@ -30,15 +33,24 @@ import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests; +/** + * @author Jonas Partner + * @author Mark Fisher + */ @ContextConfiguration public class EndpointSecurityIntegrationTest extends AbstractJUnit4SpringContextTests { @Autowired - MessageChannel channel; + @Qualifier("input") + MessageChannel input; + + @Autowired + TestHandler testHandler; @Autowired TestTarget testTarget; + @After public void tearDown() { SecurityContextHolder.clearContext(); @@ -48,22 +60,20 @@ public class EndpointSecurityIntegrationTest extends AbstractJUnit4SpringContext @DirtiesContext public void testWithPermision() { login("bob", "bobspassword", "ROLE_ADMIN"); - channel.send(new StringMessage("test")); - assertEquals("Wrong size of message list in target ", 1, testTarget.sentMessages.size()); + input.send(new StringMessage("test")); + assertEquals("Wrong size of message list in handler", 1, testHandler.sentMessages.size()); + assertEquals("Wrong size of message list in target", 1, testTarget.sentMessages.size()); } - /** - * - */ @Test(expected = AccessDeniedException.class) @DirtiesContext public void testWithoutPermision() { login("bob", "bobspassword", "ROLE_USER"); - channel.send(new StringMessage("test")); - assertEquals("Wrong size of message list in target ", 1, testTarget.sentMessages.size()); + input.send(new StringMessage("test")); } - public void login(String username, String password, String... roles) { + + private void login(String username, String password, String... roles) { SecurityContext context = SecurityTestUtil.createContext(username, password, roles); SecurityContextHolder.setContext(context); } diff --git a/org.springframework.integration.security/src/test/java/org/springframework/integration/security/endpoint/TestHandler.java b/org.springframework.integration.security/src/test/java/org/springframework/integration/security/endpoint/TestHandler.java new file mode 100644 index 0000000000..d00a7e2b06 --- /dev/null +++ b/org.springframework.integration.security/src/test/java/org/springframework/integration/security/endpoint/TestHandler.java @@ -0,0 +1,37 @@ +/* + * 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.security.endpoint; + +import java.util.ArrayList; +import java.util.List; + +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.message.Message; + +/** + * @author Mark Fisher + */ +public class TestHandler implements MessageHandler { + + public List> sentMessages = new ArrayList>(); + + public Message handle(Message message) { + sentMessages.add(message); + return message; + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractChannelAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractChannelAdapter.java new file mode 100644 index 0000000000..d9523a7fa0 --- /dev/null +++ b/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); + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/PollableChannelAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/PollableChannelAdapter.java new file mode 100644 index 0000000000..aa892ea62c --- /dev/null +++ b/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> clear() { + if (this.source != null && this.source instanceof PollableChannel) { + return ((PollableChannel) this.source).clear(); + } + return null; + } + + public List> 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); + } + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/SubscribableChannelAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/SubscribableChannelAdapter.java new file mode 100644 index 0000000000..429a4c402b --- /dev/null +++ b/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); + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/config/AbstractChannelParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/config/AbstractChannelParser.java index 819a88fac3..058feb2bff 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/config/AbstractChannelParser.java +++ b/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); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/config/ChannelAdapterFactoryBean.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/config/ChannelAdapterFactoryBean.java new file mode 100644 index 0000000000..c9aa56f493 --- /dev/null +++ b/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 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 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); + } + } + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/config/PriorityChannelParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/config/PriorityChannelParser.java index 539cb01329..b928f98d5f 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/config/PriorityChannelParser.java +++ b/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); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/config/QueueChannelParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/config/QueueChannelParser.java index 3be4fc14b7..ce87a2425b 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/config/QueueChannelParser.java +++ b/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; diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractHandlerEndpointParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractHandlerEndpointParser.java index 3b999084ad..143eb61f98 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractHandlerEndpointParser.java +++ b/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); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractInterceptorParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractInterceptorParser.java new file mode 100644 index 0000000000..135064795b --- /dev/null +++ b/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 interceptors element. + * + * @author Mark Fisher + */ +public abstract class AbstractInterceptorParser { + + private final Map parsers = new HashMap(); + + private volatile boolean initialized; + + private final Object initializationMonitor = new Object(); + + + protected abstract Map getParserMap(); + + private void initializeParserMap() { + synchronized (this.initializationMonitor) { + if (!this.initialized) { + Map 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; + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java index 8fe83e3de8..0eac754573 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java +++ b/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 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"); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelInterceptorParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelInterceptorParser.java new file mode 100644 index 0000000000..508f6d7118 --- /dev/null +++ b/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 getParserMap() { + return null; + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/PublishSubscribeChannelParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/PublishSubscribeChannelParser.java index 7900e48d6a..6ed63cc98e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/PublishSubscribeChannelParser.java +++ b/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); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd index 4f5c0b544a..babfdfa896 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd @@ -25,7 +25,11 @@ - + + + + + @@ -126,23 +130,12 @@ - + - - - - - Provides a channel interceptor reference. - - - - - - @@ -171,16 +164,9 @@ - - - - - - + - - @@ -223,7 +209,7 @@ - + @@ -428,10 +414,28 @@ - + - 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. + + + + + + + + + + + + + + + + + + Defines a list of interceptors. Each element may be an EndpointInterceptor, ref, or inner-bean. @@ -443,7 +447,7 @@ - + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelAdapterFactoryBeanContextTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelAdapterFactoryBeanContextTests.java new file mode 100644 index 0000000000..abad2054b5 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelAdapterFactoryBeanContextTests.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.config; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import org.junit.Test; + +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PollableChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class ChannelAdapterFactoryBeanContextTests { + + @Test + public void testPollableSourceWithoutTarget() throws Exception { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "channelAdapterFactoryBeanTests.xml", this.getClass()); + MessageChannel channel = (MessageChannel) + context.getBean("adapterWithPollableSourceAndNoTarget"); + assertTrue(channel instanceof PollableChannel); + assertEquals("adapterWithPollableSourceAndNoTarget", channel.getName()); + Message reply = ((PollableChannel) channel).receive(); + assertNotNull(reply); + assertEquals("test", reply.getPayload()); + assertFalse(channel.send(new StringMessage("no target"))); + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelAdapterFactoryBeanTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelAdapterFactoryBeanTests.java new file mode 100644 index 0000000000..8de12aa7c1 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelAdapterFactoryBeanTests.java @@ -0,0 +1,99 @@ +/* + * 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.Test; + +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PollableChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.message.StringMessage; +import org.springframework.integration.message.SubscribableSource; + +/** + * @author Mark Fisher + */ +public class ChannelAdapterFactoryBeanTests { + + @Test + public void testPollableSourceWithoutTarget() throws Exception { + ChannelAdapterFactoryBean factoryBean = new ChannelAdapterFactoryBean(); + factoryBean.setBeanName("testChannel"); + factoryBean.setSource(new TestPollableSource()); + Object bean = factoryBean.getObject(); + assertTrue(bean instanceof PollableChannel); + PollableChannel channel = (PollableChannel) bean; + Message reply = channel.receive(); + assertNotNull(reply); + assertEquals("test", reply.getPayload()); + assertFalse(channel.send(new StringMessage("no target"))); + } + + @Test + public void testSubscribableSourceWithoutTarget() throws Exception { + ChannelAdapterFactoryBean factoryBean = new ChannelAdapterFactoryBean(); + factoryBean.setBeanName("testChannel"); + TestSubscribableSource source = new TestSubscribableSource(); + factoryBean.setSource(source); + Object bean = factoryBean.getObject(); + assertTrue(bean instanceof MessageChannel); + assertTrue(bean instanceof SubscribableSource); + MessageChannel channel = (MessageChannel) bean; + final AtomicReference> messageReference = new AtomicReference>(); + ((SubscribableSource) bean).subscribe(new MessageTarget() { + public boolean send(Message message) { + messageReference.set(message); + return true; + } + }); + source.publishMessage(new StringMessage("foo")); + assertNotNull(messageReference.get()); + assertEquals("foo", (messageReference.get()).getPayload()); + assertFalse(channel.send(new StringMessage("no target"))); + } + + @Test + public void testTargetOnly() throws Exception { + ChannelAdapterFactoryBean factoryBean = new ChannelAdapterFactoryBean(); + factoryBean.setBeanName("testChannel"); + final AtomicReference> messageRef = new AtomicReference>(); + factoryBean.setTarget(new MessageTarget() { + public boolean send(Message message) { + messageRef.set(message); + return true; + } + }); + Object bean = factoryBean.getObject(); + assertTrue(bean instanceof PollableChannel); + PollableChannel channel = (PollableChannel) bean; + Message reply = channel.receive(0); + assertNull(reply); + assertTrue(channel.send(new StringMessage("hello"))); + assertNotNull(messageRef.get()); + assertEquals("hello", messageRef.get().getPayload()); + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java index b50b72cb34..5b36c9122d 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java @@ -138,10 +138,10 @@ public class ChannelParserTests { } @Test - public void testChannelInteceptors() { + public void testChannelInteceptorRef() { ApplicationContext context = new ClassPathXmlApplicationContext( "channelInterceptorParserTests.xml", this.getClass()); - PollableChannel channel = (PollableChannel) context.getBean("channel"); + PollableChannel channel = (PollableChannel) context.getBean("channelWithInterceptorRef"); TestChannelInterceptor interceptor = (TestChannelInterceptor) context.getBean("interceptor"); assertEquals(0, interceptor.getSendCount()); channel.send(new StringMessage("test")); @@ -151,6 +151,16 @@ public class ChannelParserTests { assertEquals(1, interceptor.getReceiveCount()); } + @Test + public void testChannelInteceptorInnerBean() { + ApplicationContext context = new ClassPathXmlApplicationContext( + "channelInterceptorParserTests.xml", this.getClass()); + PollableChannel channel = (PollableChannel) context.getBean("channelWithInterceptorInnerBean"); + channel.send(new StringMessage("test")); + Message transformed = channel.receive(1000); + assertEquals("TEST", transformed.getPayload()); + } + @Test public void testPriorityChannelWithDefaultComparator() { ApplicationContext context = new ClassPathXmlApplicationContext( @@ -215,6 +225,7 @@ public class ChannelParserTests { } + @SuppressWarnings("unchecked") public static MessageChannel extractProxifiedChannel (Object channelProxy) { InvocationHandler handler = Proxy.getInvocationHandler(channelProxy); DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(handler); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/TestPollableSource.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/TestPollableSource.java new file mode 100644 index 0000000000..08f5317940 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/TestPollableSource.java @@ -0,0 +1,32 @@ +/* + * 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 org.springframework.integration.message.Message; +import org.springframework.integration.message.PollableSource; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class TestPollableSource implements PollableSource { + + public Message receive() { + return new StringMessage("test"); + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/TestSubscribableSource.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/TestSubscribableSource.java new file mode 100644 index 0000000000..8950ff6822 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/TestSubscribableSource.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.config; + +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.message.SubscribableSource; + +/** + * @author Mark Fisher + */ +public class TestSubscribableSource implements SubscribableSource { + + private final List targets = new CopyOnWriteArrayList(); + + + public boolean subscribe(MessageTarget target) { + return this.targets.add(target); + } + + public boolean unsubscribe(MessageTarget target) { + return this.targets.remove(target); + } + + public void publishMessage(Message message) { + for (MessageTarget target : this.targets) { + target.send(message); + } + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/TestTarget.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/TestTarget.java new file mode 100644 index 0000000000..e5d4c8fac2 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/TestTarget.java @@ -0,0 +1,42 @@ +/* + * 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.ArrayList; +import java.util.List; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageTarget; + +/** + * @author Mark Fisher + */ +public class TestTarget implements MessageTarget { + + private final List> receivedMessages = new ArrayList>(); + + + public List> getReceivedMessages() { + return this.receivedMessages; + } + + public boolean send(Message message) { + this.receivedMessages.add(message); + return true; + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/TestTransformer.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/TestTransformer.java new file mode 100644 index 0000000000..974870b46e --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/TestTransformer.java @@ -0,0 +1,32 @@ +/* + * 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 org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageBuilder; + +/** + * @author Mark Fisher + */ +public class TestTransformer implements MessageHandler { + + public Message handle(Message message) { + return MessageBuilder.fromPayload(message.getPayload().toString().toUpperCase()).build(); + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/channelAdapterFactoryBeanTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/channelAdapterFactoryBeanTests.xml new file mode 100644 index 0000000000..550a1ef223 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/channelAdapterFactoryBeanTests.xml @@ -0,0 +1,23 @@ + + + + + + + + + + + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/channelInterceptorParserTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/channelInterceptorParserTests.xml index 85334c4bc3..2f8ac81685 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/channelInterceptorParserTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/channelInterceptorParserTests.xml @@ -7,8 +7,20 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd"> - - + + + + + + + + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests-context.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests-context.xml new file mode 100644 index 0000000000..da8b1bd3bc --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests-context.xml @@ -0,0 +1,18 @@ + + + + + + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java new file mode 100644 index 0000000000..8a1a0398df --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java @@ -0,0 +1,60 @@ +/* + * 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 static org.junit.Assert.assertTrue; + +import org.junit.Test; + +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PollableChannel; +import org.springframework.integration.channel.PollableChannelAdapter; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.MessageDeliveryException; +import org.springframework.integration.message.StringMessage; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests; + +/** + * @author Mark Fisher + */ +@ContextConfiguration +public class ChannelAdapterParserTests extends AbstractJUnit4SpringContextTests { + + @Test + public void targetOnly() { + Object bean = this.applicationContext.getBean("targetOnly"); + assertTrue(bean instanceof MessageChannel); + assertTrue(bean instanceof PollableChannel); + assertTrue(bean instanceof PollableChannelAdapter); + MessageChannel channel = (MessageChannel) bean; + assertTrue(channel.send(new StringMessage("test"))); + } + + @Test + public void targetWithDatatypeAccepts() { + MessageChannel channel = (MessageChannel) this.applicationContext.getBean("targetWithDatatype"); + assertTrue(channel.send(new StringMessage("test"))); + } + + @Test(expected = MessageDeliveryException.class) + public void targetWithDatatypeRejects() { + MessageChannel channel = (MessageChannel) this.applicationContext.getBean("targetWithDatatype"); + channel.send(new GenericMessage(123)); + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/EndpointInterceptorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/EndpointInterceptorTests.java index d940fbd409..cf332145ea 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/EndpointInterceptorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/EndpointInterceptorTests.java @@ -21,13 +21,11 @@ import static org.junit.Assert.assertEquals; import java.util.List; import org.junit.Test; + import org.springframework.beans.DirectFieldAccessor; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.endpoint.EndpointInterceptor; import org.springframework.integration.endpoint.MessageEndpoint; -import org.springframework.integration.endpoint.SourceEndpoint; -import org.springframework.integration.endpoint.TriggerMessage; import org.springframework.integration.message.StringMessage; /** @@ -39,7 +37,7 @@ public class EndpointInterceptorTests { public void testHandlerEndpointWithBeanInterceptors() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "endpointInterceptorTests.xml", this.getClass()); - MessageEndpoint endpoint = (MessageEndpoint) context.getBean("handlerEndpointWithBeanInterceptors"); + MessageEndpoint endpoint = (MessageEndpoint) context.getBean("endpointWithBeanInterceptors"); testInterceptors(endpoint, context, true); } @@ -47,39 +45,7 @@ public class EndpointInterceptorTests { public void testHandlerEndpointWithRefInterceptors() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "endpointInterceptorTests.xml", this.getClass()); - MessageEndpoint endpoint = (MessageEndpoint) context.getBean("handlerEndpointWithRefInterceptors"); - testInterceptors(endpoint, context, false); - } - - @Test - public void testTargetEndpointWithBeanInterceptors() { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "endpointInterceptorTests.xml", this.getClass()); - MessageEndpoint endpoint = (MessageEndpoint) context.getBean("targetEndpointWithBeanInterceptors"); - testInterceptors(endpoint, context, true); - } - - @Test - public void testTargetEndpointWithRefInterceptors() { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "endpointInterceptorTests.xml", this.getClass()); - MessageEndpoint endpoint = (MessageEndpoint) context.getBean("targetEndpointWithRefInterceptors"); - testInterceptors(endpoint, context, false); - } - - @Test - public void testSourceEndpointWithBeanInterceptors() { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "endpointInterceptorTests.xml", this.getClass()); - MessageEndpoint endpoint = (MessageEndpoint) context.getBean("sourceEndpointWithBeanInterceptors"); - testInterceptors(endpoint, context, true); - } - - @Test - public void testSourceEndpointWithRefInterceptors() { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "endpointInterceptorTests.xml", this.getClass()); - MessageEndpoint endpoint = (MessageEndpoint) context.getBean("sourceEndpointWithRefInterceptors"); + MessageEndpoint endpoint = (MessageEndpoint) context.getBean("endpointWithRefInterceptors"); testInterceptors(endpoint, context, false); } @@ -100,14 +66,7 @@ public class EndpointInterceptorTests { } assertEquals(0, preInterceptor.getCount()); assertEquals(0, aroundInterceptor.getCount()); - if (endpoint instanceof SourceEndpoint) { - MessageChannel channel = (MessageChannel) context.getBean("testChannel"); - channel.send(new StringMessage("foo")); - endpoint.send(new TriggerMessage()); - } - else { - endpoint.send(new StringMessage("test")); - } + endpoint.send(new StringMessage("test")); assertEquals(1, preInterceptor.getCount()); assertEquals(2, aroundInterceptor.getCount()); context.stop(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/endpointInterceptorTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/endpointInterceptorTests.xml index 00cd10b5d6..876cfa3af1 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/endpointInterceptorTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/endpointInterceptorTests.xml @@ -13,7 +13,7 @@ - @@ -24,7 +24,7 @@ - @@ -35,52 +35,9 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/gateway/gatewayWithResponseCorrelator.xml b/org.springframework.integration/src/test/java/org/springframework/integration/gateway/gatewayWithResponseCorrelator.xml index 50283b97fa..cc0469e3e1 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/gateway/gatewayWithResponseCorrelator.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/gateway/gatewayWithResponseCorrelator.xml @@ -12,7 +12,9 @@ - + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/handler/MethodInvokingAdapterTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/handler/MethodInvokingAdapterTests.java index bf524ab84e..a302b63dcb 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/handler/MethodInvokingAdapterTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/handler/MethodInvokingAdapterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -21,13 +21,15 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import java.io.IOException; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; import org.junit.Test; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PollableChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; /** * @author Mark Fisher @@ -35,31 +37,24 @@ import org.springframework.context.support.ClassPathXmlApplicationContext; public class MethodInvokingAdapterTests { @Test - public void testAdaptersWithBeanDefinitions() throws IOException, InterruptedException { + public void testMethodInvokingSourceChannelAdapter() throws IOException, InterruptedException { AbstractApplicationContext context = new ClassPathXmlApplicationContext("adapterTests.xml", this.getClass()); - TestSink sink = (TestSink) context.getBean("sink"); - CountDownLatch latch = new CountDownLatch(1); - sink.setLatch(latch); - assertNull(sink.get()); - context.start(); - latch.await(3000, TimeUnit.MILLISECONDS); - assertEquals("latch should have counted down within allotted time", 0, latch.getCount()); - assertNotNull(sink.get()); - context.close(); + PollableChannel channel = (PollableChannel) context.getBean("sourceChannelAdapter"); + Message message = channel.receive(1000); + assertNotNull(message); + assertEquals("foo", message.getPayload()); } @Test - public void testAdaptersWithNamespace() throws IOException, InterruptedException { - AbstractApplicationContext context = new ClassPathXmlApplicationContext("adapterTestsWithNamespace.xml", this.getClass()); + public void testMethodInvokingTargetChannelAdapter() throws IOException, InterruptedException { + AbstractApplicationContext context = new ClassPathXmlApplicationContext("adapterTests.xml", this.getClass()); + MessageChannel channel = (MessageChannel) context.getBean("targetChannelAdapter"); TestSink sink = (TestSink) context.getBean("sink"); - CountDownLatch latch = new CountDownLatch(1); - sink.setLatch(latch); assertNull(sink.get()); - context.start(); - latch.await(3000, TimeUnit.MILLISECONDS); - assertEquals("latch should have counted down within allotted time", 0, latch.getCount()); - assertNotNull(sink.get()); - context.close(); + channel.send(new StringMessage("foo")); + String result = sink.get(); + assertNotNull(result); + assertEquals("foo", result); } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/handler/TestSink.java b/org.springframework.integration/src/test/java/org/springframework/integration/handler/TestSink.java index 85fe8030ab..ae65ac84ff 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/handler/TestSink.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/handler/TestSink.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -16,8 +16,6 @@ package org.springframework.integration.handler; -import java.util.concurrent.CountDownLatch; - /** * @author Mark Fisher */ @@ -25,12 +23,6 @@ public class TestSink { private String result; - private CountDownLatch latch; - - - public void setLatch(CountDownLatch latch) { - this.latch = latch; - } public void validMethod(String s) { } @@ -44,9 +36,6 @@ public class TestSink { public void store(String s) { this.result = s; - if (this.latch != null) { - this.latch.countDown(); - } } public String get() { diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/handler/TestSource.java b/org.springframework.integration/src/test/java/org/springframework/integration/handler/TestSource.java index f77d7c95bf..41d4bec4ac 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/handler/TestSource.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/handler/TestSource.java @@ -21,8 +21,6 @@ package org.springframework.integration.handler; */ public class TestSource { - private boolean fooCalled; - public String validMethod() { return "valid"; } @@ -35,16 +33,7 @@ public class TestSource { } public String foo() { - if (this.fooCalled) { - try { - Thread.sleep(5000); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - } - this.fooCalled = true; - return "bar"; + return "foo"; } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/handler/adapterTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/handler/adapterTests.xml index 24cb2973d0..14a1a45c5d 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/handler/adapterTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/handler/adapterTests.xml @@ -4,37 +4,24 @@ xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd"> - - + + - - - - - - - - - - - - - - - - + + + + + + + + - - - - -