Added initial implementation of MessageBus.

spring-eai-core/src/main/java/org/springframework/integration/MessageBus.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2002-2007 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;
+
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.Lifecycle;
+import org.springframework.integration.channel.ChannelRegistry;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.endpoint.EndpointRegistry;
+import org.springframework.integration.endpoint.MessageEndpoint;
+import org.springframework.util.Assert;
+
+/**
+ * A central component for registering channels and endpoints. The message bus
+ * will autodetect channels and endpoints from its host application context.
+ * 
+ * @author Mark Fisher
+ */
+public class MessageBus implements ApplicationContextAware {
+
+	private final Log logger = LogFactory.getLog(getClass());
+
+	private ChannelRegistry channelRegistry = new ChannelRegistry();
+
+	private EndpointRegistry endpointRegistry = new EndpointRegistry();
+
+	private ApplicationContext applicationContext;
+
+
+	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+		Assert.notNull(applicationContext, "applicationContext must not be null");
+		this.applicationContext = applicationContext;
+		this.initChannels();
+		this.initEndpoints();
+	}
+
+	@SuppressWarnings("unchecked")
+	private void initChannels() {
+		Map<String, MessageChannel> channelBeans = (Map<String, MessageChannel>) this.applicationContext
+				.getBeansOfType(MessageChannel.class);
+		for (Map.Entry<String, MessageChannel> entry : channelBeans.entrySet()) {
+			this.registerChannel(entry.getKey(), entry.getValue());
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	private void initEndpoints() {
+		Map<String, MessageEndpoint> endpointBeans = (Map<String, MessageEndpoint>) this.applicationContext
+				.getBeansOfType(MessageEndpoint.class);
+		for (Map.Entry<String, MessageEndpoint> entry : endpointBeans.entrySet()) {
+			this.registerEndpoint(entry.getKey(), entry.getValue());
+		}
+	}
+
+	public void registerChannel(String name, MessageChannel channel) {
+		this.channelRegistry.register(name, channel);
+		if (logger.isInfoEnabled()) {
+			logger.info("registering channel '" + name + "'");
+		}
+	}
+
+	public void registerEndpoint(String name, MessageEndpoint endpoint) {
+		if (endpoint instanceof Lifecycle) {
+			((Lifecycle) endpoint).start();
+		}
+		this.endpointRegistry.register(name, endpoint);
+		if (logger.isInfoEnabled()) {
+			logger.info("registering endpoint '" + name + "'");
+		}
+	}
+
+}

spring-eai-core/src/main/java/org/springframework/integration/endpoint/EndpointRegistry.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2002-2007 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.endpoint;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.springframework.util.Assert;
+
+/**
+ * A registry for maintaining message endpoints in a map keyed by name.
+ * 
+ * @author Mark Fisher
+ */
+public class EndpointRegistry {
+
+	private Map<String, MessageEndpoint> mappings = new ConcurrentHashMap<String, MessageEndpoint>();
+
+
+	public void register(String name, MessageEndpoint endpoint) {
+		Assert.notNull(name, "name must not be null");
+		Assert.notNull(endpoint, "endpoint must not be null");
+		this.mappings.put(name, endpoint);
+	}
+
+	public MessageEndpoint lookup(String name) {
+		return this.mappings.get(name);
+	}
+
+}

spring-eai-core/src/test/java/org/springframework/integration/MessageBusTests.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2002-2007 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;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import org.junit.Test;
+
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.channel.PointToPointChannel;
+import org.springframework.integration.endpoint.GenericMessageEndpoint;
+import org.springframework.integration.message.DocumentMessage;
+import org.springframework.integration.message.Message;
+
+/**
+ * @author Mark Fisher
+ */
+public class MessageBusTests {
+
+	@Test
+	public void testStandaloneWithEndpoint() {
+		MessageBus bus = new MessageBus();
+		MessageChannel sourceChannel = new PointToPointChannel();
+		MessageChannel targetChannel = new PointToPointChannel();
+		bus.registerChannel("sourceChannel", sourceChannel);
+		sourceChannel.send(new DocumentMessage("123", "test"));
+		bus.registerChannel("targetChannel", targetChannel);
+		GenericMessageEndpoint endpoint = new GenericMessageEndpoint(sourceChannel);
+		endpoint.setTarget(targetChannel);
+		bus.registerEndpoint("endpoint", endpoint);
+		Message result = targetChannel.receive();
+		assertEquals("test", result.getPayload());
+	}
+
+	@Test
+	public void testStandaloneWithoutEndpoint() {
+		MessageBus bus = new MessageBus();
+		MessageChannel sourceChannel = new PointToPointChannel();
+		sourceChannel.send(new DocumentMessage("123", "test"));
+		MessageChannel targetChannel = new PointToPointChannel();
+		bus.registerChannel("sourceChannel", sourceChannel);
+		bus.registerChannel("targetChannel", targetChannel);
+		Message result = targetChannel.receive(10);
+		assertNull(result);
+	}
+
+	@Test
+	public void testAutodetectionWithApplicationContext() {
+		ApplicationContext context = new ClassPathXmlApplicationContext("messageBusTests.xml", this.getClass());
+		MessageChannel sourceChannel = (MessageChannel) context.getBean("sourceChannel");
+		sourceChannel.send(new DocumentMessage("123", "test"));		
+		MessageChannel targetChannel = (MessageChannel) context.getBean("targetChannel");
+		Message result = targetChannel.receive(10);
+		assertEquals("test", result.getPayload());
+	}
+
+}

spring-eai-core/src/test/java/org/springframework/integration/messageBusTests.xml

@@ -0,0 +1,18 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans.xsd">
+
+	<bean class="org.springframework.integration.MessageBus"/>
+
+	<bean id="sourceChannel" class="org.springframework.integration.channel.PointToPointChannel"/>
+
+	<bean id="targetChannel" class="org.springframework.integration.channel.PointToPointChannel"/>
+
+	<bean id="endpoint" class="org.springframework.integration.endpoint.GenericMessageEndpoint">
+		<constructor-arg ref="sourceChannel"/>
+		<property name="target" ref="targetChannel"/>
+	</bean>
+
+</beans>