Added initial implementation of GatewayProxyFactoryBean.

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;
+		}
+	}
+
+}

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;
+	}
+
+}

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;
+	}
+
+}

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());
+	}
+
+}

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();
+		}
+	}
+
+}

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() + "!!!");
+	}
+
+}

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);
+
+}

spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayWithRendezvousChannel.xml

@@ -0,0 +1,23 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
+
+	<message-bus/>
+
+	<channel id="requestChannel"/>
+
+	<endpoint handler="handler" input-channel="requestChannel"/>
+
+	<beans:bean id="proxy" class="org.springframework.integration.gateway.GatewayProxyFactoryBean">
+		<beans:property name="serviceInterface" value="org.springframework.integration.gateway.TestService"/>
+		<beans:property name="requestChannel" ref="requestChannel"/>
+	</beans:bean>
+
+	<beans:bean id="handler" class="org.springframework.integration.gateway.TestHandler"/>
+
+</beans:beans>

spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayWithResponseCorrelator.xml

@@ -0,0 +1,30 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
+
+	<message-bus/>
+
+	<channel id="requestChannel"/>
+
+	<channel id="responseChannel">
+		<interceptor ref="interceptor"/>
+	</channel>
+
+	<endpoint handler="handler" input-channel="requestChannel"/>
+
+	<beans:bean id="proxy" class="org.springframework.integration.gateway.GatewayProxyFactoryBean">
+		<beans:property name="serviceInterface" value="org.springframework.integration.gateway.TestService"/>
+		<beans:property name="requestChannel" ref="requestChannel"/>
+		<beans:property name="responseChannel" ref="responseChannel"/>
+	</beans:bean>
+
+	<beans:bean id="handler" class="org.springframework.integration.gateway.TestHandler"/>
+
+	<beans:bean id="interceptor" class="org.springframework.integration.gateway.TestChannelInterceptor"/>
+
+</beans:beans>

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);
+	}
+
+}

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<Object>(test));
+		assertEquals(test, result);
+	}
+
+	@Test
+	public void testNullMessage() {
+		DefaultMessageMapper mapper = new DefaultMessageMapper();
+		Object result = mapper.mapMessage(null);
+		assertEquals(null, result);
+	}
+
+}