Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Added namespace support for JMS source adapters (JmsPollingSourceAdapter and JmsMessageDrivenSourceAdapter).

Browse Source
This commit is contained in:
Mark Fisher
2008-01-08 01:47:37 +00:00
parent 86965e0e2e
commit 67d6e8aa2c
11 changed files with 217 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java
View File

@@ -75,6 +75,12 @@ public class AbstractSourceAdapter<T> implements SourceAdapter, InitializingBean
}
protected boolean sendToChannel(T object) {
if (object == null) {
if (logger.isDebugEnabled()) {
logger.debug("adapter attempted to send a null object");
}
return false;
}
Message<?> message = null;
if (object instanceof Message<?>) {
message = (Message<?>) object;

4
spring-integration-core/src/main/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParser.java
View File

@@ -29,12 +29,12 @@ import org.springframework.integration.adapter.file.FileSourceAdapter;
*/
public class FileSourceAdapterParser extends AbstractSingleBeanDefinitionParser {
protected Class getBeanClass(Element element) {
protected Class<?> getBeanClass(Element element) {
return FileSourceAdapter.class;
}
protected boolean shouldGenerateId() {
return true;
return false;
}
protected void doParse(Element element, BeanDefinitionBuilder builder) {

4
spring-integration-core/src/main/java/org/springframework/integration/adapter/file/config/FileTargetAdapterParser.java
View File

@@ -29,12 +29,12 @@ import org.springframework.integration.adapter.file.FileTargetAdapter;
*/
public class FileTargetAdapterParser extends AbstractSingleBeanDefinitionParser {
protected Class getBeanClass(Element element) {
protected Class<?> getBeanClass(Element element) {
return FileTargetAdapter.class;
}
protected boolean shouldGenerateId() {
return true;
return false;
}
protected void doParse(Element element, BeanDefinitionBuilder builder) {

8
spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java
View File

@@ -22,7 +22,7 @@ import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.Lifecycle;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.MessageHandlingException;
@@ -41,7 +41,7 @@ import org.springframework.util.Assert;
* @author Mark Fisher
*/
public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter<Object> implements MessageListener, Lifecycle,
InitializingBean {
DisposableBean {
private AbstractJmsListeningContainer container;
@@ -133,6 +133,10 @@ public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter<Object>
container.stop();
}
public void destroy() {
container.destroy();
}
public void onMessage(Message message) {
try {
this.sendToChannel(messageConverter.fromMessage(message));

32
spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java
View File

@@ -29,9 +29,9 @@ import org.springframework.jms.core.JmsTemplate;
/**
* A source for receiving JMS Messages with a polling listener. This source is
* only recommended for very low message volume. Otherwise, an event-driven
* option that uses one of Spring's MessageListener containers is highly
* recommended.
* only recommended for very low message volume. Otherwise, the
* {@link JmsMessageDrivenSourceAdapter} that uses Spring's MessageListener
* container support is highly recommended.
*
* @author Mark Fisher
*/
@@ -41,6 +41,8 @@ public class JmsPollableSource implements PollableSource<Object>, InitializingBe
private Destination destination;
private String destinationName;
private JmsTemplate jmsTemplate;
@@ -54,6 +56,12 @@ public class JmsPollableSource implements PollableSource<Object>, InitializingBe
this.initJmsTemplate();
}
public JmsPollableSource(ConnectionFactory connectionFactory, String destinationName) {
this.connectionFactory = connectionFactory;
this.destinationName = destinationName;
this.initJmsTemplate();
}
/**
* No-arg constructor provided for convenience when configuring with
* setters. Note that the initialization callback will validate.
@@ -69,15 +77,19 @@ public class JmsPollableSource implements PollableSource<Object>, InitializingBe
this.destination = destination;
}
public void setDestinationName(String destinationName) {
this.destinationName = destinationName;
}
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
public void afterPropertiesSet() {
if (this.jmsTemplate == null) {
if (this.connectionFactory == null || this.destination == null) {
if (this.connectionFactory == null || (this.destination == null && this.destinationName == null)) {
throw new MessagingConfigurationException("Either a 'jmsTemplate' or "
+ "*both* 'connectionFactory' and 'destination' are required.");
+ "both 'connectionFactory' and 'destination' (or 'destinationName') are required.");
}
this.initJmsTemplate();
}
@@ -86,7 +98,15 @@ public class JmsPollableSource implements PollableSource<Object>, InitializingBe
private void initJmsTemplate() {
this.jmsTemplate = new JmsTemplate();
this.jmsTemplate.setConnectionFactory(this.connectionFactory);
this.jmsTemplate.setDefaultDestination(this.destination);
if (this.destination != null) {
this.jmsTemplate.setDefaultDestination(this.destination);
}
else if (this.destinationName != null) {
this.jmsTemplate.setDefaultDestinationName(this.destinationName);
}
else {
throw new MessagingConfigurationException("either 'destination' or 'destinationName' is required");
}
}
public Collection<Object> poll(int limit) {

12
spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsPollingSourceAdapter.java
View File

@@ -29,12 +29,16 @@ import org.springframework.jms.core.JmsTemplate;
*/
public class JmsPollingSourceAdapter extends PollingSourceAdapter<Object> {
public JmsPollingSourceAdapter(ConnectionFactory connectionFactory, Destination destination) {
super(new JmsPollableSource(connectionFactory, destination));
}
public JmsPollingSourceAdapter(JmsTemplate jmsTemplate) {
super(new JmsPollableSource(jmsTemplate));
}
public JmsPollingSourceAdapter(ConnectionFactory connectionFactory, Destination destination) {
super(new JmsPollableSource(connectionFactory, destination));
}
public JmsPollingSourceAdapter(ConnectionFactory connectionFactory, String destinationName) {
super(new JmsPollableSource(connectionFactory, destinationName));
}
}

141
spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParser.java Normal file
View File

@@ -0,0 +1,141 @@
/*
* Copyright 2002-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.adapter.jms.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.integration.adapter.jms.JmsMessageDrivenSourceAdapter;
import org.springframework.integration.adapter.jms.JmsPollingSourceAdapter;
import org.springframework.util.StringUtils;
/**
* Parser for the &lt;jms-source/&gt; element.
*
* @author Mark Fisher
*/
public class JmsSourceAdapterParser extends AbstractSingleBeanDefinitionParser {
private static final String JMS_TEMPLATE_ATTRIBUTE = "jms-template";
private static final String CONNECTION_FACTORY_ATTRIBUTE = "connection-factory";
private static final String CONNECTION_FACTORY_PROPERTY = "connectionFactory";
private static final String DESTINATION_ATTRIBUTE = "destination";
private static final String DESTINATION_PROPERTY = "destination";
private static final String DESTINATION_NAME_ATTRIBUTE = "destination-name";
private static final String DESTINATION_NAME_PROPERTY = "destinationName";
private static final String CHANNEL_ATTRIBUTE = "channel";
private static final String CHANNEL_PROPERTY = "channel";
private static final String POLL_PERIOD_ATTRIBUTE = "poll-period";
private static final String POLL_PERIOD_PROPERTY = "period";
protected Class<?> getBeanClass(Element element) {
if (StringUtils.hasText(element.getAttribute(POLL_PERIOD_ATTRIBUTE))) {
return JmsPollingSourceAdapter.class;
}
return JmsMessageDrivenSourceAdapter.class;
}
protected boolean shouldGenerateId() {
return false;
}
protected void doParse(Element element, BeanDefinitionBuilder builder) {
if (builder.getBeanDefinition().getBeanClass().equals(JmsPollingSourceAdapter.class)) {
parsePollingSourceAdapter(element, builder);
}
else {
parseMessageDrivenSourceAdapter(element, builder);
}
String channel = element.getAttribute(CHANNEL_ATTRIBUTE);
builder.addPropertyReference(CHANNEL_PROPERTY, channel);
}
private void parsePollingSourceAdapter(Element element, BeanDefinitionBuilder builder) {
String pollPeriod = element.getAttribute(POLL_PERIOD_ATTRIBUTE);
if (!StringUtils.hasText(pollPeriod)) {
throw new BeanCreationException("'" + POLL_PERIOD_ATTRIBUTE +
"' is required for a " + JmsPollingSourceAdapter.class.getSimpleName());
}
builder.addPropertyValue(POLL_PERIOD_PROPERTY, pollPeriod);
String jmsTemplate = element.getAttribute(JMS_TEMPLATE_ATTRIBUTE);
String connectionFactory = element.getAttribute(CONNECTION_FACTORY_ATTRIBUTE);
String destination = element.getAttribute(DESTINATION_ATTRIBUTE);
String destinationName = element.getAttribute(DESTINATION_NAME_ATTRIBUTE);
if (StringUtils.hasText(jmsTemplate)) {
if (StringUtils.hasText(connectionFactory) || StringUtils.hasText(destination) || StringUtils.hasText(destinationName)) {
throw new BeanCreationException("when providing '" + JMS_TEMPLATE_ATTRIBUTE +
"', none of '" + CONNECTION_FACTORY_ATTRIBUTE + "', '" + DESTINATION_ATTRIBUTE +
"', or '" + DESTINATION_NAME_ATTRIBUTE + "' should be provided.");
}
builder.addConstructorArgReference(jmsTemplate);
}
else if (StringUtils.hasText(connectionFactory) && (StringUtils.hasText(destination) || StringUtils.hasText(destinationName))) {
builder.addConstructorArgReference(connectionFactory);
if (StringUtils.hasText(destination)) {
builder.addConstructorArgReference(destination);
}
else if (StringUtils.hasText(destinationName)) {
builder.addConstructorArg(destinationName);
}
}
else {
throw new BeanCreationException("either a '" + JMS_TEMPLATE_ATTRIBUTE + "' or both '" +
CONNECTION_FACTORY_ATTRIBUTE + "' and '" + DESTINATION_ATTRIBUTE + "' (or '" +
DESTINATION_NAME_ATTRIBUTE + "') attributes must be provided for a " +
JmsPollingSourceAdapter.class.getSimpleName());
}
}
private void parseMessageDrivenSourceAdapter(Element element, BeanDefinitionBuilder builder) {
String connectionFactory = element.getAttribute(CONNECTION_FACTORY_ATTRIBUTE);
String destination = element.getAttribute(DESTINATION_ATTRIBUTE);
String destinationName = element.getAttribute(DESTINATION_NAME_ATTRIBUTE);
if (StringUtils.hasText(element.getAttribute(JMS_TEMPLATE_ATTRIBUTE))) {
throw new BeanCreationException(JmsMessageDrivenSourceAdapter.class.getSimpleName() +
" does not accept a '" + JMS_TEMPLATE_ATTRIBUTE + "' reference, both " +
"'" + CONNECTION_FACTORY_ATTRIBUTE + "' and '" + DESTINATION_ATTRIBUTE +
"' (or '" + DESTINATION_NAME_ATTRIBUTE + "') must be provided.");
}
if (StringUtils.hasText(connectionFactory) && (StringUtils.hasText(destination) || StringUtils.hasText(destinationName))) {
builder.addPropertyReference(CONNECTION_FACTORY_PROPERTY, connectionFactory);
if (StringUtils.hasText(destination)) {
builder.addPropertyReference(DESTINATION_PROPERTY, destination);
}
else {
builder.addPropertyValue(DESTINATION_NAME_PROPERTY, destinationName);
}
}
else {
throw new BeanCreationException("Both '" + CONNECTION_FACTORY_ATTRIBUTE + "' and '" +
DESTINATION_ATTRIBUTE + "' (or '" + DESTINATION_NAME_ATTRIBUTE + "') must be provided.");
}
}
}

2
spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/config/JmsTargetAdapterParser.java
View File

@@ -53,7 +53,7 @@ public class JmsTargetAdapterParser extends AbstractSingleBeanDefinitionParser {
}
protected boolean shouldGenerateId() {
return true;
return false;
}
protected void doParse(Element element, BeanDefinitionBuilder builder) {

2
spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java
View File

@@ -19,6 +19,7 @@ package org.springframework.integration.config;
import org.springframework.beans.factory.xml.NamespaceHandlerSupport;
import org.springframework.integration.adapter.file.config.FileSourceAdapterParser;
import org.springframework.integration.adapter.file.config.FileTargetAdapterParser;
import org.springframework.integration.adapter.jms.config.JmsSourceAdapterParser;
import org.springframework.integration.adapter.jms.config.JmsTargetAdapterParser;
/**
@@ -37,6 +38,7 @@ public class IntegrationNamespaceHandler extends NamespaceHandlerSupport {
registerBeanDefinitionParser("endpoint", new EndpointParser());
registerBeanDefinitionParser("file-source", new FileSourceAdapterParser());
registerBeanDefinitionParser("file-target", new FileTargetAdapterParser());
registerBeanDefinitionParser("jms-source", new JmsSourceAdapterParser());
registerBeanDefinitionParser("jms-target", new JmsTargetAdapterParser());
}

20
spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd
View File

@@ -141,6 +141,7 @@
Defines a file-based source channel adapter.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="directory" type="xsd:string" use="required"/>
<xsd:attribute name="channel" type="xsd:string" use="required"/>
<xsd:attribute name="poll-period" type="xsd:int" use="required"/>
@@ -154,11 +155,29 @@
Defines a file-based target channel adapter.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="directory" type="xsd:string" use="required"/>
<xsd:attribute name="channel" type="xsd:string" use="required"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="jms-source">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines a jms-based source channel adapter.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="jms-template" type="xsd:string"/>
<xsd:attribute name="connection-factory" type="xsd:string"/>
<xsd:attribute name="destination" type="xsd:string"/>
<xsd:attribute name="destination-name" type="xsd:string"/>
<xsd:attribute name="channel" type="xsd:string" use="required"/>
<xsd:attribute name="poll-period" type="xsd:int"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="jms-target">
<xsd:complexType>
<xsd:annotation>
@@ -166,6 +185,7 @@
Defines a jms-based target channel adapter.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="jms-template" type="xsd:string"/>
<xsd:attribute name="connection-factory" type="xsd:string"/>
<xsd:attribute name="destination" type="xsd:string"/>