From a62f2cf60c23a57023c1d2143bf8cbba93faf7ae Mon Sep 17 00:00:00 2001
From: Oleg Zhurakousky <ozhurakousky@vmware.com>
Date: Tue, 3 Nov 2009 10:59:56 +0000
Subject: [PATCH] INT-869. Committed initial support for OSGi Control Bus

---
 org.springframework.integration.osgi/ivy.xml  |   4 +-
 ...AbstractOSGiServiceManagingParserUtil.java |  55 ++++++
 .../osgi/config/xml/BusConfigParser.java      | 108 +++++++++++
 ...grationOSGiControlBusNamespaceHandler.java |  33 ++++
 .../osgi/OSGiIntegrationControlBus.java       |  62 ++++++
 .../main/resources/META-INF/spring.handlers   |   2 +-
 .../main/resources/META-INF/spring.schemas    |   2 +-
 .../spring/integration-control-bus.xml        |  38 ++++
 .../config/xml/spring-integration-bus-1.0.xsd |  25 ++-
 .../AbstractSIConfigBundleTestDeployer.java   |  53 ++++++
 .../xml/BusConfigParserTests-default.xml      |  14 ++
 ...BusConfigParserTests-overrideGroupName.xml |  14 ++
 .../osgi/config/xml/BusConfigParserTests.java |  68 +++++++
 ...BusUsageFailoverReferenceTests-context.xml |  18 ++
 .../BusUsageFailoverTests-context-primary.xml |  13 ++
 ...usUsageFailoverTests-context-secondary.xml |  13 ++
 .../config/xml/BusUsageFailoverTests.java     |  78 ++++++++
 ...nSingleApplicationContextTests-context.xml |  14 ++
 ...nitionInSingleApplicationContextTests.java |  38 ++++
 .../osgi/config/xml/base-bus-config.xml       |  36 ++++
 .../integration/osgi/stubs/OSGiMockUtils.java | 117 ++++++++++++
 .../osgi/stubs/SIBundleContextStub.java       | 180 ++++++++++++++++++
 .../osgi/stubs/SIBundleContextStubTest.java   |  83 ++++++++
 .../osgi/stubs/SIServiceRegistrationStub.java |  52 +++++
 .../template.mf                               |   7 +-
 25 files changed, 1111 insertions(+), 16 deletions(-)
 create mode 100644 org.springframework.integration.osgi/src/main/java/org/springframework/integration/osgi/config/xml/AbstractOSGiServiceManagingParserUtil.java
 create mode 100644 org.springframework.integration.osgi/src/main/java/org/springframework/integration/osgi/config/xml/BusConfigParser.java
 create mode 100644 org.springframework.integration.osgi/src/main/java/org/springframework/integration/osgi/config/xml/IntegrationOSGiControlBusNamespaceHandler.java
 create mode 100644 org.springframework.integration.osgi/src/main/java/org/springframework/intergration/osgi/OSGiIntegrationControlBus.java
 create mode 100644 org.springframework.integration.osgi/src/main/resources/META-INF/spring/integration-control-bus.xml
 create mode 100644 org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/AbstractSIConfigBundleTestDeployer.java
 create mode 100644 org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusConfigParserTests-default.xml
 create mode 100644 org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusConfigParserTests-overrideGroupName.xml
 create mode 100644 org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusConfigParserTests.java
 create mode 100644 org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusUsageFailoverReferenceTests-context.xml
 create mode 100644 org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusUsageFailoverTests-context-primary.xml
 create mode 100644 org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusUsageFailoverTests-context-secondary.xml
 create mode 100644 org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusUsageFailoverTests.java
 create mode 100644 org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/MultiBusDefinitionInSingleApplicationContextTests-context.xml
 create mode 100644 org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/MultiBusDefinitionInSingleApplicationContextTests.java
 create mode 100644 org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/base-bus-config.xml
 create mode 100644 org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/stubs/OSGiMockUtils.java
 create mode 100644 org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/stubs/SIBundleContextStub.java
 create mode 100644 org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/stubs/SIBundleContextStubTest.java
 create mode 100644 org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/stubs/SIServiceRegistrationStub.java

diff --git a/org.springframework.integration.osgi/ivy.xml b/org.springframework.integration.osgi/ivy.xml
index 63a151148f..ec7f19344b 100644
--- a/org.springframework.integration.osgi/ivy.xml
+++ b/org.springframework.integration.osgi/ivy.xml
@@ -5,7 +5,7 @@
 		xsi:noNamespaceSchemaLocation="http://incubator.apache.org/ivy/schemas/ivy.xsd"
 		version="1.3">
 
-	<info organisation="org.springframework.integration.osgi" module="${ant.project.name}">
+	<info organisation="org.springframework.integration" module="${ant.project.name}">
 		<license name="Apache 2.0" url="http://www.apache.org/licenses/LICENSE-2.0"/>
 		<ivyauthor name="Oleg Zhurakousky"/>
 	</info>
@@ -22,7 +22,6 @@
 	<dependencies>
 		<dependency org="net.sourceforge.cglib" name="com.springsource.net.sf.cglib" rev="2.1.3" conf="test->runtime"/>
 		<dependency org="org.junit" name="com.springsource.org.junit" rev="${junit.version}" conf="test->runtime"/>
-		<dependency org="org.easymock" name="com.springsource.org.easymock" rev="2.3.0" conf="test->runtime"/>
 		<dependency org="org.springframework" name="org.springframework.aop" rev="${spring.version}" conf="compile->runtime"/>
 		<dependency org="org.springframework" name="org.springframework.context" rev="${spring.version}" conf="compile->runtime"/>
 		<dependency org="org.springframework" name="org.springframework.transaction" rev="${spring.version}" conf="compile->runtime"/>
@@ -36,7 +35,6 @@
 		<dependency org="org.springframework.osgi" name="org.springframework.osgi.mock" rev="1.2.0" conf="compile->compile"/>
 		<dependency org="org.springframework.osgi" name="org.springframework.osgi.io" rev="1.2.0" conf="compile->compile"/>
 		<dependency org="org.springframework.osgi" name="org.springframework.osgi.core" rev="1.2.0" conf="compile->compile"/>
-		<dependency org="org.apache.log4j" name="com.springsource.org.apache.log4j" rev="1.2.15" conf="compile->runtime"/>
 		<dependency org="org.springframework.integration" name="org.springframework.integration.event" rev="latest.integration" conf="compile->compile"/>
 	</dependencies>
 
diff --git a/org.springframework.integration.osgi/src/main/java/org/springframework/integration/osgi/config/xml/AbstractOSGiServiceManagingParserUtil.java b/org.springframework.integration.osgi/src/main/java/org/springframework/integration/osgi/config/xml/AbstractOSGiServiceManagingParserUtil.java
new file mode 100644
index 0000000000..54d66684bc
--- /dev/null
+++ b/org.springframework.integration.osgi/src/main/java/org/springframework/integration/osgi/config/xml/AbstractOSGiServiceManagingParserUtil.java
@@ -0,0 +1,55 @@
+/*
+ * 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.osgi.config.xml;
+
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
+import org.springframework.beans.factory.support.BeanDefinitionRegistry;
+import org.springframework.integration.controlbus.ControlBus;
+import org.springframework.osgi.service.exporter.support.AutoExport;
+import org.springframework.osgi.service.exporter.support.OsgiServiceFactoryBean;
+
+
+/**
+ * Will register {@link OsgiServiceFactoryBean} to export {@link ControlBus} as an OSGi service
+ * 
+ * @author Oleg Zhurakousky
+ * @since 2.0
+ */
+public class AbstractOSGiServiceManagingParserUtil {
+
+	/**
+	 * Will export a bean identified by the 'beanName' as an OSGi Service. It will publish the service under all
+	 * interfaces visible this class represents.
+	 * 
+	 * @param beanName
+	 * @param registry
+	 */
+	@SuppressWarnings("unchecked")
+	public static void registerServiceExporterFor(String beanName, BeanDefinitionRegistry registry, Class... publishedIntefaces){
+		BeanDefinitionBuilder serviceBuilder = BeanDefinitionBuilder.genericBeanDefinition(OsgiServiceFactoryBean.class);
+		serviceBuilder.addPropertyValue("targetBeanName", beanName);
+		serviceBuilder.addPropertyValue("interfaces", new Class[]{ControlBus.class});
+		if (publishedIntefaces != null && publishedIntefaces.length > 0){
+			serviceBuilder.addPropertyValue("interfaces", publishedIntefaces);
+		} else {
+			serviceBuilder.addPropertyValue("autoExport", AutoExport.INTERFACES);
+		}
+		serviceBuilder.addPropertyValue("registerService", true);
+		BeanDefinitionReaderUtils.registerWithGeneratedName(serviceBuilder.getBeanDefinition(), registry);
+	}
+
+}
diff --git a/org.springframework.integration.osgi/src/main/java/org/springframework/integration/osgi/config/xml/BusConfigParser.java b/org.springframework.integration.osgi/src/main/java/org/springframework/integration/osgi/config/xml/BusConfigParser.java
new file mode 100644
index 0000000000..bc4cc4297b
--- /dev/null
+++ b/org.springframework.integration.osgi/src/main/java/org/springframework/integration/osgi/config/xml/BusConfigParser.java
@@ -0,0 +1,108 @@
+/*
+ * 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.osgi.config.xml;
+
+import java.util.concurrent.ThreadPoolExecutor;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.beans.factory.BeanDefinitionStoreException;
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
+import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.channel.PublishSubscribeChannel;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.intergration.osgi.OSGiIntegrationControlBus;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+import org.w3c.dom.Element;
+
+/**
+ * Parser to handle 'bus-config' element.
+ * 
+ * @author Oleg Zhurakousky
+ * @since 2.0
+ */
+public class BusConfigParser extends AbstractBeanDefinitionParser {
+	private static final Log log = LogFactory.getLog(BusConfigParser.class);
+	
+	private String beanName;
+	/**
+	 * 
+	 */
+	protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
+			throws BeanDefinitionStoreException {
+		return beanName;
+	}
+	/**
+	 * 
+	 */
+	protected AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) {
+		String busGroupName = element.getAttribute("group-name");
+		Assert.isTrue(StringUtils.hasText(busGroupName), "bus-config 'group-name' attribute must be provided");
+		beanName = "controlBus-" + busGroupName;
+		if (parserContext.getRegistry().containsBeanDefinition(beanName)){
+			throw new BeanDefinitionStoreException("You atempted to register a second instance of the Control Bus with the same 'group-name' " +
+					"in the single Application Context which is not allowed.");
+		}
+		
+		BeanDefinitionBuilder rootBuilder = BeanDefinitionBuilder.rootBeanDefinition(OSGiIntegrationControlBus.class);
+		
+		//String busChannelName = "controlMessagesDistributionChannel";
+			//this.registerPubSubChannelDefinition(element.getAttribute("task-executor"), element, parserContext);
+		rootBuilder.addConstructorArgReference("controlMessagesDistributionChannel");
+		
+		AbstractOSGiServiceManagingParserUtil.registerServiceExporterFor(beanName, parserContext.getRegistry());
+		return rootBuilder.getBeanDefinition();
+	}
+	/**
+	 * 
+	 * @param taskExecutorName
+	 * @param element
+	 * @param parserContext
+	 * @return
+	 */
+	private String registerPubSubChannelDefinition(String taskExecutorName, Element element, ParserContext parserContext){
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(PublishSubscribeChannel.class.getName());
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-handler");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "ignore-failures");
+		if (!StringUtils.hasText(taskExecutorName)) {
+			taskExecutorName = this.createTaskExecutorDefinition(element, parserContext);
+		}
+		builder.addConstructorArgReference(taskExecutorName);
+		
+		String beanName = BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry());
+		return beanName;
+	}
+	/**
+	 * 
+	 * @param element
+	 * @param parserContext
+	 * @return
+	 */
+	private String createTaskExecutorDefinition(Element element, ParserContext parserContext){
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ThreadPoolTaskExecutor.class.getName());
+		builder.addPropertyValue("corePoolSize", 5);
+		builder.addPropertyValue("maxPoolSize", 10);
+		builder.addPropertyValue("rejectedExecutionHandler", new ThreadPoolExecutor.DiscardPolicy());
+		String beanName = BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry());
+		return beanName;
+	}
+	
+}
diff --git a/org.springframework.integration.osgi/src/main/java/org/springframework/integration/osgi/config/xml/IntegrationOSGiControlBusNamespaceHandler.java b/org.springframework.integration.osgi/src/main/java/org/springframework/integration/osgi/config/xml/IntegrationOSGiControlBusNamespaceHandler.java
new file mode 100644
index 0000000000..ef91ee98d0
--- /dev/null
+++ b/org.springframework.integration.osgi/src/main/java/org/springframework/integration/osgi/config/xml/IntegrationOSGiControlBusNamespaceHandler.java
@@ -0,0 +1,33 @@
+/*
+ * 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.osgi.config.xml;
+
+import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler;
+
+/**
+ * @author Oleg Zhurakousky
+ * @since 2.0
+ */
+public class IntegrationOSGiControlBusNamespaceHandler extends AbstractIntegrationNamespaceHandler {
+
+	/**
+	 * Will register the required parsers
+	 */
+	public void init() {
+		registerBeanDefinitionParser("bus-config", new BusConfigParser());
+	}
+
+}
diff --git a/org.springframework.integration.osgi/src/main/java/org/springframework/intergration/osgi/OSGiIntegrationControlBus.java b/org.springframework.integration.osgi/src/main/java/org/springframework/intergration/osgi/OSGiIntegrationControlBus.java
new file mode 100644
index 0000000000..8cc9e9e74b
--- /dev/null
+++ b/org.springframework.integration.osgi/src/main/java/org/springframework/intergration/osgi/OSGiIntegrationControlBus.java
@@ -0,0 +1,62 @@
+/*
+ * 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.intergration.osgi;
+
+import org.springframework.integration.channel.SubscribableChannel;
+import org.springframework.integration.controlbus.ControlBus;
+import org.springframework.integration.core.Message;
+import org.springframework.integration.message.MessageHandler;
+
+/**
+ * Implementation of the {@link ControlBus} interface. 
+ * Control Bus itself wrapper over {@link SubscribableChannel}, 
+ * which represents the entry point to Control BUs infrastructure.
+ * 
+ * @author Oleg Zhurakousky
+ * @since 2.0
+ */
+public class OSGiIntegrationControlBus implements ControlBus {
+	private SubscribableChannel channel;
+	
+	public OSGiIntegrationControlBus(SubscribableChannel channel){
+		this.channel = channel;
+	}
+
+	public boolean subscribe(MessageHandler handler) {
+		return channel.subscribe(handler);
+	}
+
+	
+	public boolean unsubscribe(MessageHandler handler) {
+		return channel.unsubscribe(handler);
+	}
+
+	public String getName() {
+		return channel.getName();
+	}
+
+	
+	public boolean send(Message<?> message) {
+		return channel.send(message);
+	}
+
+	
+	public boolean send(Message<?> message, long timeout) {
+		return channel.send(message, timeout);
+	}
+
+
+}
diff --git a/org.springframework.integration.osgi/src/main/resources/META-INF/spring.handlers b/org.springframework.integration.osgi/src/main/resources/META-INF/spring.handlers
index 9b37098145..ced91b7818 100644
--- a/org.springframework.integration.osgi/src/main/resources/META-INF/spring.handlers
+++ b/org.springframework.integration.osgi/src/main/resources/META-INF/spring.handlers
@@ -1 +1 @@
-http\://www.springframework.org/schema/integration-bus=org.springframework.integration.osgi.config.xml.IntegrationOSGiNamespaceHandler
\ No newline at end of file
+http\://www.springframework.org/schema/integration/integration-control-bus=org.springframework.integration.osgi.config.xml.IntegrationOSGiControlBusNamespaceHandler
\ No newline at end of file
diff --git a/org.springframework.integration.osgi/src/main/resources/META-INF/spring.schemas b/org.springframework.integration.osgi/src/main/resources/META-INF/spring.schemas
index 1083efd130..785dbf4d6f 100644
--- a/org.springframework.integration.osgi/src/main/resources/META-INF/spring.schemas
+++ b/org.springframework.integration.osgi/src/main/resources/META-INF/spring.schemas
@@ -1 +1 @@
-http\://www.springframework.org/schema/integration-bus/spring-integration-bus-1.0.xsd=org/springframework/integration/osgi/config/xml/spring-integration-bus-1.0.xsd
+http\://www.springframework.org/schema/integration/integration-control-bus/spring-integration-bus-1.0.xsd=org/springframework/integration/osgi/config/xml/spring-integration-bus-1.0.xsd
diff --git a/org.springframework.integration.osgi/src/main/resources/META-INF/spring/integration-control-bus.xml b/org.springframework.integration.osgi/src/main/resources/META-INF/spring/integration-control-bus.xml
new file mode 100644
index 0000000000..2be951a774
--- /dev/null
+++ b/org.springframework.integration.osgi/src/main/resources/META-INF/spring/integration-control-bus.xml
@@ -0,0 +1,38 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:si-control="http://www.springframework.org/schema/integration/integration-control-bus"
+	xmlns:si="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
+	                   http://www.springframework.org/schema/integration/integration-control-bus http://www.springframework.org/schema/integration/integration-control-bus/spring-integration-bus-1.0.xsd
+                       http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd">
+	
+	<si-control:bus-config group-name="DEFAULT_CONTROL_GROUP"/>
+	
+	<!-- SERVICE REGISTRATION CONFIG-->
+	<si:publish-subscribe-channel id="serviceRegistrationControlChannel" task-executor="executor"/>
+	<si:service-activator input-channel="serviceRegistrationControlChannel">
+		<bean class="org.springframework.integration.controlbus.ControlBusMessageHandler"/>
+	</si:service-activator>
+	 <!-- END SERVICE REGISTRATION CONFIG-->
+
+	<!-- SERVICE BINDING CONFIG-->
+	<si:publish-subscribe-channel id="serviceBindingControlChannel" task-executor="executor"/>
+	<si:service-activator input-channel="serviceBindingControlChannel">
+		<bean class="org.springframework.integration.controlbus.ControlBusMessageHandler"/>
+	</si:service-activator> 
+	<!-- END SERVICE BINDING CONFIG-->
+	
+	<!-- CONTROL BUS channel, exposed as an OSGi service --> 
+	<si:publish-subscribe-channel id="controlMessagesDistributionChannel"/>
+	<!-- END CONTROL BUS channel, exposed as an OSGi service --> 
+	
+	<si:header-value-router input-channel="controlMessagesDistributionChannel" header-name="INTEGRATION_EVENT_TYPE">
+		<si:mapping value="REGISTRATION" channel="serviceRegistrationControlChannel"/>
+		<si:mapping value="UNREGISTRATION" channel="serviceRegistrationControlChannel"/>
+		<si:mapping value="BINDING" channel="serviceBindingControlChannel"/>
+		<si:mapping value="UNBINDING" channel="serviceBindingControlChannel"/>
+	</si:header-value-router>
+	
+	<si:thread-pool-task-executor id="executor" core-size="10"/>
+</beans>
diff --git a/org.springframework.integration.osgi/src/main/resources/org/springframework/integration/osgi/config/xml/spring-integration-bus-1.0.xsd b/org.springframework.integration.osgi/src/main/resources/org/springframework/integration/osgi/config/xml/spring-integration-bus-1.0.xsd
index ca61592e32..ed207e7133 100644
--- a/org.springframework.integration.osgi/src/main/resources/org/springframework/integration/osgi/config/xml/spring-integration-bus-1.0.xsd
+++ b/org.springframework.integration.osgi/src/main/resources/org/springframework/integration/osgi/config/xml/spring-integration-bus-1.0.xsd
@@ -1,26 +1,33 @@
-<xsd:schema xmlns="http://www.springframework.org/schema/integration-bus"
-	xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:beans="http://www.springframework.org/schema/beans"
+<xsd:schema xmlns="http://www.springframework.org/schema/integration/integration-control-bus"
+	xmlns:xsd="http://www.w3.org/2001/XMLSchema" 
+	xmlns:beans="http://www.springframework.org/schema/beans"
 	xmlns:tool="http://www.springframework.org/schema/tool"
-	targetNamespace="http://www.springframework.org/schema/integration-bus"
+	targetNamespace="http://www.springframework.org/schema/integration/integration-control-bus"
 	elementFormDefault="qualified" attributeFormDefault="unqualified">
-
 	<xsd:import namespace="http://www.springframework.org/schema/beans" />
 	<xsd:import namespace="http://www.springframework.org/schema/tool" />
-
-	<xsd:element name="osgi">
+	
+	<xsd:element name="bus-config">
 		<xsd:complexType>
 			<xsd:annotation>
 				<xsd:documentation>
-					Auto-exports SI artifacts as OSGi services
+					Configures all required components for a named Control Bus configuration 
 				</xsd:documentation>
 			</xsd:annotation>
 			<xsd:attribute name="group-name" type="xsd:string" default="DEFAULT_CONTROL_GROUP">
 				<xsd:annotation>
 					<xsd:documentation>
-						Identified group-name of a Message Broker this configuration will be connected to.
+						Identifies group-name of a Control Bus this configuration defines
+				</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="task-executor" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						Identifies the 'task-executor' for this bus's instance 
 				</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
 		</xsd:complexType>
-	</xsd:element>
+	</xsd:element>	
 </xsd:schema>
\ No newline at end of file
diff --git a/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/AbstractSIConfigBundleTestDeployer.java b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/AbstractSIConfigBundleTestDeployer.java
new file mode 100644
index 0000000000..54b073a1e5
--- /dev/null
+++ b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/AbstractSIConfigBundleTestDeployer.java
@@ -0,0 +1,53 @@
+/*
+ * 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.osgi;
+
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.Enumeration;
+
+import org.osgi.framework.BundleContext;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.integration.osgi.stubs.SIBundleContextStub;
+import org.springframework.osgi.context.ConfigurableOsgiBundleApplicationContext;
+import org.springframework.osgi.context.support.OsgiBundleXmlApplicationContext;
+import org.springframework.osgi.mock.ArrayEnumerator;
+import org.springframework.osgi.mock.MockBundle;
+
+/**
+ * @author Oleg Zhurakousky
+ * @since 2.0
+ */
+public abstract class AbstractSIConfigBundleTestDeployer {
+	protected OsgiBundleXmlApplicationContext applicationContext;
+	
+	@SuppressWarnings("unchecked")
+	public ConfigurableOsgiBundleApplicationContext deploySIConfig(BundleContext bundleContext, String configPackage, String... configFiles) throws Exception {
+		applicationContext = new OsgiBundleXmlApplicationContext(configFiles);
+		final ArrayList<URL> tempConfigurations = new ArrayList<URL>();
+		for (String configFile : configFiles) {
+			tempConfigurations.add(new ClassPathResource(configPackage + configFile).getURL());
+		}
+		((SIBundleContextStub)bundleContext).setBundle(new MockBundle() {
+			public Enumeration findEntries(String path, String filePattern, boolean recurse) {
+				return new ArrayEnumerator(tempConfigurations.toArray());
+			}
+		});
+		((OsgiBundleXmlApplicationContext)applicationContext).setBundleContext(bundleContext);
+		((OsgiBundleXmlApplicationContext)applicationContext).refresh();
+		return applicationContext;
+	}
+}
diff --git a/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusConfigParserTests-default.xml b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusConfigParserTests-default.xml
new file mode 100644
index 0000000000..dc50cfdfc3
--- /dev/null
+++ b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusConfigParserTests-default.xml
@@ -0,0 +1,14 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:si-control="http://www.springframework.org/schema/integration/integration-control-bus"
+	xmlns:si="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
+	                   http://www.springframework.org/schema/integration/integration-control-bus http://www.springframework.org/schema/integration/integration-control-bus/spring-integration-bus-1.0.xsd
+                       http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd">
+	
+	<import resource="base-bus-config.xml"/>
+	
+	<si-control:bus-config/>
+
+</beans>
diff --git a/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusConfigParserTests-overrideGroupName.xml b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusConfigParserTests-overrideGroupName.xml
new file mode 100644
index 0000000000..394b7fd93f
--- /dev/null
+++ b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusConfigParserTests-overrideGroupName.xml
@@ -0,0 +1,14 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:si-control="http://www.springframework.org/schema/integration/integration-control-bus"
+	xmlns:si="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
+	                   http://www.springframework.org/schema/integration/integration-control-bus http://www.springframework.org/schema/integration/integration-control-bus/spring-integration-bus-1.0.xsd
+                       http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd">
+	
+	<import resource="base-bus-config.xml"/>
+	
+	<si-control:bus-config group-name="FOO"/>
+
+</beans>
diff --git a/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusConfigParserTests.java b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusConfigParserTests.java
new file mode 100644
index 0000000000..9f0fbbd877
--- /dev/null
+++ b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusConfigParserTests.java
@@ -0,0 +1,68 @@
+/*
+ * 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.osgi.config.xml;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Test;
+import org.osgi.framework.BundleContext;
+import org.osgi.framework.ServiceReference;
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.context.ApplicationContext;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.integration.channel.SubscribableChannel;
+import org.springframework.integration.controlbus.ControlBus;
+import org.springframework.integration.osgi.AbstractSIConfigBundleTestDeployer;
+import org.springframework.integration.osgi.stubs.SIBundleContextStub;
+import org.springframework.integration.util.ErrorHandlingTaskExecutor;
+
+
+/**
+ * @author Oleg Zhurakousky
+ * @since 2.0
+ */
+public class BusConfigParserTests extends AbstractSIConfigBundleTestDeployer {
+
+	@Test
+	public void testDefaultControlBusConfig() throws Exception {
+		BundleContext bundleContext = SIBundleContextStub.getInstance();
+		ApplicationContext ac = this.deploySIConfig(bundleContext, 
+				                                    "org/springframework/integration/osgi/config/xml/", 
+				                                    "BusConfigParserTests-default.xml");
+		ControlBus controlBus = (ControlBus) ac.getBean("controlBus-DEFAULT_CONTROL_GROUP");
+		assertNotNull(controlBus);
+		ServiceReference[] sr = bundleContext.getServiceReferences(ControlBus.class.getName(), 
+				                                                  "(&(org.springframework.osgi.bean.name=controlBus-DEFAULT_CONTROL_GROUP))");
+		assertNotNull(sr);
+		assertTrue(sr.length == 1);
+		controlBus = (ControlBus) bundleContext.getService(sr[0]);
+	}
+	
+	@Test
+	public void testNamededControlBusConfig() throws Exception {
+		BundleContext bundleContext = SIBundleContextStub.getInstance();
+		ApplicationContext ac = this.deploySIConfig(bundleContext, 
+				                                    "org/springframework/integration/osgi/config/xml/", 
+				                                    "BusConfigParserTests-overrideGroupName.xml");
+		ControlBus controlBus = (ControlBus) ac.getBean("controlBus-FOO");
+		assertNotNull(controlBus);
+		ServiceReference sr = bundleContext.getServiceReference(ControlBus.class.getName());
+		assertNotNull(sr);
+		controlBus = (ControlBus) bundleContext.getService(sr);
+	}
+}
diff --git a/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusUsageFailoverReferenceTests-context.xml b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusUsageFailoverReferenceTests-context.xml
new file mode 100644
index 0000000000..19482fdcca
--- /dev/null
+++ b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusUsageFailoverReferenceTests-context.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"
+	xmlns:si-control="http://www.springframework.org/schema/integration/integration-control-bus"
+	xmlns:si="http://www.springframework.org/schema/integration"
+	xmlns:osgi="http://www.springframework.org/schema/osgi"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
+	                   http://www.springframework.org/schema/integration/integration-control-bus http://www.springframework.org/schema/integration/integration-control-bus/spring-integration-bus-1.0.xsd
+                       http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
+                       http://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.xsd">
+
+
+	<osgi:reference id="bus" interface="org.springframework.integration.controlbus.ControlBus">
+		<osgi:listener ref="busListener" bind-method="bind" unbind-method="unBind"/>
+	</osgi:reference>
+
+	<bean id="busListener" class="org.springframework.integration.osgi.config.xml.BusUsageFailoverTests$BusListener"/>
+</beans>
diff --git a/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusUsageFailoverTests-context-primary.xml b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusUsageFailoverTests-context-primary.xml
new file mode 100644
index 0000000000..5a7a111150
--- /dev/null
+++ b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusUsageFailoverTests-context-primary.xml
@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:si-control="http://www.springframework.org/schema/integration/integration-control-bus"
+	xmlns:si="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
+	                   http://www.springframework.org/schema/integration/integration-control-bus http://www.springframework.org/schema/integration/integration-control-bus/spring-integration-bus-1.0.xsd
+                       http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd">
+	
+	<import resource="base-bus-config.xml"/>
+	<si-control:bus-config/>
+
+</beans>
diff --git a/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusUsageFailoverTests-context-secondary.xml b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusUsageFailoverTests-context-secondary.xml
new file mode 100644
index 0000000000..5a7a111150
--- /dev/null
+++ b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusUsageFailoverTests-context-secondary.xml
@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:si-control="http://www.springframework.org/schema/integration/integration-control-bus"
+	xmlns:si="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
+	                   http://www.springframework.org/schema/integration/integration-control-bus http://www.springframework.org/schema/integration/integration-control-bus/spring-integration-bus-1.0.xsd
+                       http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd">
+	
+	<import resource="base-bus-config.xml"/>
+	<si-control:bus-config/>
+
+</beans>
diff --git a/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusUsageFailoverTests.java b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusUsageFailoverTests.java
new file mode 100644
index 0000000000..774bffe6a3
--- /dev/null
+++ b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/BusUsageFailoverTests.java
@@ -0,0 +1,78 @@
+/*
+ * 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.osgi.config.xml;
+
+import java.util.Map;
+
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.osgi.framework.BundleContext;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.integration.controlbus.ControlBus;
+import org.springframework.integration.message.MessageHandler;
+import org.springframework.integration.message.StringMessage;
+import org.springframework.integration.osgi.AbstractSIConfigBundleTestDeployer;
+import org.springframework.integration.osgi.stubs.SIBundleContextStub;
+
+/**
+ * THis test bootstraps two instances of the Control Bus, then shuts down one demonstrating and testing 
+ * transparent fail-over to another instance of the Control Bus
+ * 
+ * @author Oleg Zhurakousky
+ * @since 2.0
+ */
+public class BusUsageFailoverTests extends AbstractSIConfigBundleTestDeployer {	
+	static MessageHandler handler = Mockito.mock(MessageHandler.class);
+	
+	@Test
+	public void testDefaultControlBusConfig() throws Exception {
+		BundleContext bundleContext = SIBundleContextStub.getInstance();
+
+		ConfigurableApplicationContext primaryAc = this.deploySIConfig(bundleContext, 
+				            "org/springframework/integration/osgi/config/xml/", 
+				            "BusUsageFailoverTests-context-primary.xml");
+		
+		this.deploySIConfig(bundleContext, 
+	                        "org/springframework/integration/osgi/config/xml/", 
+	                        "BusUsageFailoverTests-context-secondary.xml");
+		
+		ApplicationContext userAC = this.deploySIConfig(bundleContext, 
+                									"org/springframework/integration/osgi/config/xml/", 
+                									"BusUsageFailoverReferenceTests-context.xml");
+		
+		ControlBus bus = (ControlBus) userAC.getBean("bus");
+		StringMessage message = new StringMessage("hello");
+		
+		bus.send(message);
+		Thread.sleep(200);
+		Mockito.verify(handler, Mockito.times(1)).handleMessage(message);
+		primaryAc.close(); // shut down primary Bus
+			  
+		Mockito.reset(handler);
+		bus.send(message);
+		Thread.sleep(200);
+		Mockito.verify(handler, Mockito.times(1)).handleMessage(message);
+	}
+	
+	public static class BusListener{
+		public void bind(ControlBus bus, Map properties){
+			bus.subscribe(handler);
+		}
+		public void unBind(ControlBus bus, Map properties){
+		}
+	}
+}
diff --git a/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/MultiBusDefinitionInSingleApplicationContextTests-context.xml b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/MultiBusDefinitionInSingleApplicationContextTests-context.xml
new file mode 100644
index 0000000000..677be9d45d
--- /dev/null
+++ b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/MultiBusDefinitionInSingleApplicationContextTests-context.xml
@@ -0,0 +1,14 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:si-control="http://www.springframework.org/schema/integration/integration-control-bus"
+	xmlns:si="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
+	                   http://www.springframework.org/schema/integration/integration-control-bus http://www.springframework.org/schema/integration/integration-control-bus/spring-integration-bus-1.0.xsd
+                       http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd">
+	
+	<import resource="base-bus-config.xml"/>
+	<si-control:bus-config/>
+	<si-control:bus-config/>
+
+</beans>
diff --git a/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/MultiBusDefinitionInSingleApplicationContextTests.java b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/MultiBusDefinitionInSingleApplicationContextTests.java
new file mode 100644
index 0000000000..1132a0721a
--- /dev/null
+++ b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/MultiBusDefinitionInSingleApplicationContextTests.java
@@ -0,0 +1,38 @@
+/*
+ * 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.osgi.config.xml;
+
+import org.junit.Test;
+import org.osgi.framework.BundleContext;
+import org.springframework.beans.factory.BeanDefinitionStoreException;
+import org.springframework.integration.osgi.AbstractSIConfigBundleTestDeployer;
+import org.springframework.integration.osgi.stubs.SIBundleContextStub;
+
+/**
+ * Will make sure only one instance of the Control Bus can exist per single Application Context
+ * 
+ * @author Oleg Zhurakousky
+ * @since 2.0
+ */
+public class MultiBusDefinitionInSingleApplicationContextTests extends AbstractSIConfigBundleTestDeployer{
+	@Test(expected=BeanDefinitionStoreException.class)
+	public void testMultiBusDefinitionConfig() throws Exception {
+		BundleContext bundleContext = SIBundleContextStub.getInstance();
+		this.deploySIConfig(bundleContext, 
+				            "org/springframework/integration/osgi/config/xml/", 
+				            "MultiBusDefinitionInSingleApplicationContextTests-context.xml");
+	}
+}
diff --git a/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/base-bus-config.xml b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/base-bus-config.xml
new file mode 100644
index 0000000000..4cc4ee4722
--- /dev/null
+++ b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/config/xml/base-bus-config.xml
@@ -0,0 +1,36 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:si-control="http://www.springframework.org/schema/integration/integration-control-bus"
+	xmlns:si="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
+	                   http://www.springframework.org/schema/integration/integration-control-bus http://www.springframework.org/schema/integration/integration-control-bus/spring-integration-bus-1.0.xsd
+                       http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd">
+	
+	<!-- SERVICE REGISTRATION CONFIG-->
+	<si:publish-subscribe-channel id="serviceRegistrationControlChannel" task-executor="executor"/>
+	<si:service-activator input-channel="serviceRegistrationControlChannel">
+		<bean class="org.springframework.integration.controlbus.ControlBusMessageHandler"/>
+	</si:service-activator>
+	 <!-- END SERVICE REGISTRATION CONFIG-->
+
+	<!-- SERVICE BINDING CONFIG-->
+	<si:publish-subscribe-channel id="serviceBindingControlChannel" task-executor="executor"/>
+	<si:service-activator input-channel="serviceBindingControlChannel">
+		<bean class="org.springframework.integration.controlbus.ControlBusMessageHandler"/>
+	</si:service-activator> 
+	<!-- END SERVICE BINDING CONFIG-->
+	
+	<!-- CONTROL BUS channel, exposed as an OSGi service --> 
+	<si:publish-subscribe-channel id="controlMessagesDistributionChannel"/>
+	<!-- END CONTROL BUS channel, exposed as an OSGi service --> 
+	
+	<si:header-value-router input-channel="controlMessagesDistributionChannel" header-name="INTEGRATION_EVENT_TYPE">
+		<si:mapping value="REGISTRATION" channel="serviceRegistrationControlChannel"/>
+		<si:mapping value="UNREGISTRATION" channel="serviceRegistrationControlChannel"/>
+		<si:mapping value="BINDING" channel="serviceBindingControlChannel"/>
+		<si:mapping value="UNBINDING" channel="serviceBindingControlChannel"/>
+	</si:header-value-router>
+	
+	<si:thread-pool-task-executor id="executor" core-size="10"/>
+</beans>
diff --git a/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/stubs/OSGiMockUtils.java b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/stubs/OSGiMockUtils.java
new file mode 100644
index 0000000000..99b53da463
--- /dev/null
+++ b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/stubs/OSGiMockUtils.java
@@ -0,0 +1,117 @@
+/*
+ * 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.osgi.stubs;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Dictionary;
+import java.util.Enumeration;
+import java.util.List;
+import java.util.Set;
+import java.util.StringTokenizer;
+
+import org.osgi.framework.ServiceReference;
+import org.springframework.osgi.util.internal.MapBasedDictionary;
+import org.springframework.util.StringUtils;
+
+/**
+ * @author Oleg Zhurakousky
+ * @since 2.0
+ */
+public class OSGiMockUtils {
+
+	public static Dictionary parseFilterIntoDictionary(String filter){
+		Dictionary properties = new MapBasedDictionary();
+		String filterToEvaluate = null;
+		if (filter != null){
+			StringTokenizer tokenizer = new StringTokenizer(filter, "()&!");
+			while(tokenizer.hasMoreTokens()){
+				String nextToken = tokenizer.nextToken();
+				String name = nextToken.substring(0, nextToken.indexOf("="));
+				String value = nextToken.substring(nextToken.indexOf("=")+1);
+				properties.put(name, value);
+			}
+		}	
+		return properties;
+	}
+	/**
+	 * 
+	 * @param originalFilter
+	 * @param valueToAdd
+	 */
+	public static String addToFilter(String originalFilter, String valueToAdd) {
+		MapBasedDictionary original = new MapBasedDictionary(OSGiMockUtils.parseFilterIntoDictionary(originalFilter));
+		Dictionary added = OSGiMockUtils.parseFilterIntoDictionary(valueToAdd);
+		original.putAll(added);
+		StringBuffer newFilter = new StringBuffer("(&");
+		Enumeration emum = original.keys();
+		while (emum.hasMoreElements()) {
+			Object key = (Object) emum.nextElement();
+			Object value = original.get(key);
+			newFilter.append("(");
+			newFilter.append(key);
+			newFilter.append("=");
+			newFilter.append(value);
+			newFilter.append(")");
+		}
+		newFilter.append(")");
+		return newFilter.toString();
+	}
+	/**
+	 * 
+	 * @param allServiceReferences
+	 * @param filter
+	 * @return
+	 */
+	public static ServiceReference[] buildFilteredServiceReferences(Set<ServiceReference> allServiceReferences, String filter){
+		Dictionary properties = OSGiMockUtils.parseFilterIntoDictionary(filter);
+		ArrayList<ServiceReference> filteredReferences = new ArrayList<ServiceReference>();
+		
+		for (ServiceReference sr : allServiceReferences) {	
+			Enumeration keys = properties.keys();
+			boolean match = true;
+			inner:
+			while (keys.hasMoreElements()) {
+				String key = (String) keys.nextElement();
+				Object value = properties.get(key);
+				Object compareToValue = sr.getProperty(key);
+				if (compareToValue != null && compareToValue instanceof String[] && key.equals("objectClass")){
+					List<String> srImplementedClasses  = Arrays.asList((String[])compareToValue);
+					List<String> filteredClasses  = Arrays.asList(StringUtils.commaDelimitedListToStringArray((String)value));
+					match = srImplementedClasses.containsAll(filteredClasses);
+					if (!match){
+						break inner;
+					}
+				} else if (!value.equals(compareToValue)){
+					match = false;
+					break inner;
+				}
+			}
+			if (match){
+				filteredReferences.add(sr);
+			}
+		}
+		ServiceReference[] references = null;
+		if (filteredReferences.size() > 0){
+			references = new ServiceReference[filteredReferences.size()];
+			for (int i = 0; i < filteredReferences.size(); i++) {
+				references[i] = filteredReferences.get(i);
+			}
+		}		
+		return references;
+	}
+	
+}
diff --git a/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/stubs/SIBundleContextStub.java b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/stubs/SIBundleContextStub.java
new file mode 100644
index 0000000000..b64657ee28
--- /dev/null
+++ b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/stubs/SIBundleContextStub.java
@@ -0,0 +1,180 @@
+/*
+ * 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.osgi.stubs;
+
+import java.util.Arrays;
+import java.util.Dictionary;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Hashtable;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.osgi.framework.InvalidSyntaxException;
+import org.osgi.framework.ServiceEvent;
+import org.osgi.framework.ServiceFactory;
+import org.osgi.framework.ServiceListener;
+import org.osgi.framework.ServiceReference;
+import org.osgi.framework.ServiceRegistration;
+import org.springframework.osgi.mock.MockBundleContext;
+import org.springframework.osgi.mock.MockServiceReference;
+import org.springframework.osgi.util.internal.MapBasedDictionary;
+
+/**
+ * Mock BundleContext to be used for testing
+ * 
+ * @author Oleg Zhurakousky
+ * @since 2.0
+ */
+public class SIBundleContextStub extends MockBundleContext {
+	private static final Log log = LogFactory.getLog(SIBundleContextStub.class);
+	private static  SIBundleContextStub bundleContext = new SIBundleContextStub();
+	
+	private Map<ServiceReference, Object> services = new HashMap<ServiceReference, Object>();
+	private Set<ServiceReference> serviceReferences = new HashSet<ServiceReference>();
+	private Map<MapBasedDictionary, ServiceListener> serviceListenerMap = 
+							new HashMap<MapBasedDictionary, ServiceListener>();
+	
+	/**
+	 * 
+	 * @return
+	 */
+	public static SIBundleContextStub getInstance(){	
+		log.debug("Returning Stubed BundleContext");
+		return bundleContext;
+	}
+	/**
+	 * 
+	 * @return
+	 */
+	public static SIBundleContextStub getNewInstance(){	
+		log.debug("Returning newly created Stubed BundleContext");
+		return new SIBundleContextStub();
+	}
+	/**
+	 * 
+	 */
+	@SuppressWarnings("unchecked")
+	public ServiceRegistration registerService(String[] clazzes, Object service, Dictionary properties) {
+		if (properties instanceof Hashtable){
+			properties = new MapBasedDictionary(properties);
+		}
+		log.info("Registering SERVICE: " + service);
+		SIServiceRegistrationStub reg = new SIServiceRegistrationStub(clazzes, properties);
+		reg.setBundleContext(this);
+
+		MockServiceReference ref = new MockServiceReference(this.getBundle(), properties, reg, clazzes);
+		for (int i = 0; i < clazzes.length; i++) {
+			serviceReferences.add(ref);
+		}
+		if (service instanceof ServiceFactory){
+			service = ((ServiceFactory)service).getService(this.getBundle(), reg);
+		}
+		reg.setReference(ref);
+		services.put(ref, service);
+		final ServiceEvent event = new ServiceEvent(ServiceEvent.REGISTERED, ref);
+		final Set<ServiceListener> listeners = this.getFilteredListeners(properties);
+		for (final ServiceListener listener : listeners) {
+//			Thread t = new Thread(new Runnable() {
+//				public void run() {
+					listener.serviceChanged(event);
+//				}
+//			});
+//			t.start();
+		}
+		log.debug("Service: " + ref + " is registered");
+		return reg;
+	}
+	/**
+	 * 
+	 */
+	public Object getService(ServiceReference sr){
+		return services.get(sr);
+	}
+	/**
+	 * 
+	 */
+	public ServiceReference getServiceReference(String serviceName){
+		String filter = "(&(objectClass=" + serviceName + "))";
+		ServiceReference[] references = OSGiMockUtils.buildFilteredServiceReferences(serviceReferences, filter);
+		return references == null ? null : references[0];
+	}
+	/**
+	 * 
+	 */
+	public ServiceReference[] getServiceReferences(String serviceName, String filter) throws InvalidSyntaxException {
+		if (serviceName != null){
+			String classFilter = "(&(objectClass=" + serviceName + "))";
+			filter = OSGiMockUtils.addToFilter(filter, classFilter);
+		}	
+		ServiceReference[] references = OSGiMockUtils.buildFilteredServiceReferences(serviceReferences, filter);
+		return references;
+	}
+	/**
+	 * 
+	 */
+	@SuppressWarnings("unchecked")
+	public void addServiceListener(ServiceListener listener) {
+		MapBasedDictionary properties = new MapBasedDictionary();
+		serviceListenerMap.put(properties, listener);
+	}
+	/**
+	 * 
+	 */
+	public void addServiceListener(ServiceListener listener, String filter) throws InvalidSyntaxException {
+		MapBasedDictionary properties = (MapBasedDictionary) OSGiMockUtils.parseFilterIntoDictionary(filter);
+		serviceListenerMap.put(properties, listener);
+	}
+	@SuppressWarnings("unchecked")
+	public Set<ServiceListener> getFilteredListeners(Dictionary properties){
+		Set filteredListeners = new HashSet<ServiceListener>();
+		if (properties != null && properties.size() > 0){	
+			for (Dictionary filter : serviceListenerMap.keySet()) {
+				MapBasedDictionary listenerFilter = new MapBasedDictionary(filter);
+				
+				log.debug("Trying to match filter properties: " + listenerFilter);
+				MapBasedDictionary inFilter = new MapBasedDictionary(properties);
+				log.debug("Current filter entry: " + inFilter);
+				boolean objecClassMatch = true;
+				if (listenerFilter.containsKey("objectClass")){
+					objecClassMatch = this.matchObjectClass(listenerFilter, inFilter);
+				}
+				listenerFilter.remove("objectClass");
+				inFilter.remove("objectClass");
+				if (inFilter.keySet().containsAll(listenerFilter.keySet()) &&
+					inFilter.values().containsAll(listenerFilter.values()) && objecClassMatch){
+					filteredListeners.add(serviceListenerMap.get(filter));
+					log.debug("Found listener for properties: " + listenerFilter + " -  " + serviceListenerMap.get(filter));
+				}
+			}				
+		}
+		return filteredListeners;
+	}
+	
+	public boolean matchObjectClass(MapBasedDictionary listenerFilter, MapBasedDictionary inFilter){
+		String interfaze = (String) listenerFilter.get("objectClass");
+		String[] interfaces = (String[]) inFilter.get("objectClass");
+		
+		return Arrays.binarySearch(interfaces, interfaze) >=0;
+	}
+	
+	public void removeService(ServiceReference sr){
+		services.remove(sr);
+		serviceReferences.remove(sr);
+	}
+}
diff --git a/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/stubs/SIBundleContextStubTest.java b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/stubs/SIBundleContextStubTest.java
new file mode 100644
index 0000000000..31fea698ff
--- /dev/null
+++ b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/stubs/SIBundleContextStubTest.java
@@ -0,0 +1,83 @@
+/*
+ * 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.osgi.stubs;
+
+import java.io.Serializable;
+import java.util.Dictionary;
+import java.util.Hashtable;
+
+import junit.framework.Assert;
+
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.osgi.framework.BundleContext;
+import org.osgi.framework.ServiceEvent;
+import org.osgi.framework.ServiceListener;
+import org.osgi.framework.ServiceReference;
+import org.osgi.framework.ServiceRegistration;
+
+/**
+ * @author Oleg Zhurakousky
+ * @since 2.0
+ */
+public class SIBundleContextStubTest {
+
+	@Test
+	public void validateServiceRegistrationWithNoFilters(){
+		BundleContext context = SIBundleContextStub.getInstance();
+		context.registerService(Serializable.class.getName(), new Serializable(){}, null);
+
+		ServiceReference sr = context.getServiceReference(Serializable.class.getName());
+		Assert.assertNotNull(sr);
+		Serializable service = (Serializable) context.getService(sr);
+		Assert.assertNotNull(service);
+	}
+	@Test
+	public void validateServiceRegistrationWithFiltersNegativeReturn() throws Exception {
+		BundleContext context = SIBundleContextStub.getInstance();
+		Dictionary<String, String> properties = new Hashtable<String, String>();
+		properties.put("name", "foo");
+		context.registerService(Serializable.class.getName(), new Serializable(){}, properties);
+	
+		ServiceReference[] srs = context.getServiceReferences(Serializable.class.getName(), "(&(name=bar))");
+		Assert.assertTrue("Expected no ServiceReferences", srs == null);
+	}
+	@Test
+	public void validateServiceRegistrationWithFiltersPositiveReturn() throws Exception {
+		BundleContext context = SIBundleContextStub.getInstance();
+		Dictionary<String, String> properties = new Hashtable<String, String>();
+		properties.put("name", "foo");
+		context.registerService(Serializable.class.getName(), new Serializable(){}, properties);
+		
+		properties = new Hashtable<String, String>();
+		properties.put("name", "bar");
+		context.registerService(Serializable.class.getName(), new Serializable(){}, properties);
+	
+		ServiceReference[] srs = context.getServiceReferences(Serializable.class.getName(), "(&(name=bar))");
+		Assert.assertNotNull(srs);
+	}
+	@Test
+	public void validateServiceListenerRegistrationWithFilter() throws Exception{
+		BundleContext context = SIBundleContextStub.getInstance();
+		ServiceListener sl = Mockito.mock(ServiceListener.class);
+		context.addServiceListener(sl, "(&(name=foo))");
+		Dictionary<String, String> properties = new Hashtable<String, String>();
+		properties.put("name", "foo");
+		ServiceRegistration sr = 
+			context.registerService(Serializable.class.getName(), new Serializable(){}, properties);
+		Mockito.verify(sl, Mockito.times(1)).serviceChanged((ServiceEvent) Mockito.anyObject());
+	}
+}
diff --git a/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/stubs/SIServiceRegistrationStub.java b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/stubs/SIServiceRegistrationStub.java
new file mode 100644
index 0000000000..b544240031
--- /dev/null
+++ b/org.springframework.integration.osgi/src/test/java/org/springframework/integration/osgi/stubs/SIServiceRegistrationStub.java
@@ -0,0 +1,52 @@
+/*
+ * 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.osgi.stubs;
+
+import java.util.Dictionary;
+import java.util.Set;
+
+import org.osgi.framework.ServiceEvent;
+import org.osgi.framework.ServiceListener;
+import org.osgi.framework.ServiceReference;
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.osgi.mock.MockServiceRegistration;
+
+/**
+ * TODO - insert COMMENT
+ * @author Oleg Zhurakousky
+ * @since 2.0
+ */
+public class SIServiceRegistrationStub extends MockServiceRegistration {
+	private SIBundleContextStub context;
+	
+	public SIServiceRegistrationStub(String[] clazz, Dictionary props) {
+		super(clazz, props);
+	}
+	public void setBundleContext(SIBundleContextStub context){
+		this.context = context;
+	}
+	public void unregister() {
+		ServiceReference ref = this.getReference();
+		DirectFieldAccessor refAccessor = new DirectFieldAccessor(ref);
+		Dictionary properties = (Dictionary) refAccessor.getPropertyValue("properties");
+		
+		context.removeService(this.getReference());
+		Set<ServiceListener> listeners = context.getFilteredListeners(properties);
+		for (ServiceListener serviceListener : listeners) {
+			serviceListener.serviceChanged(new ServiceEvent(ServiceEvent.UNREGISTERING, ref));
+		}
+	}
+}
diff --git a/org.springframework.integration.osgi/template.mf b/org.springframework.integration.osgi/template.mf
index 8b8951662b..5cb104d33a 100644
--- a/org.springframework.integration.osgi/template.mf
+++ b/org.springframework.integration.osgi/template.mf
@@ -2,9 +2,12 @@ Bundle-SymbolicName: org.springframework.integration.osgi
 Bundle-Name: Spring Integration OSGi Extender
 Bundle-Vendor: SpringSource
 Bundle-ManifestVersion: 2
+Import-Package: org.springframework.integration.handler;version="[2.0.0, 2.0.1)",
+ org.springframework.integration.config;version="[2.0.0, 2.0.1)",
+ org.springframework.integration.router;version="[2.0.0, 2.0.1)",
+ org.springframework.integration.endpoint;version="[2.0.0, 2.0.1)"
 Import-Template: org.springframework.integration.*;version="[2.0.0, 2.0.1)",
- com.sun.*;version="0",
- javax.xml.*;version="0",
+ org.springframework.osgi.*;version="[1.2.0, 2.0.1)",
  org.apache.commons.logging.*;version="1.1.1",
  org.springframework.*;version="[3.0.0,4.0.0)",
  org.w3c.dom.*;version="0"