INT-869. Committed initial support for OSGi Control Bus

This commit is contained in:
Oleg Zhurakousky
2009-11-03 10:59:56 +00:00
parent e560a5fdbc
commit a62f2cf60c
25 changed files with 1111 additions and 16 deletions

View File

@@ -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>

View File

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

View File

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

View File

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

View File

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

View File

@@ -1 +1 @@
http\://www.springframework.org/schema/integration-bus=org.springframework.integration.osgi.config.xml.IntegrationOSGiNamespaceHandler
http\://www.springframework.org/schema/integration/integration-control-bus=org.springframework.integration.osgi.config.xml.IntegrationOSGiControlBusNamespaceHandler

View File

@@ -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

View File

@@ -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>

View File

@@ -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>

View File

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

View File

@@ -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>

View File

@@ -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>

View File

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

View File

@@ -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>

View File

@@ -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>

View File

@@ -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>

View File

@@ -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){
}
}
}

View File

@@ -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>

View File

@@ -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");
}
}

View File

@@ -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>

View File

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

View File

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

View File

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

View File

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

View File

@@ -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"