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

Migrated JMS adapter and parser code from "org.springframework.integration.adapter" to the new "org.springframework.integration.jms" module, and added a dedicated spring-integration-jms-1.0.xsd schema and JmsNamespaceHandler.

Browse Source
This commit is contained in:
Mark Fisher
2008-09-20 15:37:21 +00:00
parent c0e134c4a9
commit 82630b96dc
55 changed files with 417 additions and 330 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd
View File

@@ -65,102 +65,6 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="jms-source">
<xsd:annotation>
<xsd:documentation>
Defines a JMS-based source channel adapter.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="jmsInboundAdapterType">
<xsd:attribute name="header-mapper" type="xsd:string"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="jms-gateway">
<xsd:annotation>
<xsd:documentation>
Defines a JMS-based gateway adapter.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="jmsInboundAdapterType">
<xsd:attribute name="message-converter" type="xsd:string"/>
<xsd:attribute name="expect-reply" type="xsd:boolean" default="false"/>
<xsd:attribute name="request-channel" type="xsd:string" use="required"/>
<xsd:attribute name="reply-channel" type="xsd:string"/>
<xsd:attribute name="request-timeout" type="xsd:long"/>
<xsd:attribute name="reply-timeout" type="xsd:long"/>
<xsd:attribute name="transaction-manager" type="xsd:string"/>
<xsd:attribute name="concurrent-consumers" type="xsd:string"/>
<xsd:attribute name="max-concurrent-consumers" type="xsd:string"/>
<xsd:attribute name="max-messages-per-task" type="xsd:string"/>
<xsd:attribute name="idle-task-execution-limit" type="xsd:string"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="jms-target">
<xsd:annotation>
<xsd:documentation>
Defines a target that sends JMS Messages.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="jmsAdapterType">
<xsd:attribute name="header-mapper" type="xsd:string"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:complexType name="jmsInboundAdapterType">
<xsd:annotation>
<xsd:documentation>
Common configuration for inbound JMS-based adapters.
</xsd:documentation>
</xsd:annotation>
<xsd:complexContent>
<xsd:extension base="jmsAdapterType">
<xsd:attribute name="acknowledge" default="auto">
<xsd:annotation>
<xsd:documentation><![CDATA[
The native JMS acknowledge mode: "auto", "client", "dups-ok" or "transacted".
The latter effectively activates a locally transacted Session.
]]></xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:restriction base="xsd:NMTOKEN">
<xsd:enumeration value="auto"/>
<xsd:enumeration value="client"/>
<xsd:enumeration value="dups-ok"/>
<xsd:enumeration value="transacted"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
<xsd:complexType name="jmsAdapterType">
<xsd:annotation>
<xsd:documentation>
Common configuration for JMS-based adapters.
</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:complexType>
<xsd:element name="rmi-gateway">
<xsd:complexType>
<xsd:annotation>

133
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/jms/AbstractJmsTemplateBasedAdapter.java
View File

@@ -1,133 +0,0 @@
/*
* 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.adapter.jms;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.adapter.MessageHeaderMapper;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.support.converter.MessageConverter;
/**
* Base class for adapters that delegate to a {@link JmsTemplate}.
*
* @author Mark Fisher
*/
public abstract class AbstractJmsTemplateBasedAdapter implements InitializingBean {
private volatile ConnectionFactory connectionFactory;
private volatile Destination destination;
private volatile String destinationName;
private volatile JmsTemplate jmsTemplate;
private volatile MessageHeaderMapper<javax.jms.Message> headerMapper;
private volatile boolean initialized;
private final Object initializationMonitor = new Object();
public AbstractJmsTemplateBasedAdapter(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
public AbstractJmsTemplateBasedAdapter(ConnectionFactory connectionFactory, Destination destination) {
this.connectionFactory = connectionFactory;
this.destination = destination;
this.jmsTemplate = createDefaultJmsTemplate();
}
public AbstractJmsTemplateBasedAdapter(ConnectionFactory connectionFactory, String destinationName) {
this.connectionFactory = connectionFactory;
this.destinationName = destinationName;
this.jmsTemplate = createDefaultJmsTemplate();
}
/**
* No-arg constructor provided for convenience when configuring with
* setters. Note that the initialization callback will validate.
*/
public AbstractJmsTemplateBasedAdapter() {
}
public void setConnectionFactory(ConnectionFactory connectionFactory) {
this.connectionFactory = connectionFactory;
}
public void setDestination(Destination destination) {
this.destination = destination;
}
public void setDestinationName(String destinationName) {
this.destinationName = destinationName;
}
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
public void setHeaderMapper(MessageHeaderMapper<javax.jms.Message> headerMapper) {
this.headerMapper = headerMapper;
}
protected JmsTemplate getJmsTemplate() {
if (this.jmsTemplate == null) {
this.afterPropertiesSet();
}
return this.jmsTemplate;
}
public void afterPropertiesSet() {
synchronized (this.initializationMonitor) {
if (this.initialized) {
return;
}
if (this.jmsTemplate == null) {
if (this.connectionFactory == null || (this.destination == null && this.destinationName == null)) {
throw new ConfigurationException("Either a 'jmsTemplate' or " +
"*both* 'connectionFactory' and 'destination' (or 'destination-name') are required.");
}
this.jmsTemplate = this.createDefaultJmsTemplate();
}
MessageConverter converter = this.jmsTemplate.getMessageConverter();
converter = (converter != null && converter instanceof HeaderMappingMessageConverter) ?
converter : new HeaderMappingMessageConverter(converter, this.headerMapper);
this.jmsTemplate.setMessageConverter(converter);
this.initialized = true;
}
}
private JmsTemplate createDefaultJmsTemplate() {
JmsTemplate jmsTemplate = new JmsTemplate();
jmsTemplate.setConnectionFactory(this.connectionFactory);
if (this.destination != null) {
jmsTemplate.setDefaultDestination(this.destination);
}
else {
jmsTemplate.setDefaultDestinationName(this.destinationName);
}
return jmsTemplate;
}
}

64
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/jms/ChannelPublishingJmsListener.java
View File

@@ -1,64 +0,0 @@
/*
* 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.adapter.jms;
import javax.jms.MessageListener;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageChannelTemplate;
import org.springframework.integration.message.MessagingException;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.util.Assert;
/**
* JMS {@link MessageListener} implementation that converts the received JMS
* message into a Spring Integration message and then sends that to a channel.
*
* @author Mark Fisher
*/
public class ChannelPublishingJmsListener implements MessageListener {
private final MessageChannel channel;
private final MessageConverter converter;
private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate();
public ChannelPublishingJmsListener(MessageChannel channel, MessageConverter converter) {
Assert.notNull(channel, "channel must not be null");
this.channel = channel;
this.converter = (converter != null && converter instanceof HeaderMappingMessageConverter) ?
converter : new HeaderMappingMessageConverter(converter);
}
public void onMessage(javax.jms.Message jmsMessage) {
try {
Message<?> messageToSend = (Message<?>) this.converter.fromMessage(jmsMessage);
if (!this.channelTemplate.send(messageToSend, this.channel)) {
throw new MessageDeliveryException(messageToSend, "failed to send Message to channel: " + this.channel);
}
}
catch (Exception e) {
throw new MessagingException("failed to convert and send JMS Message", e);
}
}
}

124
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/jms/DefaultJmsHeaderMapper.java
View File

@@ -1,124 +0,0 @@
/*
* 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.adapter.jms;
import java.util.Arrays;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import javax.jms.Destination;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.adapter.MessageHeaderMapper;
import org.springframework.integration.message.MessageHeaders;
import org.springframework.util.StringUtils;
/**
* A {@link HeaderMapper} implementation for JMS {@link javax.jms.Message Messages}.
*
* @author Mark Fisher
*/
public class DefaultJmsHeaderMapper implements MessageHeaderMapper<javax.jms.Message> {
private static List<Class<?>> SUPPORTED_PROPERTY_TYPES = Arrays.asList(new Class<?>[] {
Boolean.class, Byte.class, Double.class, Float.class, Integer.class, Long.class, Short.class, String.class });
private final Log logger = LogFactory.getLog(this.getClass());
public void mapFromMessageHeaders(MessageHeaders headers, javax.jms.Message jmsMessage) {
try {
Object jmsCorrelationId = headers.get(JmsHeaders.CORRELATION_ID);
if (jmsCorrelationId != null && (jmsCorrelationId instanceof String)) {
jmsMessage.setJMSCorrelationID((String) jmsCorrelationId);
}
Object jmsReplyTo = headers.get(JmsHeaders.REPLY_TO);
if (jmsReplyTo != null && (jmsReplyTo instanceof Destination)) {
jmsMessage.setJMSReplyTo((Destination) jmsReplyTo);
}
Object jmsType = headers.get(JmsHeaders.TYPE);
if (jmsType != null && (jmsType instanceof String)) {
jmsMessage.setJMSType((String) jmsType);
}
String prefix = JmsHeaders.USER_PREFIX;
Set<String> attributeNames = headers.keySet();
for (String attributeName : attributeNames) {
if (attributeName.startsWith(prefix)) {
String jmsAttributeName = attributeName.substring(prefix.length());
if (StringUtils.hasText(attributeName)) {
Object value = headers.get(attributeName);
if (value != null && SUPPORTED_PROPERTY_TYPES.contains(value.getClass())) {
try {
jmsMessage.setObjectProperty(jmsAttributeName, value);
}
catch (Throwable t) {
if (logger.isWarnEnabled()) {
logger.warn("failed to map property '" + jmsAttributeName + "' from MessageHeader", t);
}
}
}
}
}
}
}
catch (Throwable t) {
if (logger.isWarnEnabled()) {
logger.warn("error occurred while mapping properties from MessageHeader", t);
}
}
}
public Map<String, Object> mapToMessageHeaders(javax.jms.Message jmsMessage) {
Map<String, Object> headers = new HashMap<String, Object>();
try {
String correlationId = jmsMessage.getJMSCorrelationID();
if (correlationId != null) {
headers.put(JmsHeaders.CORRELATION_ID, correlationId);
}
Destination replyTo = jmsMessage.getJMSReplyTo();
if (replyTo != null) {
headers.put(JmsHeaders.REPLY_TO, replyTo);
}
headers.put(JmsHeaders.REDELIVERED, jmsMessage.getJMSRedelivered());
String type = jmsMessage.getJMSType();
if (type != null) {
headers.put(JmsHeaders.TYPE, type);
}
Enumeration<?> jmsPropertyNames = jmsMessage.getPropertyNames();
if (jmsPropertyNames != null) {
while (jmsPropertyNames.hasMoreElements()) {
String propertyName = jmsPropertyNames.nextElement().toString();
headers.put(JmsHeaders.USER_PREFIX + propertyName,
jmsMessage.getObjectProperty(propertyName));
}
}
}
catch (Throwable t) {
if (logger.isWarnEnabled()) {
logger.warn("error occurred while mapping properties to MessageHeader", t);
}
}
return headers;
}
}

73
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/jms/HeaderMappingMessageConverter.java
View File

@@ -1,73 +0,0 @@
/*
* 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.adapter.jms;
import java.util.Map;
import javax.jms.JMSException;
import javax.jms.Session;
import org.springframework.integration.adapter.MessageHeaderMapper;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessagingException;
import org.springframework.jms.support.converter.MessageConversionException;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.converter.SimpleMessageConverter;
/**
* A {@link MessageConverter} implementation that delegates to an existing
* converter as well as an implementation of {@link MessageHeaderMapper}.
*
* @author Mark Fisher
*/
public class HeaderMappingMessageConverter implements MessageConverter {
private final MessageConverter converter;
private final MessageHeaderMapper<javax.jms.Message> headerMapper;
public HeaderMappingMessageConverter(MessageConverter converter) {
this(converter, null);
}
public HeaderMappingMessageConverter(MessageConverter converter, MessageHeaderMapper<javax.jms.Message> headerMapper) {
this.converter = (converter != null ? converter : new SimpleMessageConverter());
this.headerMapper = (headerMapper != null ? headerMapper : new DefaultJmsHeaderMapper());
}
public Object fromMessage(javax.jms.Message jmsMessage) throws JMSException, MessageConversionException {
Object payload = this.converter.fromMessage(jmsMessage);
Map<String, Object> headerMap = this.headerMapper.mapToMessageHeaders(jmsMessage);
Message<?> message = MessageBuilder.withPayload(payload).copyHeaders(headerMap).build();
return message;
}
public javax.jms.Message toMessage(Object object, Session session) throws JMSException, MessageConversionException {
if (!(object instanceof Message<?>)) {
throw new MessagingException("expected a '" + Message.class.getName() +
"', but received '" + object.getClass() + "'");
}
Message<?> message = (Message<?>) object;
javax.jms.Message jmsMessage = this.converter.toMessage(message.getPayload(), session);
this.headerMapper.mapFromMessageHeaders(message.getHeaders(), jmsMessage);
return jmsMessage;
}
}

197
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/jms/JmsGateway.java
View File

@@ -1,197 +0,0 @@
/*
* 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.adapter.jms;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.Session;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.Lifecycle;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.gateway.SimpleMessagingGateway;
import org.springframework.jms.listener.AbstractMessageListenerContainer;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.listener.adapter.MessageListenerAdapter;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.converter.SimpleMessageConverter;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.util.Assert;
/**
* A message-driven adapter for receiving JMS messages and sending to a channel.
*
* @author Mark Fisher
*/
public class JmsGateway extends SimpleMessagingGateway implements Lifecycle, DisposableBean {
private volatile AbstractMessageListenerContainer container;
private volatile ConnectionFactory connectionFactory;
private volatile Destination destination;
private volatile String destinationName;
private volatile MessageConverter messageConverter;
private volatile TaskExecutor taskExecutor;
private volatile PlatformTransactionManager transactionManager;
private volatile boolean sessionTransacted;
private volatile int sessionAcknowledgeMode = Session.AUTO_ACKNOWLEDGE;
private volatile int concurrentConsumers = 1;
private volatile int maxConcurrentConsumers = 1;
private volatile int maxMessagesPerTask = Integer.MIN_VALUE;
private volatile int idleTaskExecutionLimit = 1;
private boolean expectReply = false;
public void setContainer(AbstractMessageListenerContainer container) {
this.container = container;
}
public void setConnectionFactory(ConnectionFactory connectionFactory) {
this.connectionFactory = connectionFactory;
}
public void setDestination(Destination destination) {
this.destination = destination;
}
public void setDestinationName(String destinationName) {
this.destinationName = destinationName;
}
public void setMessageConverter(MessageConverter messageConverter) {
Assert.notNull(messageConverter, "'messageConverter' must not be null");
this.messageConverter = messageConverter;
}
public void setTaskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
}
public void setTransactionManager(PlatformTransactionManager transactionManager) {
this.transactionManager = transactionManager;
}
public void setSessionTransacted(boolean sessionTransacted) {
this.sessionTransacted = sessionTransacted;
}
public void setSessionAcknowledgeMode(int sessionAcknowledgeMode) {
this.sessionAcknowledgeMode = sessionAcknowledgeMode;
}
public void setExpectReply(boolean expectReply) {
this.expectReply = expectReply;
}
public void setConcurrentConsumers(int concurrentConsumers) {
this.concurrentConsumers = concurrentConsumers;
}
public void setMaxConcurrentConsumers(int maxConcurrentConsumers) {
this.maxConcurrentConsumers = maxConcurrentConsumers;
}
public void setMaxMessagesPerTask(int maxMessagesPerTask) {
this.maxMessagesPerTask = maxMessagesPerTask;
}
public void setIdleTaskExecutionLimit(int idleTaskExecutionLimit) {
this.idleTaskExecutionLimit = idleTaskExecutionLimit;
}
private void initialize() {
if (this.container == null) {
this.container = createDefaultContainer();
}
MessageListenerAdapter listener = new MessageListenerAdapter();
listener.setDelegate(this);
listener.setDefaultListenerMethod(this.expectReply ? "sendAndReceive" : "send");
if (this.messageConverter == null) {
this.messageConverter = new SimpleMessageConverter();
}
if (!(this.messageConverter instanceof HeaderMappingMessageConverter)) {
this.messageConverter = new HeaderMappingMessageConverter(this.messageConverter);
}
listener.setMessageConverter(this.messageConverter);
this.container.setMessageListener(listener);
if (!this.container.isActive()) {
this.container.afterPropertiesSet();
}
}
private AbstractMessageListenerContainer createDefaultContainer() {
if (this.connectionFactory == null || (this.destination == null && this.destinationName == null)) {
throw new ConfigurationException("If a 'container' reference is not provided, then "
+ "'connectionFactory' and 'destination' (or 'destinationName') are required.");
}
DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer();
dmlc.setConcurrentConsumers(this.concurrentConsumers);
dmlc.setMaxConcurrentConsumers(this.maxConcurrentConsumers);
dmlc.setMaxMessagesPerTask(this.maxMessagesPerTask);
dmlc.setIdleTaskExecutionLimit(this.idleTaskExecutionLimit);
dmlc.setConnectionFactory(this.connectionFactory);
if (this.destination != null) {
dmlc.setDestination(this.destination);
}
if (this.destinationName != null) {
dmlc.setDestinationName(this.destinationName);
}
dmlc.setTransactionManager(this.transactionManager);
dmlc.setSessionTransacted(this.sessionTransacted);
dmlc.setSessionAcknowledgeMode(this.sessionAcknowledgeMode);
dmlc.setAutoStartup(false);
if (this.taskExecutor != null) {
dmlc.setTaskExecutor(this.taskExecutor);
}
return dmlc;
}
public boolean isRunning() {
return (this.container != null && this.container.isRunning());
}
public void start() {
this.initialize();
this.container.start();
}
public void stop() {
if (this.container != null) {
this.container.stop();
}
}
public void destroy() {
if (this.container != null) {
this.container.destroy();
}
}
}

46
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/jms/JmsHeaders.java
View File

@@ -1,46 +0,0 @@
/*
* 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.adapter.jms;
/**
* Pre-defined names and prefixes to be used for setting and/or retrieving JMS attributes
* from/to integration Message Headers.
*
* @author Mark Fisher
*/
public abstract class JmsHeaders {
/**
* Prefix for any message header that should be passed for usage by the JMS transport.
*/
public static final String TRANSPORT_PREFIX = "spring.integration.transport.jms.";
/**
* Prefix for any user-defined message header that should be passed within JMS properties.
*/
public static final String USER_PREFIX = "spring.integration.user.jms.";
public static final String CORRELATION_ID = TRANSPORT_PREFIX + "JMSCorrelationID";
public static final String REPLY_TO = TRANSPORT_PREFIX + "JMSReplyTo";
public static final String REDELIVERED = TRANSPORT_PREFIX + "JMSRedelivered";
public static final String TYPE = TRANSPORT_PREFIX + "JMSType";
}

61
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/jms/JmsSource.java
View File

@@ -1,61 +0,0 @@
/*
* 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.adapter.jms;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.PollableSource;
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, the
* {@link JmsGateway} that uses Spring's MessageListener
* container support is highly recommended.
*
* @author Mark Fisher
*/
public class JmsSource extends AbstractJmsTemplateBasedAdapter implements PollableSource<Object> {
public JmsSource(JmsTemplate jmsTemplate) {
super(jmsTemplate);
}
public JmsSource(ConnectionFactory connectionFactory, Destination destination) {
super(connectionFactory, destination);
}
public JmsSource(ConnectionFactory connectionFactory, String destinationName) {
super(connectionFactory, destinationName);
}
public Message<Object> receive() {
Object receivedObject = this.getJmsTemplate().receiveAndConvert();
if (receivedObject == null) {
return null;
}
if (receivedObject instanceof Message) {
return (Message<Object>) receivedObject;
}
return new GenericMessage<Object>(receivedObject);
}
}

36
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/jms/JmsTarget.java
View File

@@ -1,36 +0,0 @@
/*
* 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.adapter.jms;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageConsumer;
/**
* A target for sending JMS Messages.
*
* @author Mark Fisher
*/
public class JmsTarget extends AbstractJmsTemplateBasedAdapter implements MessageConsumer {
public final void onMessage(final Message<?> message) {
if (message == null) {
throw new IllegalArgumentException("message must not be null");
}
this.getJmsTemplate().convertAndSend(message);
}
}

104
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/jms/config/JmsAdapterParserUtils.java
View File

@@ -1,104 +0,0 @@
/*
* 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.adapter.jms.config;
import javax.jms.Session;
import org.w3c.dom.Element;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.util.StringUtils;
/**
* Utility methods and constants for JMS adapter parsers.
*
* @author Mark Fisher
*/
public abstract class JmsAdapterParserUtils {
public static final String JMS_TEMPLATE_ATTRIBUTE = "jms-template";
public static final String JMS_TEMPLATE_PROPERTY = "jmsTemplate";
public static final String CONNECTION_FACTORY_ATTRIBUTE = "connection-factory";
public static final String CONNECTION_FACTORY_PROPERTY = "connectionFactory";
public static final String DESTINATION_ATTRIBUTE = "destination";
public static final String DESTINATION_PROPERTY = "destination";
public static final String DESTINATION_NAME_ATTRIBUTE = "destination-name";
public static final String DESTINATION_NAME_PROPERTY = "destinationName";
public static final String HEADER_MAPPER_ATTRIBUTE = "header-mapper";
public static final String HEADER_MAPPER_PROPERTY = "headerMapper";
public static final String MESSAGE_CONVERTER_ATTRIBUTE = "message-converter";
public static final String MESSAGE_CONVERTER_PROPERTY = "messageConverter";
private static final String ACKNOWLEDGE_ATTRIBUTE = "acknowledge";
private static final String ACKNOWLEDGE_AUTO = "auto";
private static final String ACKNOWLEDGE_CLIENT = "client";
private static final String ACKNOWLEDGE_DUPS_OK = "dups-ok";
private static final String ACKNOWLEDGE_TRANSACTED = "transacted";
public static String determineConnectionFactoryBeanName(Element element) {
String connectionFactoryBeanName = "connectionFactory";
if (element.hasAttribute(CONNECTION_FACTORY_ATTRIBUTE)) {
connectionFactoryBeanName = element.getAttribute(CONNECTION_FACTORY_ATTRIBUTE);
if (!StringUtils.hasText(connectionFactoryBeanName)) {
throw new BeanCreationException(
"JMS adapter 'connection-factory' attribute must not be empty");
}
}
return connectionFactoryBeanName;
}
public static Integer parseAcknowledgeMode(Element element) {
String acknowledge = element.getAttribute(ACKNOWLEDGE_ATTRIBUTE);
if (StringUtils.hasText(acknowledge)) {
int acknowledgeMode = Session.AUTO_ACKNOWLEDGE;
if (ACKNOWLEDGE_TRANSACTED.equals(acknowledge)) {
acknowledgeMode = Session.SESSION_TRANSACTED;
}
else if (ACKNOWLEDGE_DUPS_OK.equals(acknowledge)) {
acknowledgeMode = Session.DUPS_OK_ACKNOWLEDGE;
}
else if (ACKNOWLEDGE_CLIENT.equals(acknowledge)) {
acknowledgeMode = Session.CLIENT_ACKNOWLEDGE;
}
else if (!ACKNOWLEDGE_AUTO.equals(acknowledge)) {
throw new BeanCreationException("Invalid JMS 'acknowledge' setting: " +
"only \"auto\", \"client\", \"dups-ok\" and \"transacted\" supported.");
}
return acknowledgeMode;
}
else {
return null;
}
}
}

119
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/jms/config/JmsGatewayParser.java
View File

@@ -1,119 +0,0 @@
/*
* 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.adapter.jms.config;
import javax.jms.Session;
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.beans.factory.xml.ParserContext;
import org.springframework.integration.adapter.jms.JmsGateway;
import org.springframework.integration.config.IntegrationNamespaceUtils;
import org.springframework.util.StringUtils;
/**
* Parser for the &lt;jms-gateway&gt; element.
*
* @author Mark Fisher
*/
public class JmsGatewayParser extends AbstractSingleBeanDefinitionParser {
@Override
protected Class<?> getBeanClass(Element element) {
return JmsGateway.class;
}
@Override
protected boolean shouldGenerateId() {
return false;
}
@Override
protected boolean shouldGenerateIdAsFallback() {
return true;
}
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
String destination = element.getAttribute(JmsAdapterParserUtils.DESTINATION_ATTRIBUTE);
String destinationName = element.getAttribute(JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE);
String messageConverter = element.getAttribute(JmsAdapterParserUtils.MESSAGE_CONVERTER_ATTRIBUTE);
if (StringUtils.hasText(element.getAttribute(JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE))) {
throw new BeanCreationException(JmsGateway.class.getSimpleName() +
" does not accept a '" + JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE +
"' reference. One of '" + JmsAdapterParserUtils.DESTINATION_ATTRIBUTE + "' or '" +
JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "' must be provided.");
}
if (StringUtils.hasText(destination) || StringUtils.hasText(destinationName)) {
builder.addPropertyReference(JmsAdapterParserUtils.CONNECTION_FACTORY_PROPERTY,
JmsAdapterParserUtils.determineConnectionFactoryBeanName(element));
if (StringUtils.hasText(destination)) {
builder.addPropertyReference(JmsAdapterParserUtils.DESTINATION_PROPERTY, destination);
}
else {
builder.addPropertyValue(JmsAdapterParserUtils.DESTINATION_NAME_PROPERTY, destinationName);
}
}
else {
throw new BeanCreationException("One of '" + JmsAdapterParserUtils.DESTINATION_ATTRIBUTE +
"' or '" + JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "' must be provided.");
}
if (StringUtils.hasText(messageConverter)) {
builder.addPropertyReference(JmsAdapterParserUtils.MESSAGE_CONVERTER_PROPERTY, messageConverter);
}
Integer acknowledgeMode = JmsAdapterParserUtils.parseAcknowledgeMode(element);
if (acknowledgeMode != null) {
if (acknowledgeMode.intValue() == Session.SESSION_TRANSACTED) {
builder.addPropertyValue("sessionTransacted", Boolean.TRUE);
}
else {
builder.addPropertyValue("sessionAcknowledgeMode", acknowledgeMode);
}
}
String transactionManager = element.getAttribute("transaction-manager");
if (StringUtils.hasText(transactionManager)) {
builder.addPropertyReference("transactionManager", transactionManager);
}
String requestChannel = element.getAttribute("request-channel");
if (StringUtils.hasText(requestChannel)) {
builder.addPropertyReference("requestChannel", requestChannel);
}
String requestTimeout = element.getAttribute("request-timeout");
if (StringUtils.hasText(requestTimeout)) {
builder.addPropertyValue("requestTimeout", Long.parseLong(requestTimeout));
}
String replyChannel = element.getAttribute("reply-channel");
if (StringUtils.hasText(replyChannel)) {
builder.addPropertyReference("replyChannel", replyChannel);
}
String replyTimeout = element.getAttribute("reply-timeout");
if (StringUtils.hasText(replyTimeout)) {
builder.addPropertyValue("replyTimeout", Long.parseLong(replyTimeout));
}
if ("true".equals(element.getAttribute("expect-reply"))) {
builder.addPropertyValue("expectReply", Boolean.TRUE);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "concurrent-consumers");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-concurrent-consumers");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-messages-per-task");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "idle-task-execution-limit");
}
}

83
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/jms/config/JmsSourceParser.java
View File

@@ -1,83 +0,0 @@
/*
* 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.adapter.jms.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.adapter.jms.JmsSource;
import org.springframework.util.StringUtils;
/**
* Parser for the &lt;jms-source/&gt; element.
*
* @author Mark Fisher
*/
public class JmsSourceParser extends AbstractBeanDefinitionParser {
protected boolean shouldGenerateId() {
return false;
}
protected boolean shouldGenerateIdAsFallback() {
return true;
}
@Override
protected AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(JmsSource.class);
String jmsTemplate = element.getAttribute(JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE);
String destination = element.getAttribute(JmsAdapterParserUtils.DESTINATION_ATTRIBUTE);
String destinationName = element.getAttribute(JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE);
String headerMapper = element.getAttribute(JmsAdapterParserUtils.HEADER_MAPPER_ATTRIBUTE);
if (StringUtils.hasText(jmsTemplate)) {
if (element.hasAttribute(JmsAdapterParserUtils.CONNECTION_FACTORY_ATTRIBUTE) ||
element.hasAttribute(JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE) ||
element.hasAttribute(JmsAdapterParserUtils.DESTINATION_ATTRIBUTE)) {
throw new BeanCreationException(
"When providing '" + JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE +
"', none of '" + JmsAdapterParserUtils.CONNECTION_FACTORY_ATTRIBUTE +
"', '" + JmsAdapterParserUtils.DESTINATION_ATTRIBUTE + "', or '" +
JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "' should be provided.");
}
builder.addConstructorArgReference(jmsTemplate);
}
else if (StringUtils.hasText(destination) || StringUtils.hasText(destinationName)) {
builder.addConstructorArgReference(JmsAdapterParserUtils.determineConnectionFactoryBeanName(element));
if (StringUtils.hasText(destination)) {
builder.addConstructorArgReference(destination);
}
else if (StringUtils.hasText(destinationName)) {
builder.addConstructorArgValue(destinationName);
}
}
else {
throw new BeanCreationException("either a '" + JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE + "' or one of '" +
JmsAdapterParserUtils.DESTINATION_ATTRIBUTE + "' or '" + JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE +
"' attributes must be provided for a polling JMS adapter");
}
if (StringUtils.hasText(headerMapper)) {
builder.addPropertyReference(JmsAdapterParserUtils.HEADER_MAPPER_PROPERTY, headerMapper);
}
return builder.getBeanDefinition();
}
}

80
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/jms/config/JmsTargetParser.java
View File

@@ -1,80 +0,0 @@
/*
* 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.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.beans.factory.xml.ParserContext;
import org.springframework.integration.adapter.jms.JmsTarget;
import org.springframework.util.StringUtils;
/**
* Parser for the &lt;jms-target/&gt; element.
*
* @author Mark Fisher
*/
public class JmsTargetParser extends AbstractSingleBeanDefinitionParser {
protected Class<?> getBeanClass(Element element) {
return JmsTarget.class;
}
protected boolean shouldGenerateId() {
return false;
}
protected boolean shouldGenerateIdAsFallback() {
return true;
}
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
String jmsTemplate = element.getAttribute(JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE);
String destination = element.getAttribute(JmsAdapterParserUtils.DESTINATION_ATTRIBUTE);
String destinationName = element.getAttribute(JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE);
String headerMapper = element.getAttribute(JmsAdapterParserUtils.HEADER_MAPPER_ATTRIBUTE);
if (StringUtils.hasText(jmsTemplate)) {
if (element.hasAttribute(JmsAdapterParserUtils.CONNECTION_FACTORY_ATTRIBUTE) ||
element.hasAttribute(JmsAdapterParserUtils.DESTINATION_ATTRIBUTE) ||
element.hasAttribute(JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE)) {
throw new BeanCreationException("when providing a 'jms-template' reference, none of " +
"'connection-factory', 'destination', or 'destination-name' should be provided.");
}
builder.addPropertyReference(JmsAdapterParserUtils.JMS_TEMPLATE_PROPERTY, jmsTemplate);
}
else if (StringUtils.hasText(destination) ^ StringUtils.hasText(destinationName)) {
builder.addPropertyReference(JmsAdapterParserUtils.CONNECTION_FACTORY_PROPERTY,
JmsAdapterParserUtils.determineConnectionFactoryBeanName(element));
if (StringUtils.hasText(destination)) {
builder.addPropertyReference(JmsAdapterParserUtils.DESTINATION_PROPERTY, destination);
}
else {
builder.addPropertyValue(JmsAdapterParserUtils.DESTINATION_NAME_PROPERTY, destinationName);
}
}
else {
throw new BeanCreationException("Either a 'jms-template' reference or " +
"one of 'destination' or 'destination-name' attributes must be provided.");
}
if (StringUtils.hasText(headerMapper)) {
builder.addPropertyReference(JmsAdapterParserUtils.HEADER_MAPPER_PROPERTY, headerMapper);
}
}
}

3
org.springframework.integration.adapter/src/main/resources/META-INF/spring-integration.parsers
View File

@@ -3,9 +3,6 @@ file-target=org.springframework.integration.adapter.file.config.FileTargetParser
ftp-source=org.springframework.integration.adapter.ftp.config.FtpSourceParser
httpinvoker-gateway=org.springframework.integration.adapter.httpinvoker.config.HttpInvokerGatewayParser
httpinvoker-handler=org.springframework.integration.adapter.httpinvoker.config.HttpInvokerHandlerParser
jms-gateway=org.springframework.integration.adapter.jms.config.JmsGatewayParser
jms-source=org.springframework.integration.adapter.jms.config.JmsSourceParser
jms-target=org.springframework.integration.adapter.jms.config.JmsTargetParser
mail-target=org.springframework.integration.adapter.mail.config.MailTargetParser
polling-mail-source=org.springframework.integration.adapter.mail.config.PollingMailSourceParser
imap-idle-mail-source=org.springframework.integration.adapter.mail.config.SubscribableImapIdleMailSourceParser