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

Migrating adapters to spring-integration-adapters (INT-83).

Browse Source
This commit is contained in:
Mark Fisher
2008-02-21 22:52:22 +00:00
parent 7208c5ec24
commit 1f2bf5abc0
41 changed files with 3344 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -0,0 +1,82 @@
<?xml version="1.0" encoding="UTF-8"?>
<xsd:schema xmlns="http://www.springframework.org/schema/integration"
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"
elementFormDefault="qualified"
attributeFormDefault="unqualified">
<xsd:include schemaLocation="http://www.springframework.org/schema/integration/spring-integration-1.0.xsd"/>
<xsd:import namespace="http://www.springframework.org/schema/beans"/>
<xsd:import namespace="http://www.springframework.org/schema/tool"/>
<xsd:annotation>
<xsd:documentation><![CDATA[
Defines the configuration elements for Spring Integration adapters.
]]></xsd:documentation>
</xsd:annotation>
<xsd:element name="file-source">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
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"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="file-target">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
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:attribute name="message-converter" type="xsd:string"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="jms-target">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
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"/>
<xsd:attribute name="destination-name" type="xsd:string"/>
<xsd:attribute name="channel" type="xsd:string" use="required"/>
</xsd:complexType>
</xsd:element>
</xsd:schema>

96
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/ChannelPublishingJmsListener.java Normal file
View File

@@ -0,0 +1,96 @@
/*
* 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;
import javax.jms.JMSException;
import javax.jms.MessageListener;
import org.springframework.integration.MessagingConfigurationException;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageMapper;
import org.springframework.integration.message.SimplePayloadMessageMapper;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.converter.SimpleMessageConverter;
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 MessageChannel channel;
private long timeout = -1;
private MessageConverter converter = new SimpleMessageConverter();
private MessageMapper mapper = new SimplePayloadMessageMapper();
public ChannelPublishingJmsListener() {
}
public ChannelPublishingJmsListener(MessageChannel channel) {
Assert.notNull(channel, "'channel' must not be null");
this.channel = channel;
}
public void setChannel(MessageChannel channel) {
Assert.notNull(channel, "'channel' must not be null");
this.channel = channel;
}
public void setTimeout(long timeout) {
this.timeout = timeout;
}
public void setMessageConverter(MessageConverter messageConverter) {
Assert.notNull(messageConverter, "'messageConverter' must not be null");
this.converter = messageConverter;
}
public void setMessageMapper(MessageMapper messageMapper) {
Assert.notNull(messageMapper, "'messageMapper' must not be null");
this.mapper = messageMapper;
}
public void onMessage(javax.jms.Message jmsMessage) {
if (this.channel == null) {
throw new MessagingConfigurationException("'channel' must not be null");
}
try {
Object payload = converter.fromMessage(jmsMessage);
Message messageToSend = mapper.toMessage(payload);
if (this.timeout < 0) {
this.channel.send(messageToSend);
}
else {
this.channel.send(messageToSend, timeout);
}
}
catch (JMSException e) {
throw new MessageDeliveryException("failed to convert JMS Message", e);
}
}
}

48
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/DefaultJmsMessagePostProcessor.java Normal file
View File

@@ -0,0 +1,48 @@
/*
* 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;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import org.springframework.integration.message.MessageHeader;
/**
* A {@link JmsMessagePostProcessor} that passes attributes from a
* {@link MessageHeader} to a JMS message before it is sent to its destination.
*
* @author Mark Fisher
*/
public class DefaultJmsMessagePostProcessor implements JmsMessagePostProcessor {
public void postProcessJmsMessage(Message jmsMessage, MessageHeader header) throws JMSException {
Object jmsCorrelationId = header.getAttribute(JmsTargetAdapter.JMS_CORRELATION_ID);
if (jmsCorrelationId != null && (jmsCorrelationId instanceof String)) {
jmsMessage.setJMSCorrelationID((String) jmsCorrelationId);
}
Object jmsReplyTo = header.getAttribute(JmsTargetAdapter.JMS_REPLY_TO);
if (jmsReplyTo != null && (jmsReplyTo instanceof Destination)) {
jmsMessage.setJMSReplyTo((Destination) jmsReplyTo);
}
Object jmsType = header.getAttribute(JmsTargetAdapter.JMS_TYPE);
if (jmsType != null && (jmsType instanceof String)) {
jmsMessage.setJMSType((String) jmsType);
}
}
}

149
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java Normal file
View File

@@ -0,0 +1,149 @@
/*
* 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;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.Lifecycle;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.MessagingConfigurationException;
import org.springframework.integration.adapter.AbstractSourceAdapter;
import org.springframework.jms.listener.AbstractJmsListeningContainer;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.converter.SimpleMessageConverter;
import org.springframework.util.Assert;
/**
* A message-driven adapter for receiving JMS messages and sending to a channel.
*
* @author Mark Fisher
*/
public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter<Object> implements Lifecycle, DisposableBean {
private AbstractJmsListeningContainer container;
private ConnectionFactory connectionFactory;
private Destination destination;
private String destinationName;
private MessageConverter messageConverter = new SimpleMessageConverter();
private TaskExecutor taskExecutor;
private long receiveTimeout = 1000;
private int concurrentConsumers = 1;
private int maxConcurrentConsumers = 1;
private int maxMessagesPerTask = Integer.MIN_VALUE;
private int idleTaskExecutionLimit = 1;
private long sendTimeout = -1;
public void setContainer(AbstractJmsListeningContainer 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 setSendTimeout(long sendTimeout) {
this.sendTimeout = sendTimeout;
}
public void setTaskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
}
@Override
public void initialize() {
if (this.container == null) {
initDefaultContainer();
}
}
private void initDefaultContainer() {
if (this.connectionFactory == null || (this.destination == null && this.destinationName == null)) {
throw new MessagingConfigurationException("If a 'container' reference is not provided, then "
+ "'connectionFactory' and 'destination' (or 'destinationName') are required.");
}
DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer();
dmlc.setConnectionFactory(this.connectionFactory);
if (this.destination != null) {
dmlc.setDestination(this.destination);
}
if (this.destinationName != null) {
dmlc.setDestinationName(this.destinationName);
}
dmlc.setReceiveTimeout(this.receiveTimeout);
dmlc.setConcurrentConsumers(this.concurrentConsumers);
dmlc.setMaxConcurrentConsumers(this.maxConcurrentConsumers);
dmlc.setMaxMessagesPerTask(this.maxMessagesPerTask);
dmlc.setIdleTaskExecutionLimit(this.idleTaskExecutionLimit);
dmlc.setSessionTransacted(true);
dmlc.setAutoStartup(false);
ChannelPublishingJmsListener listener = new ChannelPublishingJmsListener(this.getChannel());
listener.setMessageConverter(this.messageConverter);
listener.setMessageMapper(this.getMessageMapper());
listener.setTimeout(this.sendTimeout);
dmlc.setMessageListener(listener);
if (this.taskExecutor != null) {
dmlc.setTaskExecutor(this.taskExecutor);
}
dmlc.afterPropertiesSet();
this.container = dmlc;
}
public boolean isRunning() {
return container.isRunning();
}
public void start() {
container.start();
}
public void stop() {
container.stop();
}
public void destroy() {
container.destroy();
}
}

34
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsMessagePostProcessor.java Normal file
View File

@@ -0,0 +1,34 @@
/*
* 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;
import javax.jms.JMSException;
import javax.jms.Message;
import org.springframework.integration.message.MessageHeader;
/**
* Strategy interface for post-processing a JMS Message before it is sent to its
* destination.
*
* @author Mark Fisher
*/
public interface JmsMessagePostProcessor {
void postProcessJmsMessage(Message jmsMessage, MessageHeader header) throws JMSException;
}

116
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java Normal file
View File

@@ -0,0 +1,116 @@
/*
* 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;
import java.util.Arrays;
import java.util.Collection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.MessagingConfigurationException;
import org.springframework.integration.adapter.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 JmsMessageDrivenSourceAdapter} that uses Spring's MessageListener
* container support is highly recommended.
*
* @author Mark Fisher
*/
public class JmsPollableSource implements PollableSource<Object>, InitializingBean {
private ConnectionFactory connectionFactory;
private Destination destination;
private String destinationName;
private JmsTemplate jmsTemplate;
public JmsPollableSource(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
public JmsPollableSource(ConnectionFactory connectionFactory, Destination destination) {
this.connectionFactory = connectionFactory;
this.destination = destination;
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.
*/
public JmsPollableSource() {
}
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 afterPropertiesSet() {
if (this.jmsTemplate == null) {
if (this.connectionFactory == null || (this.destination == null && this.destinationName == null)) {
throw new MessagingConfigurationException("Either a 'jmsTemplate' or "
+ "both 'connectionFactory' and 'destination' (or 'destinationName') are required.");
}
this.initJmsTemplate();
}
}
private void initJmsTemplate() {
this.jmsTemplate = new JmsTemplate();
this.jmsTemplate.setConnectionFactory(this.connectionFactory);
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) {
return Arrays.asList(this.jmsTemplate.receiveAndConvert());
}
}

44
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollingSourceAdapter.java Normal file
View File

@@ -0,0 +1,44 @@
/*
* Copyright 2002-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.adapter.jms;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import org.springframework.integration.adapter.PollingSourceAdapter;
import org.springframework.jms.core.JmsTemplate;
/**
* A convenience adapter that wraps a {@link JmsPollableSource}.
*
* @author Mark Fisher
*/
public class JmsPollingSourceAdapter extends PollingSourceAdapter<Object> {
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));
}
}

139
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsTargetAdapter.java Normal file
View File

@@ -0,0 +1,139 @@
/*
* 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;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.MessagingConfigurationException;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessagePostProcessor;
import org.springframework.jms.support.converter.SimpleMessageConverter;
/**
* A target adapter for sending JMS Messages.
*
* @author Mark Fisher
*/
public class JmsTargetAdapter implements MessageHandler, InitializingBean {
public static final String JMS_CORRELATION_ID = "JMSCorrelationID";
public static final String JMS_REPLY_TO = "JMSReplyTo";
public static final String JMS_TYPE = "JMSType";
private ConnectionFactory connectionFactory;
private Destination destination;
private String destinationName;
private JmsTemplate jmsTemplate;
private JmsMessagePostProcessor jmsMessagePostProcessor = new DefaultJmsMessagePostProcessor();
public JmsTargetAdapter(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
public JmsTargetAdapter(ConnectionFactory connectionFactory, Destination destination) {
this.connectionFactory = connectionFactory;
this.destination = destination;
this.initJmsTemplate();
}
public JmsTargetAdapter(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.
*/
public JmsTargetAdapter() {
}
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 setJmsMessagePostProcessor(JmsMessagePostProcessor jmsMessagePostProcessor) {
this.jmsMessagePostProcessor = jmsMessagePostProcessor;
}
public void afterPropertiesSet() {
if (this.jmsTemplate == null) {
if (this.connectionFactory == null || (this.destination == null && this.destinationName == null)) {
throw new MessagingConfigurationException("Either a 'jmsTemplate' or " +
"*both* 'connectionFactory' and 'destination' (or 'destination-name') are required.");
}
this.initJmsTemplate();
}
if (this.jmsTemplate.getMessageConverter() == null) {
this.jmsTemplate.setMessageConverter(new SimpleMessageConverter());
}
}
private void initJmsTemplate() {
this.jmsTemplate = new JmsTemplate();
this.jmsTemplate.setConnectionFactory(this.connectionFactory);
if (this.destination != null) {
this.jmsTemplate.setDefaultDestination(this.destination);
}
else {
this.jmsTemplate.setDefaultDestinationName(this.destinationName);
}
}
public final Message<?> handle(final Message<?> message) {
if (message == null) {
return null;
}
this.jmsTemplate.convertAndSend(message.getPayload(), new MessagePostProcessor() {
public javax.jms.Message postProcessMessage(javax.jms.Message jmsMessage) throws JMSException {
if (jmsMessagePostProcessor != null) {
jmsMessagePostProcessor.postProcessJmsMessage(jmsMessage, message.getHeader());
}
return jmsMessage;
}
});
return null;
}
}

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

@@ -0,0 +1,158 @@
/*
* 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";
private static final String MESSAGE_CONVERTER_ATTRIBUTE = "message-converter";
private static final String MESSAGE_CONVERTER_PROPERTY = "messageConverter";
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 boolean shouldGenerateIdAsFallback() {
return true;
}
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());
}
if (StringUtils.hasText(element.getAttribute(MESSAGE_CONVERTER_ATTRIBUTE))) {
throw new BeanCreationException("The '" + MESSAGE_CONVERTER_ATTRIBUTE + "' attribute is not supported for a " +
JmsPollingSourceAdapter.class.getSimpleName() + ". Consider providing a '" + JMS_TEMPLATE_ATTRIBUTE +
"' reference where the template contains a 'messageConverter' property instead.");
}
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);
String messageConverter = element.getAttribute(MESSAGE_CONVERTER_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.");
}
if (StringUtils.hasText(messageConverter)) {
builder.addPropertyReference(MESSAGE_CONVERTER_PROPERTY, messageConverter);
}
}
}

109
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsTargetAdapterParser.java Normal file
View File

@@ -0,0 +1,109 @@
/*
* 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.config.RuntimeBeanReference;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.adapter.jms.JmsTargetAdapter;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.scheduling.Subscription;
import org.springframework.util.StringUtils;
/**
* Parser for the &lt;jms-target/&gt; element.
*
* @author Mark Fisher
*/
public class JmsTargetAdapterParser extends AbstractSingleBeanDefinitionParser {
private static final String JMS_TEMPLATE_ATTRIBUTE = "jms-template";
private static final String JMS_TEMPLATE_PROPERTY = "jmsTemplate";
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_NAME_ATTRIBUTE = "destination-name";
private static final String DESTINATION_PROPERTY = "destination";
private static final String DESTINATION_NAME_PROPERTY = "destinationName";
private static final String CHANNEL_ATTRIBUTE = "channel";
private static final String HANDLER_PROPERTY = "handler";
private static final String SUBSCRIPTION_PROPERTY = "subscription";
protected Class<?> getBeanClass(Element element) {
return DefaultMessageEndpoint.class;
}
protected boolean shouldGenerateId() {
return false;
}
protected boolean shouldGenerateIdAsFallback() {
return true;
}
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
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);
RootBeanDefinition adapterDef = new RootBeanDefinition(JmsTargetAdapter.class);
if (StringUtils.hasText(jmsTemplate)) {
if (StringUtils.hasText(connectionFactory) || StringUtils.hasText(destination) || StringUtils.hasText(destinationName)) {
throw new BeanCreationException("when providing a 'jms-template' reference, none of " +
"'connection-factory', 'destination', or 'destination-name' should be provided.");
}
adapterDef.getPropertyValues().addPropertyValue(JMS_TEMPLATE_PROPERTY, new RuntimeBeanReference(jmsTemplate));
}
else if (StringUtils.hasText(connectionFactory) && (StringUtils.hasText(destination) ^ StringUtils.hasText(destinationName))) {
adapterDef.getPropertyValues().addPropertyValue(CONNECTION_FACTORY_PROPERTY, new RuntimeBeanReference(connectionFactory));
if (StringUtils.hasText(destination)) {
adapterDef.getPropertyValues().addPropertyValue(DESTINATION_PROPERTY, new RuntimeBeanReference(destination));
}
else {
adapterDef.getPropertyValues().addPropertyValue(DESTINATION_NAME_PROPERTY, destinationName);
}
}
else {
throw new BeanCreationException("either a 'jms-template' reference or both " +
"'connection-factory' and 'destination' (or 'destination-name') references must be provided.");
}
String channel = element.getAttribute(CHANNEL_ATTRIBUTE);
Subscription subscription = new Subscription(channel);
String adapterBeanName = parserContext.getReaderContext().generateBeanName(adapterDef);
parserContext.registerBeanComponent(new BeanComponentDefinition(adapterDef, adapterBeanName));
builder.addPropertyReference(HANDLER_PROPERTY, adapterBeanName);
builder.addPropertyValue(SUBSCRIPTION_PROPERTY, subscription);
}
}

103
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java Normal file
View File

@@ -0,0 +1,103 @@
/*
* 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.stream;
import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.springframework.integration.adapter.PollableSource;
import org.springframework.integration.message.MessageDeliveryException;
/**
* A pollable source for receiving bytes from an {@link InputStream}.
*
* @author Mark Fisher
*/
public class ByteStreamSource implements PollableSource<byte[]> {
private BufferedInputStream stream;
private Object streamMonitor;
private int bytesPerMessage = 1024;
private boolean shouldTruncate = true;
public ByteStreamSource(InputStream stream) {
this(stream, -1);
}
public ByteStreamSource(InputStream stream, int bufferSize) {
this.streamMonitor = stream;
if (stream instanceof BufferedInputStream) {
this.stream = (BufferedInputStream) stream;
}
else if (bufferSize > 0) {
this.stream = new BufferedInputStream(stream, bufferSize);
}
else {
this.stream = new BufferedInputStream(stream);
}
}
public void setBytesPerMessage(int bytesPerMessage) {
this.bytesPerMessage = bytesPerMessage;
}
public void setShouldTruncate(boolean shouldTruncate) {
this.shouldTruncate = shouldTruncate;
}
public Collection<byte[]> poll(int limit) {
List<byte[]> results = new ArrayList<byte[]>();
while (results.size() < limit) {
try {
byte[] bytes;
int bytesRead = 0;
synchronized (this.streamMonitor) {
if (stream.available() == 0) {
return results;
}
bytes = new byte[bytesPerMessage];
bytesRead = stream.read(bytes, 0, bytes.length);
}
if (bytesRead <= 0) {
return results;
}
if (!this.shouldTruncate) {
results.add(bytes);
}
else {
byte[] result = new byte[bytesRead];
System.arraycopy(bytes, 0, result, 0, result.length);
results.add(result);
}
}
catch (IOException e) {
throw new MessageDeliveryException("IO failure occurred in adapter", e);
}
}
return results;
}
}

43
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSourceAdapter.java Normal file
View File

@@ -0,0 +1,43 @@
/*
* 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.stream;
import java.io.InputStream;
import org.springframework.integration.adapter.PollingSourceAdapter;
/**
* A polling source adapter that wraps a {@link ByteStreamSource}.
*
* @author Mark Fisher
*/
public class ByteStreamSourceAdapter extends PollingSourceAdapter<byte[]> {
public ByteStreamSourceAdapter(InputStream stream) {
super(new ByteStreamSource(stream));
}
public void setBytesPerMessage(int bytesPerMessage) {
((ByteStreamSource) this.getSource()).setBytesPerMessage(bytesPerMessage);
}
public void setShouldTruncate(boolean shouldTruncate) {
((ByteStreamSource) this.getSource()).setShouldTruncate(shouldTruncate);
}
}

77
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapter.java Normal file
View File

@@ -0,0 +1,77 @@
/*
* 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.stream;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import org.springframework.integration.adapter.AbstractTargetAdapter;
import org.springframework.integration.message.MessageHandlingException;
/**
* A target adapter that writes a byte array to an {@link OutputStream}.
*
* @author Mark Fisher
*/
public class ByteStreamTargetAdapter extends AbstractTargetAdapter {
private BufferedOutputStream stream;
public ByteStreamTargetAdapter(OutputStream stream) {
this(stream, -1);
}
public ByteStreamTargetAdapter(OutputStream stream, int bufferSize) {
if (bufferSize > 0) {
this.stream = new BufferedOutputStream(stream, bufferSize);
}
else {
this.stream = new BufferedOutputStream(stream);
}
}
@Override
protected boolean sendToTarget(Object object) {
if (object == null) {
if (logger.isWarnEnabled()) {
logger.warn(this.getClass().getSimpleName() + " received null object");
}
return false;
}
try {
if (object instanceof String) {
this.stream.write(((String) object).getBytes());
}
else if (object instanceof byte[]){
this.stream.write((byte[]) object);
}
else {
throw new MessageHandlingException(this.getClass().getSimpleName() +
" only supports byte array and String-based messages");
}
this.stream.flush();
return true;
}
catch (IOException e) {
throw new MessageHandlingException("IO failure occurred in adapter", e);
}
}
}

81
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java Normal file
View File

@@ -0,0 +1,81 @@
/*
* 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.stream;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.springframework.integration.adapter.PollableSource;
import org.springframework.integration.message.MessageDeliveryException;
/**
* A pollable source for text-based {@link InputStream InputStreams}.
*
* @author Mark Fisher
*/
public class CharacterStreamSource implements PollableSource<String> {
private BufferedReader reader;
private Object streamMonitor;
public CharacterStreamSource(InputStream stream) {
this(stream, -1);
}
public CharacterStreamSource(InputStream stream, int bufferSize) {
this.streamMonitor = stream;
if (bufferSize > 0) {
this.reader = new BufferedReader(new InputStreamReader(stream), bufferSize);
}
else {
this.reader = new BufferedReader(new InputStreamReader(stream));
}
}
public Collection<String> poll(int limit) {
List<String> results = new ArrayList<String>();
while (results.size() < limit) {
try {
String line = null;
synchronized (this.streamMonitor) {
boolean isReady = reader.ready();
if (!isReady) {
return results;
}
line = reader.readLine();
}
if (line == null) {
return results;
}
results.add(line);
}
catch (IOException e) {
throw new MessageDeliveryException("IO failure occurred in adapter", e);
}
}
return results;
}
}

45
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapter.java Normal file
View File

@@ -0,0 +1,45 @@
/*
* 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.stream;
import java.io.InputStream;
import org.springframework.integration.adapter.PollingSourceAdapter;
import org.springframework.integration.channel.MessageChannel;
/**
* A polling source adapter that wraps a {@link CharacterStreamSource}.
*
* @author Mark Fisher
*/
public class CharacterStreamSourceAdapter extends PollingSourceAdapter<String> {
public CharacterStreamSourceAdapter(InputStream stream) {
super(new CharacterStreamSource(stream));
}
/**
* Factory method that creates an adapter for stdin (System.in).
*/
public static CharacterStreamSourceAdapter stdinAdapter(MessageChannel channel) {
CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(System.in);
adapter.setChannel(channel);
return adapter;
}
}

103
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapter.java Normal file
View File

@@ -0,0 +1,103 @@
/*
* 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.stream;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import org.springframework.integration.adapter.AbstractTargetAdapter;
import org.springframework.integration.message.MessageHandlingException;
/**
* A target adapter that writes to an {@link OutputStream}. String-based
* objects will be written directly, but if the object is not itself a
* {@link String}, the adapter will write the result of the object's
* {@link #toString()} method. To append a new-line after each write, set the
* {@link #shouldAppendNewLine} flag to <em>true</em>. It is <em>false</em>
* by default.
*
* @author Mark Fisher
*/
public class CharacterStreamTargetAdapter extends AbstractTargetAdapter {
private BufferedWriter writer;
private boolean shouldAppendNewLine = false;
public CharacterStreamTargetAdapter(OutputStream stream) {
this(stream, -1);
}
public CharacterStreamTargetAdapter(OutputStream stream, int bufferSize) {
if (bufferSize > 0) {
this.writer = new BufferedWriter(new OutputStreamWriter(stream), bufferSize);
}
else {
this.writer = new BufferedWriter(new OutputStreamWriter(stream));
}
}
/**
* Factory method that creates an adapter for stdout (System.out).
*/
public static CharacterStreamTargetAdapter stdoutAdapter() {
return new CharacterStreamTargetAdapter(System.out);
}
/**
* Factory method that creates an adapter for stderr (System.err).
*/
public static CharacterStreamTargetAdapter stderrAdapter() {
return new CharacterStreamTargetAdapter(System.err);
}
public void setShouldAppendNewLine(boolean shouldAppendNewLine) {
this.shouldAppendNewLine = shouldAppendNewLine;
}
@Override
protected boolean sendToTarget(Object object) {
if (object == null) {
if (logger.isWarnEnabled()) {
logger.warn("target adapter received null object");
}
return false;
}
try {
if (object instanceof String) {
writer.write((String) object);
}
else {
writer.write(object.toString());
}
if (this.shouldAppendNewLine) {
writer.newLine();
}
writer.flush();
return true;
}
catch (IOException e) {
throw new MessageHandlingException("IO failure occurred in adapter", e);
}
}
}