diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java new file mode 100644 index 0000000000..8c9a4cbadc --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java @@ -0,0 +1,167 @@ +/* + * 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.gateway; + +import org.aopalliance.intercept.MethodInterceptor; +import org.aopalliance.intercept.MethodInvocation; + +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanClassLoaderAware; +import org.springframework.beans.factory.FactoryBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.integration.ConfigurationException; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.RendezvousChannel; +import org.springframework.integration.config.MessageBusParser; +import org.springframework.integration.endpoint.EndpointRegistry; +import org.springframework.integration.endpoint.HandlerEndpoint; +import org.springframework.integration.handler.ResponseCorrelator; +import org.springframework.integration.message.DefaultMessageCreator; +import org.springframework.integration.message.DefaultMessageMapper; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageCreator; +import org.springframework.integration.message.MessageMapper; +import org.springframework.integration.message.MessagingException; +import org.springframework.integration.scheduling.Subscription; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; + +/** + * Generates a proxy for the provided service interface to enable interaction + * with messaging components without application code being aware of them. + * + * @author Mark Fisher + */ +public class GatewayProxyFactoryBean implements FactoryBean, MethodInterceptor, InitializingBean, ApplicationContextAware, BeanClassLoaderAware { + + private Class serviceInterface; + + private MessageChannel requestChannel; + + private MessageChannel responseChannel; + + private ResponseCorrelator responseCorrelator; + + private EndpointRegistry endpointRegistry; + + private ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader(); + + private Object serviceProxy; + + private MessageCreator messageCreator = new DefaultMessageCreator(); + + private MessageMapper messageMapper = new DefaultMessageMapper(); + + + public void setServiceInterface(Class serviceInterface) { + this.serviceInterface = serviceInterface; + } + + public void setRequestChannel(MessageChannel requestChannel) { + this.requestChannel = requestChannel; + } + + public void setResponseChannel(MessageChannel responseChannel) { + this.responseChannel = responseChannel; + } + + public void setMessageCreator(MessageCreator messageCreator) { + Assert.notNull(messageCreator, "messageCreator must not be null"); + this.messageCreator = messageCreator; + } + + public void setMessageMapper(MessageMapper messageMapper) { + Assert.notNull(messageMapper, "messageMapper must not be null"); + this.messageMapper = messageMapper; + } + + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + if (applicationContext.containsBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME)) { + this.endpointRegistry = (EndpointRegistry) applicationContext.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); + } + } + + public void setBeanClassLoader(ClassLoader beanClassLoader) { + this.beanClassLoader = beanClassLoader; + } + + public void afterPropertiesSet() { + this.registerResponseCorrelatorIfNecessary(); + this.serviceProxy = new ProxyFactory(this.serviceInterface, this).getProxy(this.beanClassLoader); + } + + public Object getObject() throws Exception { + return this.serviceProxy; + } + + public Class getObjectType() { + return this.serviceInterface; + } + + public boolean isSingleton() { + return true; + } + + public Object invoke(MethodInvocation invocation) throws Throwable { + boolean returnsVoid = invocation.getMethod().getReturnType().equals(void.class); + int params = invocation.getMethod().getParameterTypes().length; + if (params == 0) { + // TODO: add support for receive-only + throw new MessagingException("Method invocation contains no arguments. Cannot send a message."); + } + if (this.requestChannel == null) { + throw new MessagingException("No request channel available. Cannot invoke methods with arguments."); + } + Object payload = (params == 1) ? invocation.getArguments()[0] : invocation.getArguments(); + Message message = this.messageCreator.createMessage(payload); + if (returnsVoid) { + this.requestChannel.send(message); + return null; + } + Message response = null; + if (this.responseCorrelator != null) { + message.getHeader().setReturnAddress(this.responseChannel); + this.requestChannel.send(message); + response = this.responseCorrelator.getResponse(message.getId()); + } + else { + RendezvousChannel temporaryChannel = new RendezvousChannel(); + message.getHeader().setReturnAddress(temporaryChannel); + this.requestChannel.send(message); + response = temporaryChannel.receive(); + } + return (response != null) ? this.messageMapper.mapMessage(response) : null; + } + + private void registerResponseCorrelatorIfNecessary() { + if (this.responseChannel != null) { + if (this.endpointRegistry == null) { + throw new ConfigurationException("No EndpointRegistry available. Cannot register ResponseCorrelator."); + } + ResponseCorrelator correlator = new ResponseCorrelator(10); + HandlerEndpoint endpoint = new HandlerEndpoint(correlator); + endpoint.setSubscription(new Subscription(this.responseChannel)); + this.endpointRegistry.registerEndpoint( + this.serviceInterface.getName() + "-" + this.responseChannel + "-correlator", endpoint); + this.responseCorrelator = correlator; + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/DefaultMessageCreator.java b/spring-integration-core/src/main/java/org/springframework/integration/message/DefaultMessageCreator.java new file mode 100644 index 0000000000..8bee454a0e --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/DefaultMessageCreator.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.message; + +/** + * A simple implementation of {@link MessageCreator} that uses the provide value + * as the {@link Message Message's} payload. + * + * @author Mark Fisher + */ +public class DefaultMessageCreator implements MessageCreator { + + public Message createMessage(Object object) { + return (object != null) ? new GenericMessage(object) : null; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/DefaultMessageMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/message/DefaultMessageMapper.java new file mode 100644 index 0000000000..2b4fe8afee --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/DefaultMessageMapper.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.message; + +/** + * A simple implementation of {@link MessageMapper} that returns the + * {@link Message Message's} payload. + * + * @author Mark Fisher + */ +public class DefaultMessageMapper implements MessageMapper { + + public Object mapMessage(Message message) { + return (message != null) ? message.getPayload() : null; + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java new file mode 100644 index 0000000000..b5db0318e5 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java @@ -0,0 +1,89 @@ +/* + * 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.gateway; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import org.junit.Test; + +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class GatewayProxyFactoryBeanTests { + + @Test + public void testRequestReplyWithAnonymousChannel() throws Exception { + final MessageChannel requestChannel = new QueueChannel(); + new Thread(new Runnable() { + public void run() { + Message input = requestChannel.receive(); + StringMessage response = new StringMessage(input.getPayload() + "bar"); + ((MessageChannel) input.getHeader().getReturnAddress()).send(response); + } + }).start(); + GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); + proxyFactory.setServiceInterface(TestService.class); + proxyFactory.setRequestChannel(requestChannel); + proxyFactory.afterPropertiesSet(); + TestService service = (TestService) proxyFactory.getObject(); + String result = service.requestReply("foo"); + assertEquals("foobar", result); + } + + @Test + public void testOneWay() throws Exception { + final MessageChannel requestChannel = new QueueChannel(); + GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); + proxyFactory.setServiceInterface(TestService.class); + proxyFactory.setRequestChannel(requestChannel); + proxyFactory.afterPropertiesSet(); + TestService service = (TestService) proxyFactory.getObject(); + service.oneWay("test"); + Message message = requestChannel.receive(1000); + assertNotNull(message); + assertEquals("test", message.getPayload()); + } + + @Test + public void testRequestReplyWithRendezvousChannelInApplicationContext() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "gatewayWithRendezvousChannel.xml", GatewayProxyFactoryBeanTests.class); + TestService service = (TestService) context.getBean("proxy"); + String result = service.requestReply("foo"); + assertEquals("foo!!!", result); + } + + @Test + public void testRequestReplyWithResponseCorrelatorInApplicationContext() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "gatewayWithResponseCorrelator.xml", GatewayProxyFactoryBeanTests.class); + TestService service = (TestService) context.getBean("proxy"); + String result = service.requestReply("foo"); + assertEquals("foo!!!", result); + TestChannelInterceptor interceptor = (TestChannelInterceptor) context.getBean("interceptor"); + assertEquals(1, interceptor.getSentCount()); + assertEquals(1, interceptor.getReceivedCount()); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestChannelInterceptor.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestChannelInterceptor.java new file mode 100644 index 0000000000..a8ac202edb --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestChannelInterceptor.java @@ -0,0 +1,57 @@ +/* + * 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.gateway; + +import java.util.concurrent.atomic.AtomicInteger; + +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter; +import org.springframework.integration.message.Message; + +/** + * @author Mark Fisher + */ +public class TestChannelInterceptor extends ChannelInterceptorAdapter { + + private final AtomicInteger sentCount = new AtomicInteger(); + + private final AtomicInteger receivedCount = new AtomicInteger(); + + + public int getSentCount() { + return this.sentCount.get(); + } + + public int getReceivedCount() { + return this.receivedCount.get(); + } + + @Override + public void postSend(Message message, MessageChannel channel, boolean sent) { + if (sent) { + this.sentCount.incrementAndGet(); + } + } + + @Override + public void postReceive(Message message, MessageChannel channel) { + if (message != null) { + this.receivedCount.incrementAndGet(); + } + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestHandler.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestHandler.java new file mode 100644 index 0000000000..2e9b0624b8 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestHandler.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.gateway; + +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class TestHandler implements MessageHandler { + + public Message handle(Message message) { + return new StringMessage(message.getPayload() + "!!!"); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestService.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestService.java new file mode 100644 index 0000000000..3475e628e4 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestService.java @@ -0,0 +1,28 @@ +/* + * 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.gateway; + +/** + * @author Mark Fisher + */ +public interface TestService { + + String requestReply(String input); + + void oneWay(String input); + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayWithRendezvousChannel.xml b/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayWithRendezvousChannel.xml new file mode 100644 index 0000000000..d2ea0b0be3 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayWithRendezvousChannel.xml @@ -0,0 +1,23 @@ + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayWithResponseCorrelator.xml b/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayWithResponseCorrelator.xml new file mode 100644 index 0000000000..557c246f59 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayWithResponseCorrelator.xml @@ -0,0 +1,30 @@ + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/message/DefaultMessageCreatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/message/DefaultMessageCreatorTests.java new file mode 100644 index 0000000000..9952217005 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/message/DefaultMessageCreatorTests.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.message; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + +import org.junit.Test; + +/** + * @author Mark Fisher + */ +public class DefaultMessageCreatorTests { + + @Test + public void testStringPayload() { + DefaultMessageCreator creator = new DefaultMessageCreator(); + Message message = creator.createMessage("testing"); + assertEquals("testing", message.getPayload()); + } + + @Test + public void testObjectPayload() { + DefaultMessageCreator creator = new DefaultMessageCreator(); + Object test = new Object(); + Message message = creator.createMessage(test); + assertEquals(test, message.getPayload()); + } + + @Test + public void testNull() { + DefaultMessageCreator creator = new DefaultMessageCreator(); + Message message = creator.createMessage(null); + assertNull(message); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/message/DefaultMessageMapperTests.java b/spring-integration-core/src/test/java/org/springframework/integration/message/DefaultMessageMapperTests.java new file mode 100644 index 0000000000..c7c436794a --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/message/DefaultMessageMapperTests.java @@ -0,0 +1,50 @@ +/* + * 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.message; + +import static org.junit.Assert.assertEquals; + +import org.junit.Test; + +/** + * @author Mark Fisher + */ +public class DefaultMessageMapperTests { + + @Test + public void testStringPayload() { + DefaultMessageMapper mapper = new DefaultMessageMapper(); + String result = (String) mapper.mapMessage(new StringMessage("testing")); + assertEquals("testing", result); + } + + @Test + public void testObjectPayload() { + DefaultMessageMapper mapper = new DefaultMessageMapper(); + Object test = new Object(); + Object result = mapper.mapMessage(new GenericMessage(test)); + assertEquals(test, result); + } + + @Test + public void testNullMessage() { + DefaultMessageMapper mapper = new DefaultMessageMapper(); + Object result = mapper.mapMessage(null); + assertEquals(null, result); + } + +}