diff --git a/spring-eai-core/src/main/java/org/springframework/integration/MessageBus.java b/spring-eai-core/src/main/java/org/springframework/integration/MessageBus.java new file mode 100644 index 0000000000..4a54b7538e --- /dev/null +++ b/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 channelBeans = (Map) this.applicationContext + .getBeansOfType(MessageChannel.class); + for (Map.Entry entry : channelBeans.entrySet()) { + this.registerChannel(entry.getKey(), entry.getValue()); + } + } + + @SuppressWarnings("unchecked") + private void initEndpoints() { + Map endpointBeans = (Map) this.applicationContext + .getBeansOfType(MessageEndpoint.class); + for (Map.Entry 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 + "'"); + } + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/EndpointRegistry.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/EndpointRegistry.java new file mode 100644 index 0000000000..bd2d43b05f --- /dev/null +++ b/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 mappings = new ConcurrentHashMap(); + + + 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); + } + +} diff --git a/spring-eai-core/src/test/java/org/springframework/integration/MessageBusTests.java b/spring-eai-core/src/test/java/org/springframework/integration/MessageBusTests.java new file mode 100644 index 0000000000..bb617a7647 --- /dev/null +++ b/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()); + } + +} diff --git a/spring-eai-core/src/test/java/org/springframework/integration/messageBusTests.xml b/spring-eai-core/src/test/java/org/springframework/integration/messageBusTests.xml new file mode 100644 index 0000000000..86375edf7f --- /dev/null +++ b/spring-eai-core/src/test/java/org/springframework/integration/messageBusTests.xml @@ -0,0 +1,18 @@ + + + + + + + + + + + + + + +