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:54:24 +00:00
parent 1f2bf5abc0
commit 9da59485cb
62 changed files with 61 additions and 4401 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
spring-integration-core/src/main/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapter.java
View File

@@ -1,63 +0,0 @@
/*
* 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.event;
import java.util.ArrayList;
import java.util.List;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.integration.adapter.AbstractSourceAdapter;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
/**
* A source adapter for passing Spring
* {@link ApplicationEvent ApplicationEvents} within messages.
*
* @author Mark Fisher
*/
public class ApplicationEventSourceAdapter extends AbstractSourceAdapter<ApplicationEvent> implements
ApplicationListener {
private List<Class<? extends ApplicationEvent>> eventTypes = new ArrayList<Class<? extends ApplicationEvent>>();
/**
* Set the list of event types (classes that extend ApplicationEvent) that
* this adapter should send to the message channel. By default, all event
* types will be sent.
*/
public void setEventTypes(List<Class<? extends ApplicationEvent>> eventTypes) {
Assert.notEmpty(eventTypes, "at least one event type is required");
this.eventTypes = eventTypes;
}
public void onApplicationEvent(ApplicationEvent event) {
if (CollectionUtils.isEmpty(this.eventTypes)) {
this.sendToChannel(event);
return;
}
for (Class<? extends ApplicationEvent> eventType : this.eventTypes) {
if (eventType.isAssignableFrom(event.getClass())) {
this.sendToChannel(event);
return;
}
}
}
}

51
spring-integration-core/src/main/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapter.java
View File

@@ -1,51 +0,0 @@
/*
* 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.event;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.integration.adapter.AbstractTargetAdapter;
/**
* A target adapter for publishing {@link MessagingEvent MessagingEvents}. The
* {@link MessagingEvent} is a subclass of Spring's {@link ApplicationEvent}
* used by this adapter to wrap any {@link Message} received on its channel.
*
* @author Mark Fisher
*/
public class ApplicationEventTargetAdapter extends AbstractTargetAdapter<MessagingEvent> implements
ApplicationEventPublisherAware {
private ApplicationEventPublisher applicationEventPublisher;
public ApplicationEventTargetAdapter() {
this.setMessageMapper(new MessagingEventMapper());
}
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
this.applicationEventPublisher = applicationEventPublisher;
}
@Override
protected boolean sendToTarget(MessagingEvent event) {
this.applicationEventPublisher.publishEvent(event);
return true;
}
}

37
spring-integration-core/src/main/java/org/springframework/integration/adapter/event/MessagingEvent.java
View File

@@ -1,37 +0,0 @@
/*
* 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.event;
import org.springframework.context.ApplicationEvent;
import org.springframework.integration.message.Message;
/**
* A subclass of {@link ApplicationEvent} that wraps a {@link Message}.
*
* @author Mark Fisher
*/
public class MessagingEvent<T> extends ApplicationEvent {
public MessagingEvent(Message<T> message) {
super(message);
}
public Message<T> getMessage() {
return (Message<T>) this.getSource();
}
}

38
spring-integration-core/src/main/java/org/springframework/integration/adapter/event/MessagingEventMapper.java
View File

@@ -1,38 +0,0 @@
/*
* 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.event;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageMapper;
/**
* A {@link MessageMapper} implementation for mapping to and from
* {@link MessagingEvent MessagingEvents}.
*
* @author Mark Fisher
*/
public class MessagingEventMapper<T> implements MessageMapper<T, MessagingEvent<T>> {
public MessagingEvent<T> fromMessage(Message<T> message) {
return new MessagingEvent<T>(message);
}
public Message<T> toMessage(MessagingEvent<T> event) {
return event.getMessage();
}
}

101
spring-integration-core/src/main/java/org/springframework/integration/adapter/file/AbstractFileMapper.java
View File

@@ -1,101 +0,0 @@
/*
* 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.file;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.message.AbstractMessageMapper;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.util.Assert;
import org.springframework.util.FileCopyUtils;
/**
* Base class providing common behavior for file-based message mappers.
*
* @author Mark Fisher
*/
public abstract class AbstractFileMapper<T> extends AbstractMessageMapper<T, File> {
protected Log logger = LogFactory.getLog(this.getClass());
private File parentDirectory;
private File backupDirectory;
private FileNameGenerator fileNameGenerator = new DefaultFileNameGenerator();
public AbstractFileMapper(File parentDirectory) {
this.parentDirectory = parentDirectory;
}
public void setBackupDirectory(File backupDirectory) {
this.backupDirectory = backupDirectory;
}
public void setFileNameGenerator(FileNameGenerator fileNameGenerator) {
Assert.notNull(fileNameGenerator, "'fileNameGenerator' must not be null");
this.fileNameGenerator = fileNameGenerator;
}
public File fromMessage(Message<T> message) {
try {
File file = new File(parentDirectory, this.fileNameGenerator.generateFileName(message));
this.writeToFile(file, message.getPayload());
return file;
}
catch (Exception e) {
throw new MessageHandlingException("failure occurred mapping file to message", e);
}
}
public Message<T> toMessage(File file) {
try {
T payload = this.readMessagePayload(file);
if (payload == null) {
return null;
}
Message<T> message = new GenericMessage<T>(this.getIdGenerator().generateId(), payload);
if (this.backupDirectory != null) {
FileWriter writer = new FileWriter(this.backupDirectory.getAbsolutePath() +
File.separator + file.getName());
FileCopyUtils.copy(new FileReader(file), writer);
}
file.delete();
return message;
}
catch (Exception e) {
String errorMessage = "failure occurred mapping file to message";
if (logger.isWarnEnabled()) {
logger.warn(errorMessage, e);
}
throw new MessageHandlingException(errorMessage, e);
}
}
protected abstract T readMessagePayload(File file) throws Exception;
protected abstract void writeToFile(File file, T payload) throws Exception;
}

46
spring-integration-core/src/main/java/org/springframework/integration/adapter/file/ByteArrayFileMapper.java
View File

@@ -1,46 +0,0 @@
/*
* 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.file;
import java.io.File;
import org.springframework.util.FileCopyUtils;
/**
* A {@link org.springframework.integration.message.MessageMapper}
* implementation for messages with a byte array payload.
*
* @author Mark Fisher
*/
public class ByteArrayFileMapper extends AbstractFileMapper<byte[]> {
public ByteArrayFileMapper(File parentDirectory) {
super(parentDirectory);
}
@Override
protected byte[] readMessagePayload(File file) throws Exception {
return FileCopyUtils.copyToByteArray(file);
}
@Override
protected void writeToFile(File file, byte[] payload) throws Exception {
FileCopyUtils.copy(payload, file);
}
}

36
spring-integration-core/src/main/java/org/springframework/integration/adapter/file/DefaultFileNameGenerator.java
View File

@@ -1,36 +0,0 @@
/*
* 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.file;
import org.springframework.integration.message.Message;
import org.springframework.util.StringUtils;
/**
* Default implementation of the filename generator strategy. Concatenates the
* message id and the current timestamp.
*
* @author Mark Fisher
*/
public class DefaultFileNameGenerator implements FileNameGenerator {
public String generateFileName(Message<?> message) {
String filenameProperty = message.getHeader().getProperty(FILENAME_PROPERTY_KEY);
return StringUtils.hasText(filenameProperty) ?
filenameProperty : message.getId() + "-" + System.currentTimeMillis() + ".msg";
}
}

33
spring-integration-core/src/main/java/org/springframework/integration/adapter/file/FileNameGenerator.java
View File

@@ -1,33 +0,0 @@
/*
* 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.file;
import org.springframework.integration.message.Message;
/**
* Strategy interface for generating a file name from a message.
*
* @author Mark Fisher
*/
public interface FileNameGenerator {
String FILENAME_PROPERTY_KEY = "filename";
String generateFileName(Message<?> message);
}

80
spring-integration-core/src/main/java/org/springframework/integration/adapter/file/FileSource.java
View File

@@ -1,80 +0,0 @@
/*
* 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.file;
import java.io.File;
import java.io.FileFilter;
import java.io.FilenameFilter;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.springframework.integration.adapter.PollableSource;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.util.Assert;
/**
* A messaging source that polls a directory to retrieve files.
*
* @author Mark Fisher
*/
public class FileSource implements PollableSource<File> {
private File directory;
private FileFilter fileFilter;
private FilenameFilter filenameFilter;
public FileSource(File directory) {
Assert.notNull("directory must not be null");
this.directory = directory;
}
public void setFileFilter(FileFilter fileFilter) {
this.fileFilter = fileFilter;
}
public void setFilenameFilter(FilenameFilter filenameFilter) {
this.filenameFilter = filenameFilter;
}
public Collection<File> poll(int limit) {
File[] files = null;
if (this.fileFilter != null) {
files = this.directory.listFiles(fileFilter);
}
else if (this.filenameFilter != null) {
files = this.directory.listFiles(filenameFilter);
}
else {
files = this.directory.listFiles();
}
if (files == null) {
throw new MessageHandlingException("Problem occurred while polling for files. " +
"Is '" + directory.getAbsolutePath() + "' a directory?");
}
int size = Math.min(limit, files.length);
List<File> results = new ArrayList<File>(size);
for (int i = 0; i < size; i++) {
results.add(files[i]);
}
return results;
}
}

65
spring-integration-core/src/main/java/org/springframework/integration/adapter/file/FileSourceAdapter.java
View File

@@ -1,65 +0,0 @@
/*
* 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.file;
import java.io.File;
import org.springframework.integration.adapter.PollingSourceAdapter;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.MessageMapper;
import org.springframework.util.Assert;
/**
* Channel adapter for polling a directory and creating messages from its files.
*
* @author Mark Fisher
*/
public class FileSourceAdapter extends PollingSourceAdapter<File> {
public FileSourceAdapter(File directory, MessageChannel channel, int period) {
this(directory, channel, period, true);
}
public FileSourceAdapter(File directory, MessageChannel channel, int period, boolean isTextBased) {
super(new FileSource(directory));
this.setChannel(channel);
this.setPeriod(period);
if (isTextBased) {
this.setMessageMapper(new TextFileMapper(directory));
}
else {
this.setMessageMapper(new ByteArrayFileMapper(directory));
}
}
public void setFileNameGenerator(FileNameGenerator fileNameGenerator) {
Assert.notNull(fileNameGenerator, "'fileNameGenerator' must not be null");
MessageMapper<?,?> mapper = this.getMessageMapper();
if (mapper instanceof AbstractFileMapper<?>) {
((AbstractFileMapper<?>) mapper).setFileNameGenerator(fileNameGenerator);
}
}
public void setBackupDirectory(File backupDirectory) {
Assert.notNull(backupDirectory, "'backupDirectory' must not be null");
MessageMapper<?, File> mapper = this.getMessageMapper();
if (mapper != null && (mapper instanceof AbstractFileMapper<?>)) {
((AbstractFileMapper<?>) mapper).setBackupDirectory(backupDirectory);
}
}
}

59
spring-integration-core/src/main/java/org/springframework/integration/adapter/file/FileTargetAdapter.java
View File

@@ -1,59 +0,0 @@
/*
* 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.file;
import java.io.File;
import org.springframework.integration.adapter.AbstractTargetAdapter;
import org.springframework.integration.message.MessageMapper;
import org.springframework.util.Assert;
/**
* A convenience adapter for writing files. The actual file writing occurs in
* the message mapper ({@link TextFileMapper} or {@link ByteArrayFileMapper}).
*
* @author Mark Fisher
*/
public class FileTargetAdapter extends AbstractTargetAdapter<File> {
public FileTargetAdapter(File directory) {
this(directory, true);
}
public FileTargetAdapter(File directory, boolean isTextBased) {
if (isTextBased) {
this.setMessageMapper(new TextFileMapper(directory));
}
else {
this.setMessageMapper(new ByteArrayFileMapper(directory));
}
}
public void setFileNameGenerator(FileNameGenerator fileNameGenerator) {
Assert.notNull(fileNameGenerator, "'fileNameGenerator' must not be null");
MessageMapper<?,?> mapper = this.getMessageMapper();
if (mapper instanceof AbstractFileMapper<?>) {
((AbstractFileMapper<?>) mapper).setFileNameGenerator(fileNameGenerator);
}
}
@Override
protected boolean sendToTarget(File file) {
return file.exists();
}
}

48
spring-integration-core/src/main/java/org/springframework/integration/adapter/file/TextFileMapper.java
View File

@@ -1,48 +0,0 @@
/*
* 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.file;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import org.springframework.util.FileCopyUtils;
/**
* A {@link org.springframework.integration.message.MessageMapper}
* implementation for messages with a String payload.
*
* @author Mark Fisher
*/
public class TextFileMapper extends AbstractFileMapper<String> {
public TextFileMapper(File parentDirectory) {
super(parentDirectory);
}
@Override
protected String readMessagePayload(File file) throws Exception {
return FileCopyUtils.copyToString(new FileReader(file));
}
@Override
protected void writeToFile(File file, String payload) throws Exception {
FileCopyUtils.copy(payload, new FileWriter(file));
}
}

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

@@ -1,53 +0,0 @@
/*
* 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.file.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.integration.adapter.file.FileSourceAdapter;
/**
* Parser for the &lt;file-source/&gt; element.
*
* @author Mark Fisher
*/
public class FileSourceAdapterParser extends AbstractSingleBeanDefinitionParser {
protected Class<?> getBeanClass(Element element) {
return FileSourceAdapter.class;
}
protected boolean shouldGenerateId() {
return false;
}
protected boolean shouldGenerateIdAsFallback() {
return true;
}
protected void doParse(Element element, BeanDefinitionBuilder builder) {
String directory = element.getAttribute("directory");
String channel = element.getAttribute("channel");
String pollPeriod = element.getAttribute("poll-period");
builder.addConstructorArg(directory);
builder.addConstructorArgReference(channel);
builder.addConstructorArg(pollPeriod);
}
}

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

@@ -1,60 +0,0 @@
/*
* 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.file.config;
import org.w3c.dom.Element;
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.file.FileTargetAdapter;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.scheduling.Subscription;
/**
* Parser for the &lt;file-target/&gt; element.
*
* @author Mark Fisher
*/
public class FileTargetAdapterParser extends AbstractSingleBeanDefinitionParser {
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) {
RootBeanDefinition adapterDef = new RootBeanDefinition(FileTargetAdapter.class);
adapterDef.getConstructorArgumentValues().addGenericArgumentValue(element.getAttribute("directory"));
String adapterBeanName = parserContext.getReaderContext().generateBeanName(adapterDef);
parserContext.registerBeanComponent(new BeanComponentDefinition(adapterDef, adapterBeanName));
builder.addPropertyReference("handler", adapterBeanName);
String channel = element.getAttribute("channel");
Subscription subscription = new Subscription(channel);
builder.addPropertyValue("subscription", subscription);
}
}

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

@@ -1,96 +0,0 @@
/*
* 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-core/src/main/java/org/springframework/integration/adapter/jms/DefaultJmsMessagePostProcessor.java
View File

@@ -1,48 +0,0 @@
/*
* 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-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java
View File

@@ -1,149 +0,0 @@
/*
* 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-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessagePostProcessor.java
View File

@@ -1,34 +0,0 @@
/*
* 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-core/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java
View File

@@ -1,116 +0,0 @@
/*
* 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-core/src/main/java/org/springframework/integration/adapter/jms/JmsPollingSourceAdapter.java
View File

@@ -1,44 +0,0 @@
/*
* 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-core/src/main/java/org/springframework/integration/adapter/jms/JmsTargetAdapter.java
View File

@@ -1,139 +0,0 @@
/*
* 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-core/src/main/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParser.java
View File

@@ -1,158 +0,0 @@
/*
* 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-core/src/main/java/org/springframework/integration/adapter/jms/config/JmsTargetAdapterParser.java
View File

@@ -1,109 +0,0 @@
/*
* 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-core/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java
View File

@@ -1,103 +0,0 @@
/*
* 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-core/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSourceAdapter.java
View File

@@ -1,43 +0,0 @@
/*
* 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-core/src/main/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapter.java
View File

@@ -1,77 +0,0 @@
/*
* 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-core/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java
View File

@@ -1,81 +0,0 @@
/*
* 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-core/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapter.java
View File

@@ -1,45 +0,0 @@
/*
* 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-core/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapter.java
View File

@@ -1,103 +0,0 @@
/*
* 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);
}
}
}

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

@@ -16,11 +16,19 @@
package org.springframework.integration.config;
import java.io.IOException;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.xml.BeanDefinitionParser;
import org.springframework.beans.factory.xml.NamespaceHandlerSupport;
import org.springframework.integration.adapter.file.config.FileSourceAdapterParser;
import org.springframework.integration.adapter.file.config.FileTargetAdapterParser;
import org.springframework.integration.adapter.jms.config.JmsSourceAdapterParser;
import org.springframework.integration.adapter.jms.config.JmsTargetAdapterParser;
import org.springframework.core.io.support.PropertiesLoaderUtils;
import org.springframework.util.ClassUtils;
/**
* Namespace handler for the integration namespace.
@@ -29,6 +37,12 @@ import org.springframework.integration.adapter.jms.config.JmsTargetAdapterParser
*/
public class IntegrationNamespaceHandler extends NamespaceHandlerSupport {
private static final String ADAPTER_PARSER_MAPPINGS_LOCATION = "META-INF/spring-integration.parsers";
private final Log logger = LogFactory.getLog(this.getClass());
public void init() {
registerBeanDefinitionParser("message-bus", new MessageBusParser());
registerBeanDefinitionParser("annotation-driven", new AnnotationDrivenParser());
@@ -38,10 +52,48 @@ public class IntegrationNamespaceHandler extends NamespaceHandlerSupport {
registerBeanDefinitionParser("endpoint", new EndpointParser());
registerBeanDefinitionParser("handler", new HandlerParser());
registerBeanDefinitionParser("handler-chain", new HandlerParser());
registerBeanDefinitionParser("file-source", new FileSourceAdapterParser());
registerBeanDefinitionParser("file-target", new FileTargetAdapterParser());
registerBeanDefinitionParser("jms-source", new JmsSourceAdapterParser());
registerBeanDefinitionParser("jms-target", new JmsTargetAdapterParser());
Map<String, Class<? extends BeanDefinitionParser>> parserMappings = this.loadAdapterParserMappings();
try {
for (Map.Entry<String, Class<? extends BeanDefinitionParser>> entry : parserMappings.entrySet()) {
registerBeanDefinitionParser(entry.getKey(), (BeanDefinitionParser) entry.getValue().newInstance());
}
}
catch (Exception e) {
throw new IllegalStateException("Failed to instantiate BeanDefinitionParser.", e);
}
}
private Map<String, Class<? extends BeanDefinitionParser>> loadAdapterParserMappings() {
Map<String, Class<? extends BeanDefinitionParser>> parserMappings =
new HashMap<String, Class<? extends BeanDefinitionParser>>();
ClassLoader classLoader = getClass().getClassLoader();
try {
Properties mappings =
PropertiesLoaderUtils.loadAllProperties(ADAPTER_PARSER_MAPPINGS_LOCATION, classLoader);
if (logger.isDebugEnabled()) {
logger.debug("Loaded parser mappings [" + mappings + "]");
}
Enumeration<?> propertyNames = mappings.propertyNames();
while (propertyNames.hasMoreElements()) {
String name = (String) propertyNames.nextElement();
String classname = mappings.getProperty(name);
Class<?> parserClass = ClassUtils.forName(classname, classLoader);
if (!BeanDefinitionParser.class.isAssignableFrom(parserClass)) {
throw new IllegalStateException("Expected class of type BeanDefinitionParser, but '" +
name + "' was of type '" + parserClass.getSimpleName() + "'");
}
parserMappings.put(name, (Class<? extends BeanDefinitionParser>) parserClass);
}
return parserMappings;
}
catch (IOException e) {
throw new IllegalStateException(
"Unable to load BeanDefinitionParser mappings from location [" +
ADAPTER_PARSER_MAPPINGS_LOCATION + "]. Root cause: " + e);
}
catch (ClassNotFoundException e) {
throw new IllegalStateException("Failed to load BeanDefinitionParser.", e);
}
}
}

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

@@ -13,7 +13,7 @@
<xsd:annotation>
<xsd:documentation><![CDATA[
Defines the configuration elements for Spring Integration.
Defines the core configuration elements for Spring Integration.
]]></xsd:documentation>
</xsd:annotation>
@@ -207,65 +207,4 @@
</xsd:complexType>
</xsd:element>
<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>

115
spring-integration-core/src/test/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapterTests.java
View File

@@ -1,115 +0,0 @@
/*
* 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.event;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import java.util.ArrayList;
import java.util.List;
import org.junit.Test;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.context.event.ContextStartedEvent;
import org.springframework.context.event.ContextStoppedEvent;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.message.Message;
/**
* @author Mark Fisher
*/
public class ApplicationEventSourceAdapterTests {
@Test
public void testAnyApplicationEventSentByDefault() {
MessageChannel channel = new SimpleChannel();
ApplicationEventSourceAdapter adapter = new ApplicationEventSourceAdapter();
adapter.setChannel(channel);
Message<?> message1 = channel.receive(0);
assertNull(message1);
adapter.onApplicationEvent(new TestApplicationEvent1());
adapter.onApplicationEvent(new TestApplicationEvent2());
Message<?> message2 = channel.receive(20);
assertNotNull(message2);
assertEquals("event1", ((ApplicationEvent) message2.getPayload()).getSource());
Message<?> message3 = channel.receive(20);
assertNotNull(message3);
assertEquals("event2", ((ApplicationEvent) message3.getPayload()).getSource());
}
@Test
public void testOnlyConfiguredEventTypesAreSent() {
MessageChannel channel = new SimpleChannel();
ApplicationEventSourceAdapter adapter = new ApplicationEventSourceAdapter();
List<Class<? extends ApplicationEvent>> eventTypes = new ArrayList<Class<? extends ApplicationEvent>>();
eventTypes.add(TestApplicationEvent1.class);
adapter.setEventTypes(eventTypes);
adapter.setChannel(channel);
Message<?> message1 = channel.receive(0);
assertNull(message1);
adapter.onApplicationEvent(new TestApplicationEvent1());
adapter.onApplicationEvent(new TestApplicationEvent2());
Message<?> message2 = channel.receive(20);
assertNotNull(message2);
assertEquals("event1", ((ApplicationEvent) message2.getPayload()).getSource());
Message<?> message3 = channel.receive(0);
assertNull(message3);
}
@Test
public void testApplicationContextEvents() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("applicationEventSourceAdapterTests.xml", this.getClass());
MessageChannel channel = (MessageChannel) context.getBean("channel");
Message<?> refreshedEventMessage = channel.receive(0);
assertNotNull(refreshedEventMessage);
assertEquals(ContextRefreshedEvent.class, refreshedEventMessage.getPayload().getClass());
context.start();
Message<?> startedEventMessage = channel.receive(0);
assertNotNull(startedEventMessage);
assertEquals(ContextStartedEvent.class, startedEventMessage.getPayload().getClass());
context.close();
Message<?> closedEventMessage = channel.receive(0);
assertNotNull(closedEventMessage);
assertEquals(ContextClosedEvent.class, closedEventMessage.getPayload().getClass());
Message<?> stoppedEventMessage = channel.receive(0);
assertNotNull(stoppedEventMessage);
assertEquals(ContextStoppedEvent.class, stoppedEventMessage.getPayload().getClass());
}
private static class TestApplicationEvent1 extends ApplicationEvent {
public TestApplicationEvent1() {
super("event1");
}
}
private static class TestApplicationEvent2 extends ApplicationEvent {
public TestApplicationEvent2() {
super("event2");
}
}
}

61
spring-integration-core/src/test/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapterTests.java
View File

@@ -1,61 +0,0 @@
/*
* 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.event;
import static org.junit.Assert.assertEquals;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.scheduling.Subscription;
/**
* @author Mark Fisher
*/
public class ApplicationEventTargetAdapterTests {
@Test
public void testSendingEvent() throws InterruptedException {
final CountDownLatch latch = new CountDownLatch(1);
ApplicationEventPublisher publisher = new ApplicationEventPublisher() {
public void publishEvent(ApplicationEvent event) {
latch.countDown();
}
};
MessageChannel channel = new SimpleChannel();
ApplicationEventTargetAdapter adapter = new ApplicationEventTargetAdapter();
adapter.setApplicationEventPublisher(publisher);
MessageBus bus = new MessageBus();
bus.registerChannel("channel", channel);
bus.registerHandler("adapter", adapter, new Subscription(channel));
bus.start();
assertEquals(1, latch.getCount());
channel.send(new StringMessage("123", "testing"));
latch.await(100, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
bus.stop();
}
}

15
spring-integration-core/src/test/java/org/springframework/integration/adapter/event/applicationEventSourceAdapterTests.xml
View File

@@ -1,15 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">
<bean id="bus" class="org.springframework.integration.bus.MessageBus"/>
<bean id="channel" class="org.springframework.integration.channel.SimpleChannel"/>
<bean id="adapter" class="org.springframework.integration.adapter.event.ApplicationEventSourceAdapter">
<property name="channel" ref="channel"/>
</bean>
</beans>

50
spring-integration-core/src/test/java/org/springframework/integration/adapter/file/DefaultFileNameGeneratorTests.java
View File

@@ -1,50 +0,0 @@
/*
* 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.file;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import org.junit.Test;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
/**
* @author Mark Fisher
*/
public class DefaultFileNameGeneratorTests {
@Test
public void testWithFileNamePropertyProvided() {
Message<String> message = new GenericMessage<String>("123", "testing");
message.getHeader().setProperty(FileNameGenerator.FILENAME_PROPERTY_KEY, "foo.bar");
FileNameGenerator generator = new DefaultFileNameGenerator();
String filename = generator.generateFileName(message);
assertEquals("foo.bar", filename);
}
@Test
public void testWithoutFileNamePropertyProvided() {
Message<String> message = new GenericMessage<String>("123", "testing");
FileNameGenerator generator = new DefaultFileNameGenerator();
String filename = generator.generateFileName(message);
assertTrue(filename.startsWith("123-"));
assertTrue(filename.endsWith(".msg"));
}
}

42
spring-integration-core/src/test/java/org/springframework/integration/adapter/file/config/FileTargetAdapterParserTests.java
View File

@@ -1,42 +0,0 @@
/*
* 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.file.config;
import static org.junit.Assert.assertEquals;
import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.adapter.file.FileTargetAdapter;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
/**
* @author Mark Fisher
*/
public class FileTargetAdapterParserTests {
@Test
public void testFileTargetAdapterParser() {
ApplicationContext context = new ClassPathXmlApplicationContext("fileTargetAdapterParserTests.xml", this.getClass());
DefaultMessageEndpoint endpoint = (DefaultMessageEndpoint) context.getBean("adapter");
assertEquals(FileTargetAdapter.class, endpoint.getHandler().getClass());
assertEquals("adapter", endpoint.getName());
assertEquals("testChannel", endpoint.getSubscription().getChannelName());
}
}

16
spring-integration-core/src/test/java/org/springframework/integration/adapter/file/config/fileTargetAdapterParserTests.xml
View File

@@ -1,16 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:si="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<si:message-bus/>
<si:channel id="testChannel"/>
<si:file-target id="adapter" directory="${java.io.tmpdir}" channel="testChannel"/>
</beans>

102
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/DefaultJmsMessagePostProcessorTests.java
View File

@@ -1,102 +0,0 @@
/*
* 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 static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import javax.jms.Destination;
import javax.jms.JMSException;
import org.junit.Test;
import org.springframework.integration.message.StringMessage;
/**
* @author Mark Fisher
*/
public class DefaultJmsMessagePostProcessorTests {
@Test
public void testJmsReplyTo() throws JMSException {
StringMessage message = new StringMessage("test1");
Destination replyTo = new Destination() {};
message.getHeader().setAttribute(JmsTargetAdapter.JMS_REPLY_TO, replyTo);
DefaultJmsMessagePostProcessor postProcessor = new DefaultJmsMessagePostProcessor();
javax.jms.Message jmsMessage = new StubTextMessage();
postProcessor.postProcessJmsMessage(jmsMessage, message.getHeader());
assertNotNull(jmsMessage.getJMSReplyTo());
assertSame(replyTo, jmsMessage.getJMSReplyTo());
}
@Test
public void testJmsReplyToIgnoredIfIncorrectType() throws JMSException {
StringMessage message = new StringMessage("test1");
message.getHeader().setAttribute(JmsTargetAdapter.JMS_REPLY_TO, "not-a-destination");
DefaultJmsMessagePostProcessor postProcessor = new DefaultJmsMessagePostProcessor();
javax.jms.Message jmsMessage = new StubTextMessage();
postProcessor.postProcessJmsMessage(jmsMessage, message.getHeader());
assertNull(jmsMessage.getJMSReplyTo());
}
@Test
public void testJmsCorrelationId() throws JMSException {
StringMessage message = new StringMessage("test1");
String jmsCorrelationId = "ABC-123";
message.getHeader().setAttribute(JmsTargetAdapter.JMS_CORRELATION_ID, jmsCorrelationId);
DefaultJmsMessagePostProcessor postProcessor = new DefaultJmsMessagePostProcessor();
javax.jms.Message jmsMessage = new StubTextMessage();
postProcessor.postProcessJmsMessage(jmsMessage, message.getHeader());
assertNotNull(jmsMessage.getJMSCorrelationID());
assertEquals(jmsCorrelationId, jmsMessage.getJMSCorrelationID());
}
@Test
public void testJmsCorrelationIdIgnoredIfIncorrectType() throws JMSException {
StringMessage message = new StringMessage("test1");
message.getHeader().setAttribute(JmsTargetAdapter.JMS_CORRELATION_ID, new Integer(123));
DefaultJmsMessagePostProcessor postProcessor = new DefaultJmsMessagePostProcessor();
javax.jms.Message jmsMessage = new StubTextMessage();
postProcessor.postProcessJmsMessage(jmsMessage, message.getHeader());
assertNull(jmsMessage.getJMSCorrelationID());
}
@Test
public void testJmsType() throws JMSException {
StringMessage message = new StringMessage("test1");
String jmsType = "testing";
message.getHeader().setAttribute(JmsTargetAdapter.JMS_TYPE, jmsType);
DefaultJmsMessagePostProcessor postProcessor = new DefaultJmsMessagePostProcessor();
javax.jms.Message jmsMessage = new StubTextMessage();
postProcessor.postProcessJmsMessage(jmsMessage, message.getHeader());
assertNotNull(jmsMessage.getJMSType());
assertEquals(jmsType, jmsMessage.getJMSType());
}
@Test
public void testJmsTypeIgnoredIfIncorrectType() throws JMSException {
StringMessage message = new StringMessage("test1");
message.getHeader().setAttribute(JmsTargetAdapter.JMS_TYPE, new Integer(123));
DefaultJmsMessagePostProcessor postProcessor = new DefaultJmsMessagePostProcessor();
javax.jms.Message jmsMessage = new StubTextMessage();
postProcessor.postProcessJmsMessage(jmsMessage, message.getHeader());
assertNull(jmsMessage.getJMSType());
}
}

83
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/StubConnection.java
View File

@@ -1,83 +0,0 @@
/*
* 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.Connection;
import javax.jms.ConnectionConsumer;
import javax.jms.ConnectionMetaData;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.ServerSessionPool;
import javax.jms.Session;
import javax.jms.Topic;
/**
* @author Mark Fisher
*/
public class StubConnection implements Connection {
private String messageText;
public StubConnection(String messageText) {
this.messageText = messageText;
}
public void close() throws JMSException {
}
public ConnectionConsumer createConnectionConsumer(Destination destination, String messageSelector,
ServerSessionPool sessionPool, int maxMessages) throws JMSException {
return null;
}
public ConnectionConsumer createDurableConnectionConsumer(Topic topic, String subscriptionName,
String messageSelector, ServerSessionPool sessionPool, int maxMessages) throws JMSException {
return null;
}
public Session createSession(boolean transacted, int acknowledgeMode) throws JMSException {
return new StubSession(this.messageText);
}
public String getClientID() throws JMSException {
return null;
}
public ExceptionListener getExceptionListener() throws JMSException {
return null;
}
public ConnectionMetaData getMetaData() throws JMSException {
return null;
}
public void setClientID(String clientID) throws JMSException {
}
public void setExceptionListener(ExceptionListener listener) throws JMSException {
}
public void start() throws JMSException {
}
public void stop() throws JMSException {
}
}

65
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/StubConsumer.java
View File

@@ -1,65 +0,0 @@
/*
* 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 javax.jms.MessageConsumer;
import javax.jms.MessageListener;
/**
* @author Mark Fisher
*/
public class StubConsumer implements MessageConsumer {
private String messageText;
public StubConsumer(String messageText) {
this.messageText = messageText;
}
public void close() throws JMSException {
}
public MessageListener getMessageListener() throws JMSException {
return null;
}
public String getMessageSelector() throws JMSException {
return null;
}
public Message receive() throws JMSException {
StubTextMessage message = new StubTextMessage();
message.setText(this.messageText);
return message;
}
public Message receive(long timeout) throws JMSException {
return this.receive();
}
public Message receiveNoWait() throws JMSException {
return this.receive();
}
public void setMessageListener(MessageListener listener) throws JMSException {
}
}

26
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/StubDestination.java
View File

@@ -1,26 +0,0 @@
/*
* 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;
/**
* @author Mark Fisher
*/
public class StubDestination implements Destination {
}

168
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/StubSession.java
View File

@@ -1,168 +0,0 @@
/*
* 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.io.Serializable;
import javax.jms.BytesMessage;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Queue;
import javax.jms.QueueBrowser;
import javax.jms.Session;
import javax.jms.StreamMessage;
import javax.jms.TemporaryQueue;
import javax.jms.TemporaryTopic;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicSubscriber;
/**
* @author Mark Fisher
*/
public class StubSession implements Session {
private String messageText;
public StubSession(String messageText) {
this.messageText = messageText;
}
public void close() throws JMSException {
}
public void commit() throws JMSException {
}
public QueueBrowser createBrowser(Queue queue) throws JMSException {
return null;
}
public QueueBrowser createBrowser(Queue queue, String messageSelector) throws JMSException {
return null;
}
public BytesMessage createBytesMessage() throws JMSException {
return null;
}
public MessageConsumer createConsumer(Destination destination) throws JMSException {
return new StubConsumer(this.messageText);
}
public MessageConsumer createConsumer(Destination destination, String messageSelector) throws JMSException {
return new StubConsumer(this.messageText);
}
public MessageConsumer createConsumer(Destination destination, String messageSelector, boolean NoLocal)
throws JMSException {
return new StubConsumer(this.messageText);
}
public TopicSubscriber createDurableSubscriber(Topic topic, String name) throws JMSException {
return null;
}
public TopicSubscriber createDurableSubscriber(Topic topic, String name, String messageSelector, boolean noLocal)
throws JMSException {
return null;
}
public MapMessage createMapMessage() throws JMSException {
return null;
}
public Message createMessage() throws JMSException {
return null;
}
public ObjectMessage createObjectMessage() throws JMSException {
return null;
}
public ObjectMessage createObjectMessage(Serializable object) throws JMSException {
return null;
}
public MessageProducer createProducer(Destination destination) throws JMSException {
return null;
}
public Queue createQueue(String queueName) throws JMSException {
return null;
}
public StreamMessage createStreamMessage() throws JMSException {
return null;
}
public TemporaryQueue createTemporaryQueue() throws JMSException {
return null;
}
public TemporaryTopic createTemporaryTopic() throws JMSException {
return null;
}
public TextMessage createTextMessage() throws JMSException {
return null;
}
public TextMessage createTextMessage(String text) throws JMSException {
return null;
}
public Topic createTopic(String topicName) throws JMSException {
return null;
}
public int getAcknowledgeMode() throws JMSException {
return 0;
}
public MessageListener getMessageListener() throws JMSException {
return null;
}
public boolean getTransacted() throws JMSException {
return false;
}
public void recover() throws JMSException {
}
public void rollback() throws JMSException {
}
public void run() {
}
public void setMessageListener(MessageListener listener) throws JMSException {
}
public void unsubscribe(String name) throws JMSException {
}
}

188
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/StubTextMessage.java
View File

@@ -1,188 +0,0 @@
package org.springframework.integration.adapter.jms;
import java.util.Enumeration;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.TextMessage;
public class StubTextMessage implements TextMessage {
private String text;
private Destination replyTo;
private String correlationID;
private String type;
public String getText() throws JMSException {
return this.text;
}
public void setText(String text) throws JMSException {
this.text = text;
}
public void acknowledge() throws JMSException {
}
public void clearBody() throws JMSException {
}
public void clearProperties() throws JMSException {
}
public boolean getBooleanProperty(String name) throws JMSException {
return false;
}
public byte getByteProperty(String name) throws JMSException {
return 0;
}
public double getDoubleProperty(String name) throws JMSException {
return 0;
}
public float getFloatProperty(String name) throws JMSException {
return 0;
}
public int getIntProperty(String name) throws JMSException {
return 0;
}
public String getJMSCorrelationID() throws JMSException {
return this.correlationID;
}
public byte[] getJMSCorrelationIDAsBytes() throws JMSException {
return null;
}
public int getJMSDeliveryMode() throws JMSException {
return 0;
}
public Destination getJMSDestination() throws JMSException {
return null;
}
public long getJMSExpiration() throws JMSException {
return 0;
}
public String getJMSMessageID() throws JMSException {
return null;
}
public int getJMSPriority() throws JMSException {
return 0;
}
public boolean getJMSRedelivered() throws JMSException {
return false;
}
public Destination getJMSReplyTo() throws JMSException {
return this.replyTo;
}
public long getJMSTimestamp() throws JMSException {
return 0;
}
public String getJMSType() throws JMSException {
return this.type;
}
public long getLongProperty(String name) throws JMSException {
return 0;
}
public Object getObjectProperty(String name) throws JMSException {
return null;
}
public Enumeration getPropertyNames() throws JMSException {
return null;
}
public short getShortProperty(String name) throws JMSException {
return 0;
}
public String getStringProperty(String name) throws JMSException {
return null;
}
public boolean propertyExists(String name) throws JMSException {
return false;
}
public void setBooleanProperty(String name, boolean value) throws JMSException {
}
public void setByteProperty(String name, byte value) throws JMSException {
}
public void setDoubleProperty(String name, double value) throws JMSException {
}
public void setFloatProperty(String name, float value) throws JMSException {
}
public void setIntProperty(String name, int value) throws JMSException {
}
public void setJMSCorrelationID(String correlationID) throws JMSException {
this.correlationID = correlationID;
}
public void setJMSCorrelationIDAsBytes(byte[] correlationID) throws JMSException {
}
public void setJMSDeliveryMode(int deliveryMode) throws JMSException {
}
public void setJMSDestination(Destination destination) throws JMSException {
}
public void setJMSExpiration(long expiration) throws JMSException {
}
public void setJMSMessageID(String id) throws JMSException {
}
public void setJMSPriority(int priority) throws JMSException {
}
public void setJMSRedelivered(boolean redelivered) throws JMSException {
}
public void setJMSReplyTo(Destination replyTo) throws JMSException {
this.replyTo = replyTo;
}
public void setJMSTimestamp(long timestamp) throws JMSException {
}
public void setJMSType(String type) throws JMSException {
this.type = type;
}
public void setLongProperty(String name, long value) throws JMSException {
}
public void setObjectProperty(String name, Object value) throws JMSException {
}
public void setShortProperty(String name, short value) throws JMSException {
}
public void setStringProperty(String name, String value) throws JMSException {
}
}

196
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParserTests.java
View File

@@ -1,196 +0,0 @@
/*
* 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 static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import javax.jms.JMSException;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.junit.Test;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.adapter.jms.JmsMessageDrivenSourceAdapter;
import org.springframework.integration.adapter.jms.JmsPollingSourceAdapter;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
import org.springframework.jms.support.converter.MessageConversionException;
import org.springframework.jms.support.converter.MessageConverter;
/**
* @author Mark Fisher
*/
public class JmsSourceAdapterParserTests {
@Test
public void testPollingAdapterWithJmsTemplate() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"pollingAdapterWithJmsTemplate.xml", this.getClass());
context.start();
JmsPollingSourceAdapter adapter = (JmsPollingSourceAdapter) context.getBean("adapter");
adapter.processMessages();
MessageChannel channel = (MessageChannel) context.getBean("channel");
Message<?> message = channel.receive(500);
assertNotNull("message should not be null", message);
assertEquals("polling-test", message.getPayload());
context.stop();
}
@Test
public void testPollingAdapterWithConnectionFactoryAndDestination() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"pollingAdapterWithConnectionFactoryAndDestination.xml", this.getClass());
context.start();
JmsPollingSourceAdapter adapter = (JmsPollingSourceAdapter) context.getBean("adapter");
adapter.processMessages();
MessageChannel channel = (MessageChannel) context.getBean("channel");
Message<?> message = channel.receive(500);
assertNotNull("message should not be null", message);
assertEquals("polling-test", message.getPayload());
context.stop();
}
@Test
public void testPollingAdapterWithConnectionFactoryAndDestinationName() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"pollingAdapterWithConnectionFactoryAndDestinationName.xml", this.getClass());
context.start();
JmsPollingSourceAdapter adapter = (JmsPollingSourceAdapter) context.getBean("adapter");
adapter.processMessages();
MessageChannel channel = (MessageChannel) context.getBean("channel");
Message<?> message = channel.receive(500);
assertNotNull("message should not be null", message);
assertEquals("polling-test", message.getPayload());
context.stop();
}
@Test
public void testMessageDrivenAdapterWithConnectionFactoryAndDestination() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"messageDrivenAdapterWithConnectionFactoryAndDestination.xml", this.getClass());
context.start();
JmsMessageDrivenSourceAdapter adapter = (JmsMessageDrivenSourceAdapter) context.getBean("adapter");
assertEquals(JmsMessageDrivenSourceAdapter.class, adapter.getClass());
MessageChannel channel = (MessageChannel) context.getBean("channel");
Message<?> message = channel.receive(3000);
assertNotNull("message should not be null", message);
assertEquals("message-driven-test", message.getPayload());
context.stop();
}
@Test
public void testMessageDrivenAdapterWithConnectionFactoryAndDestinationName() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"messageDrivenAdapterWithConnectionFactoryAndDestinationName.xml", this.getClass());
context.start();
JmsMessageDrivenSourceAdapter adapter = (JmsMessageDrivenSourceAdapter) context.getBean("adapter");
assertEquals(JmsMessageDrivenSourceAdapter.class, adapter.getClass());
MessageChannel channel = (MessageChannel) context.getBean("channel");
Message<?> message = channel.receive(3000);
assertNotNull("message should not be null", message);
assertEquals("message-driven-test", message.getPayload());
context.stop();
}
@Test
public void testMessageDrivenAdapterWithMessageConverter() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"messageDrivenAdapterWithMessageConverter.xml", this.getClass());
JmsMessageDrivenSourceAdapter adapter = (JmsMessageDrivenSourceAdapter) context.getBean("adapter");
assertEquals(JmsMessageDrivenSourceAdapter.class, adapter.getClass());
MessageChannel channel = (MessageChannel) context.getBean("channel");
Message<?> message = channel.receive(3000);
assertNotNull("message should not be null", message);
assertEquals("converted-test-message", message.getPayload());
context.stop();
}
@Test(expected=BeanDefinitionStoreException.class)
public void testPollingAdapterWithConnectionFactoryOnly() {
try {
new ClassPathXmlApplicationContext("pollingAdapterWithConnectionFactoryOnly.xml", this.getClass());
}
catch (RuntimeException e) {
assertEquals(BeanCreationException.class, e.getCause().getClass());
throw e;
}
}
@Test(expected=BeanDefinitionStoreException.class)
public void testPollingAdapterWithDestinationOnly() {
try {
new ClassPathXmlApplicationContext("pollingAdapterWithDestinationOnly.xml", this.getClass());
}
catch (RuntimeException e) {
assertEquals(BeanCreationException.class, e.getCause().getClass());
throw e;
}
}
@Test(expected=BeanDefinitionStoreException.class)
public void testPollingAdapterWithDestinationNameOnly() {
try {
new ClassPathXmlApplicationContext("pollingAdapterWithDestinationNameOnly.xml", this.getClass());
}
catch (RuntimeException e) {
assertEquals(BeanCreationException.class, e.getCause().getClass());
throw e;
}
}
@Test(expected=BeanDefinitionStoreException.class)
public void testPollingAdapterWithoutPollPeriod() {
try {
new ClassPathXmlApplicationContext("pollingAdapterWithoutPollPeriod.xml", this.getClass());
}
catch (RuntimeException e) {
assertEquals(BeanCreationException.class, e.getCause().getClass());
throw e;
}
}
@Test(expected=BeanDefinitionStoreException.class)
public void testMessageDrivenAdapterWithConnectionFactoryOnly() {
try {
new ClassPathXmlApplicationContext("messageDrivenAdapterWithConnectionFactoryOnly.xml", this.getClass());
}
catch (RuntimeException e) {
assertEquals(BeanCreationException.class, e.getCause().getClass());
throw e;
}
}
public static class TestMessageConverter implements MessageConverter {
public Object fromMessage(javax.jms.Message message) throws JMSException, MessageConversionException {
String original = ((TextMessage) message).getText();
return "converted-" + original;
}
public javax.jms.Message toMessage(Object object, Session session) throws JMSException,
MessageConversionException {
return null;
}
}
}

52
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/JmsTargetAdapterParserTests.java
View File

@@ -1,52 +0,0 @@
/*
* 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 static org.junit.Assert.assertEquals;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.adapter.jms.JmsTargetAdapter;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
/**
* @author Mark Fisher
*/
public class JmsTargetAdapterParserTests {
@Test
public void testTargetAdapterWithConnectionFactoryAndDestination() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"targetAdapterWithConnectionFactoryAndDestination.xml", this.getClass());
DefaultMessageEndpoint endpoint = (DefaultMessageEndpoint) context.getBean("adapter");
assertEquals(JmsTargetAdapter.class, endpoint.getHandler().getClass());
assertEquals("adapter", endpoint.getName());
assertEquals("testChannel", endpoint.getSubscription().getChannelName());
}
@Test
public void testTargetAdapterWithConnectionFactoryAndDestinationName() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"targetAdapterWithConnectionFactoryAndDestinationName.xml", this.getClass());
DefaultMessageEndpoint endpoint = (DefaultMessageEndpoint) context.getBean("adapter");
assertEquals(JmsTargetAdapter.class, endpoint.getHandler().getClass());
assertEquals("adapter", endpoint.getName());
assertEquals("testChannel", endpoint.getSubscription().getChannelName());
}
}

29
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithConnectionFactoryAndDestination.xml
View File

@@ -1,29 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:si="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<si:message-bus/>
<si:channel id="channel"/>
<si:jms-source id="adapter"
connection-factory="connectionFactory"
destination="testDestination"
channel="channel"/>
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<constructor-arg>
<bean class="org.springframework.integration.adapter.jms.StubConnection">
<constructor-arg value="message-driven-test"/>
</bean>
</constructor-arg>
</bean>
<bean id="testDestination" class="org.springframework.integration.adapter.jms.StubDestination"/>
</beans>

27
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithConnectionFactoryAndDestinationName.xml
View File

@@ -1,27 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:si="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<si:message-bus/>
<si:channel id="channel"/>
<si:jms-source id="adapter"
connection-factory="connectionFactory"
destination-name="testDestinationName"
channel="channel"/>
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<constructor-arg>
<bean class="org.springframework.integration.adapter.jms.StubConnection">
<constructor-arg value="message-driven-test"/>
</bean>
</constructor-arg>
</bean>
</beans>

24
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithConnectionFactoryOnly.xml
View File

@@ -1,24 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:si="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<si:message-bus/>
<si:channel id="channel"/>
<si:jms-source id="adapter" connection-factory="connectionFactory" channel="channel"/>
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<constructor-arg>
<bean class="org.springframework.integration.adapter.jms.StubConnection">
<constructor-arg value="message-driven-test"/>
</bean>
</constructor-arg>
</bean>
</beans>

32
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/messageDrivenAdapterWithMessageConverter.xml
View File

@@ -1,32 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:si="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<si:message-bus/>
<si:channel id="channel"/>
<si:jms-source id="adapter"
connection-factory="connectionFactory"
destination="testDestination"
channel="channel"
message-converter="converter"/>
<bean id="converter" class="org.springframework.integration.adapter.jms.config.JmsSourceAdapterParserTests$TestMessageConverter"/>
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<constructor-arg>
<bean class="org.springframework.integration.adapter.jms.StubConnection">
<constructor-arg value="test-message"/>
</bean>
</constructor-arg>
</bean>
<bean id="testDestination" class="org.springframework.integration.adapter.jms.StubDestination"/>
</beans>

30
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithConnectionFactoryAndDestination.xml
View File

@@ -1,30 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:si="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<si:message-bus/>
<si:channel id="channel"/>
<si:jms-source id="adapter"
connection-factory="connectionFactory"
destination="testDestination"
channel="channel"
poll-period="5000"/>
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<constructor-arg>
<bean class="org.springframework.integration.adapter.jms.StubConnection">
<constructor-arg value="polling-test"/>
</bean>
</constructor-arg>
</bean>
<bean id="testDestination" class="org.springframework.integration.adapter.jms.StubDestination"/>
</beans>

28
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithConnectionFactoryAndDestinationName.xml
View File

@@ -1,28 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:si="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<si:message-bus/>
<si:channel id="channel"/>
<si:jms-source id="adapter"
connection-factory="connectionFactory"
destination-name="testDestinationName"
channel="channel"
poll-period="5000"/>
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<constructor-arg>
<bean class="org.springframework.integration.adapter.jms.StubConnection">
<constructor-arg value="polling-test"/>
</bean>
</constructor-arg>
</bean>
</beans>

24
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithConnectionFactoryOnly.xml
View File

@@ -1,24 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:si="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<si:message-bus/>
<si:channel id="channel"/>
<si:jms-source id="adapter" connection-factory="connectionFactory" channel="channel" poll-period="5000"/>
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<constructor-arg>
<bean class="org.springframework.integration.adapter.jms.StubConnection">
<constructor-arg value="polling-test"/>
</bean>
</constructor-arg>
</bean>
</beans>

16
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationNameOnly.xml
View File

@@ -1,16 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:si="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<si:message-bus/>
<si:channel id="channel"/>
<si:jms-source id="adapter" destination-name="testDestinationName" channel="channel" poll-period="5000"/>
</beans>

18
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithDestinationOnly.xml
View File

@@ -1,18 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:si="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<si:message-bus/>
<si:channel id="channel"/>
<si:jms-source id="adapter" destination="testDestination" channel="channel" poll-period="5000"/>
<bean id="testDestination" class="org.springframework.integration.adapter.jms.StubDestination"/>
</beans>

29
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithJmsTemplate.xml
View File

@@ -1,29 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:si="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<si:message-bus/>
<si:channel id="channel"/>
<si:jms-source id="adapter" jms-template="jmsTemplate" channel="channel" poll-period="5000"/>
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="defaultDestinationName" value="test"/>
</bean>
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<constructor-arg>
<bean class="org.springframework.integration.adapter.jms.StubConnection">
<constructor-arg value="polling-test"/>
</bean>
</constructor-arg>
</bean>
</beans>

29
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/pollingAdapterWithoutPollPeriod.xml
View File

@@ -1,29 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:si="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<si:message-bus/>
<si:channel id="channel"/>
<si:jms-source id="adapter" jms-template="jmsTemplate" channel="channel"/>
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="defaultDestinationName" value="test"/>
</bean>
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<constructor-arg>
<bean class="org.springframework.integration.adapter.jms.StubConnection">
<constructor-arg value="polling-test"/>
</bean>
</constructor-arg>
</bean>
</beans>

29
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/targetAdapterWithConnectionFactoryAndDestination.xml
View File

@@ -1,29 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:si="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<si:message-bus/>
<si:channel id="testChannel"/>
<si:jms-target id="adapter"
connection-factory="connectionFactory"
destination="testDestination"
channel="testChannel"/>
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<constructor-arg>
<bean class="org.springframework.integration.adapter.jms.StubConnection">
<constructor-arg value="target-test"/>
</bean>
</constructor-arg>
</bean>
<bean id="testDestination" class="org.springframework.integration.adapter.jms.StubDestination"/>
</beans>

27
spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/targetAdapterWithConnectionFactoryAndDestinationName.xml
View File

@@ -1,27 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:si="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<si:message-bus/>
<si:channel id="testChannel"/>
<si:jms-target id="adapter"
connection-factory="connectionFactory"
destination-name="queue.test"
channel="testChannel"/>
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<constructor-arg>
<bean class="org.springframework.integration.adapter.jms.StubConnection">
<constructor-arg value="target-test"/>
</bean>
</constructor-arg>
</bean>
</beans>

173
spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceAdapterTests.java
View File

@@ -1,173 +0,0 @@
/*
* 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 static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import java.io.ByteArrayInputStream;
import org.junit.Test;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.message.Message;
/**
* @author Mark Fisher
*/
public class ByteStreamSourceAdapterTests {
@Test
public void testEndOfStream() {
byte[] bytes = new byte[] {1,2,3};
ByteArrayInputStream stream = new ByteArrayInputStream(bytes);
MessageChannel channel = new SimpleChannel();
ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream);
adapter.setChannel(channel);
adapter.start();
int count = adapter.processMessages();
assertEquals(1, count);
Message<?> message1 = channel.receive(500);
byte[] payload = (byte[]) message1.getPayload();
assertEquals(3, payload.length);
assertEquals(1, payload[0]);
assertEquals(2, payload[1]);
assertEquals(3, payload[2]);
Message<?> message2 = channel.receive(0);
assertNull(message2);
adapter.processMessages();
Message<?> message3 = channel.receive(0);
assertNull(message3);
}
@Test
public void testEndOfStreamWithMaxMessagesPerTask() throws Exception {
byte[] bytes = new byte[] {0,1,2,3,4,5,6,7};
ByteArrayInputStream stream = new ByteArrayInputStream(bytes);
MessageChannel channel = new SimpleChannel();
ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream);
adapter.setChannel(channel);
adapter.setBytesPerMessage(8);
adapter.setMaxMessagesPerTask(5);
adapter.start();
int count = adapter.processMessages();
assertEquals(1, count);
Message<?> message1 = channel.receive(500);
assertEquals(8, ((byte[]) message1.getPayload()).length);
Message<?> message2 = channel.receive(0);
assertNull(message2);
}
@Test
public void testMultipleMessagesWithSingleMessagePerTask() {
byte[] bytes = new byte[] {0,1,2,3,4,5,6,7};
ByteArrayInputStream stream = new ByteArrayInputStream(bytes);
MessageChannel channel = new SimpleChannel();
ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream);
adapter.setBytesPerMessage(4);
adapter.setInitialDelay(10000);
adapter.setMaxMessagesPerTask(1);
adapter.setChannel(channel);
adapter.start();
int count = adapter.processMessages();
assertEquals(1, count);
Message<?> message1 = channel.receive(0);
byte[] bytes1 = (byte[]) message1.getPayload();
assertEquals(4, bytes1.length);
assertEquals(0, bytes1[0]);
Message<?> message2 = channel.receive(0);
assertNull(message2);
adapter.processMessages();
Message<?> message3 = channel.receive(0);
byte[] bytes3 = (byte[]) message3.getPayload();
assertEquals(4, bytes3.length);
assertEquals(4, bytes3[0]);
}
@Test
public void testLessThanMaxMessagesAvailable() {
byte[] bytes = new byte[] {0,1,2,3,4,5,6,7};
ByteArrayInputStream stream = new ByteArrayInputStream(bytes);
MessageChannel channel = new SimpleChannel();
ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream);
adapter.setInitialDelay(10000);
adapter.setChannel(channel);
adapter.setBytesPerMessage(4);
adapter.setMaxMessagesPerTask(5);
adapter.start();
int count = adapter.processMessages();
assertEquals(2, count);
Message<?> message1 = channel.receive(0);
byte[] bytes1 = (byte[]) message1.getPayload();
assertEquals(4, bytes1.length);
assertEquals(0, bytes1[0]);
Message<?> message2 = channel.receive(0);
byte[] bytes2 = (byte[]) message2.getPayload();
assertEquals(4, bytes2.length);
assertEquals(4, bytes2[0]);
Message<?> message3 = channel.receive(0);
assertNull(message3);
}
@Test
public void testByteArrayIsTruncated() {
byte[] bytes = new byte[] {0,1,2,3,4,5};
ByteArrayInputStream stream = new ByteArrayInputStream(bytes);
MessageChannel channel = new SimpleChannel();
ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream);
adapter.setInitialDelay(10000);
adapter.setBytesPerMessage(4);
adapter.setMaxMessagesPerTask(1);
adapter.setChannel(channel);
adapter.start();
int count = adapter.processMessages();
assertEquals(1, count);
Message<?> message1 = channel.receive(0);
assertEquals(4, ((byte[]) message1.getPayload()).length);
Message<?> message2 = channel.receive(0);
assertNull(message2);
adapter.processMessages();
Message<?> message3 = channel.receive(0);
assertEquals(2, ((byte[]) message3.getPayload()).length);
}
@Test
public void testByteArrayIsNotTruncated() {
byte[] bytes = new byte[] {0,1,2,3,4,5};
ByteArrayInputStream stream = new ByteArrayInputStream(bytes);
MessageChannel channel = new SimpleChannel();
ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream);
adapter.setInitialDelay(10000);
adapter.setBytesPerMessage(4);
adapter.setShouldTruncate(false);
adapter.setMaxMessagesPerTask(1);
adapter.setChannel(channel);
adapter.start();
int count = adapter.processMessages();
assertEquals(1, count);
Message<?> message1 = channel.receive(0);
assertEquals(4, ((byte[]) message1.getPayload()).length);
Message<?> message2 = channel.receive(0);
assertNull(message2);
adapter.processMessages();
Message<?> message3 = channel.receive(0);
assertEquals(4, ((byte[]) message3.getPayload()).length);
assertEquals(0, ((byte[]) message3.getPayload())[3]);
}
}

220
spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java
View File

@@ -1,220 +0,0 @@
/*
* 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 static org.junit.Assert.assertEquals;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import org.junit.Test;
import org.springframework.integration.channel.DispatcherPolicy;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.dispatcher.DefaultMessageDispatcher;
import org.springframework.integration.dispatcher.MessageDispatcher;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler;
/**
* @author Mark Fisher
*/
public class ByteStreamTargetAdapterTests {
@Test
public void testSingleByteArray() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream);
adapter.handle(new GenericMessage<byte[]>(new byte[] {1,2,3}));
byte[] result = stream.toByteArray();
assertEquals(3, result.length);
assertEquals(1, result[0]);
assertEquals(2, result[1]);
assertEquals(3, result[2]);
}
@Test
public void testSingleString() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream);
adapter.handle(new StringMessage("foo"));
byte[] result = stream.toByteArray();
assertEquals(3, result.length);
assertEquals("foo", new String(result));
}
@Test
public void testMaxMessagesPerTaskSameAsMessageCount() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream);
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
dispatcherPolicy.setMaxMessagesPerTask(3);
SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1);
MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(adapter);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
assertEquals(3, dispatcher.dispatch());
byte[] result = stream.toByteArray();
assertEquals(9, result.length);
assertEquals(1, result[0]);
assertEquals(9, result[8]);
}
@Test
public void testMaxMessagesPerTaskLessThanMessageCount() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream);
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
dispatcherPolicy.setMaxMessagesPerTask(2);
SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1);
MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(adapter);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
assertEquals(2, dispatcher.dispatch());
byte[] result = stream.toByteArray();
assertEquals(6, result.length);
assertEquals(1, result[0]);
}
@Test
public void testMaxMessagesPerTaskExceedsMessageCount() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream);
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
dispatcherPolicy.setMaxMessagesPerTask(5);
dispatcherPolicy.setReceiveTimeout(0);
SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1);
MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(adapter);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
assertEquals(3, dispatcher.dispatch());
byte[] result = stream.toByteArray();
assertEquals(9, result.length);
assertEquals(1, result[0]);
}
@Test
public void testMaxMessagesLessThanMessageCountWithMultipleDispatches() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream);
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
dispatcherPolicy.setMaxMessagesPerTask(2);
dispatcherPolicy.setReceiveTimeout(0);
SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1);
MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(adapter);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
assertEquals(2, dispatcher.dispatch());
byte[] result1 = stream.toByteArray();
assertEquals(6, result1.length);
assertEquals(1, result1[0]);
assertEquals(1, dispatcher.dispatch());
byte[] result2 = stream.toByteArray();
assertEquals(9, result2.length);
assertEquals(1, result2[0]);
assertEquals(7, result2[6]);
}
@Test
public void testMaxMessagesExceedsMessageCountWithMultipleDispatches() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream);
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
dispatcherPolicy.setMaxMessagesPerTask(5);
dispatcherPolicy.setReceiveTimeout(0);
SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1);
MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(adapter);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
assertEquals(3, dispatcher.dispatch());
byte[] result1 = stream.toByteArray();
assertEquals(9, result1.length);
assertEquals(1, result1[0]);
assertEquals(0, dispatcher.dispatch());
byte[] result2 = stream.toByteArray();
assertEquals(9, result2.length);
assertEquals(1, result2[0]);
}
@Test
public void testStreamResetBetweenDispatches() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream);
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
dispatcherPolicy.setMaxMessagesPerTask(2);
dispatcherPolicy.setReceiveTimeout(0);
SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1);
MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(adapter);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
assertEquals(2, dispatcher.dispatch());
byte[] result1 = stream.toByteArray();
assertEquals(6, result1.length);
stream.reset();
assertEquals(1, dispatcher.dispatch());
byte[] result2 = stream.toByteArray();
assertEquals(3, result2.length);
assertEquals(7, result2[0]);
}
@Test
public void testStreamWriteBetweenDispatches() throws IOException {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream);
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
dispatcherPolicy.setMaxMessagesPerTask(2);
dispatcherPolicy.setReceiveTimeout(0);
SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1);
MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(adapter);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
assertEquals(2, dispatcher.dispatch());
byte[] result1 = stream.toByteArray();
assertEquals(6, result1.length);
stream.write(new byte[] {123});
stream.flush();
assertEquals(1, dispatcher.dispatch());
byte[] result2 = stream.toByteArray();
assertEquals(10, result2.length);
assertEquals(1, result2[0]);
assertEquals(123, result2[6]);
assertEquals(7, result2[7]);
}
}

111
spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java
View File

@@ -1,111 +0,0 @@
/*
* 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 static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import java.io.ByteArrayInputStream;
import org.junit.Test;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.message.Message;
/**
* @author Mark Fisher
*/
public class CharacterStreamSourceAdapterTests {
@Test
public void testEndOfStream() {
byte[] bytes = "test".getBytes();
ByteArrayInputStream stream = new ByteArrayInputStream(bytes);
MessageChannel channel = new SimpleChannel();
CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(stream);
adapter.setChannel(channel);
adapter.start();
int count = adapter.processMessages();
assertEquals(1, count);
Message<?> message1 = channel.receive(0);
assertEquals("test", message1.getPayload());
Message<?> message2 = channel.receive(0);
assertNull(message2);
adapter.processMessages();
Message<?> message3 = channel.receive(0);
assertNull(message3);
}
@Test
public void testEndOfStreamWithMaxMessagesPerTask() {
byte[] bytes = "test".getBytes();
ByteArrayInputStream stream = new ByteArrayInputStream(bytes);
MessageChannel channel = new SimpleChannel();
CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(stream);
adapter.setChannel(channel);
adapter.setMaxMessagesPerTask(5);
adapter.start();
int count = adapter.processMessages();
assertEquals(1, count);
Message<?> message1 = channel.receive(0);
assertEquals("test", message1.getPayload());
Message<?> message2 = channel.receive(0);
assertNull(message2);
}
@Test
public void testMultipleLinesWithSingleMessagePerTask() {
String s = "test1" + System.getProperty("line.separator") + "test2";
ByteArrayInputStream stream = new ByteArrayInputStream(s.getBytes());
MessageChannel channel = new SimpleChannel();
CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(stream);
adapter.setInitialDelay(10000);
adapter.setMaxMessagesPerTask(1);
adapter.setChannel(channel);
adapter.start();
int count = adapter.processMessages();
assertEquals(1, count);
Message<?> message1 = channel.receive(0);
assertEquals("test1", message1.getPayload());
Message<?> message2 = channel.receive(0);
assertNull(message2);
adapter.processMessages();
Message<?> message3 = channel.receive(0);
assertEquals("test2", message3.getPayload());
}
@Test
public void testLessThanMaxMessagesAvailable() {
String s = "test1" + System.getProperty("line.separator") + "test2";
ByteArrayInputStream stream = new ByteArrayInputStream(s.getBytes());
MessageChannel channel = new SimpleChannel();
CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(stream);
adapter.setChannel(channel);
adapter.setMaxMessagesPerTask(5);
adapter.start();
int count = adapter.processMessages();
assertEquals(2, count);
Message<?> message1 = channel.receive(0);
assertEquals("test1", message1.getPayload());
Message<?> message2 = channel.receive(0);
assertEquals("test2", message2.getPayload());
Message<?> message3 = channel.receive(0);
assertNull(message3);
}
}

191
spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java
View File

@@ -1,191 +0,0 @@
/*
* 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 static org.junit.Assert.assertEquals;
import java.io.ByteArrayOutputStream;
import org.junit.Test;
import org.springframework.integration.channel.DispatcherPolicy;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.dispatcher.DefaultMessageDispatcher;
import org.springframework.integration.dispatcher.MessageDispatcher;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler;
/**
* @author Mark Fisher
*/
public class CharacterStreamTargetAdapterTests {
private SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(1);
@Test
public void testSingleString() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
adapter.handle(new StringMessage("foo"));
String result = new String(stream.toByteArray());
assertEquals("foo", result);
}
@Test
public void testTwoStringsAndNoNewLinesByDefault() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
MessageChannel channel = new SimpleChannel();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(adapter);
channel.send(new StringMessage("foo"), 0);
channel.send(new StringMessage("bar"), 0);
assertEquals(1, dispatcher.dispatch());
String result1 = new String(stream.toByteArray());
assertEquals("foo", result1);
assertEquals(1, dispatcher.dispatch());
String result2 = new String(stream.toByteArray());
assertEquals("foobar", result2);
}
@Test
public void testTwoStringsWithNewLines() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
MessageChannel channel = new SimpleChannel();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
adapter.setShouldAppendNewLine(true);
MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(adapter);
channel.send(new StringMessage("foo"), 0);
channel.send(new StringMessage("bar"), 0);
assertEquals(1, dispatcher.dispatch());
String result1 = new String(stream.toByteArray());
String newLine = System.getProperty("line.separator");
assertEquals("foo" + newLine, result1);
assertEquals(1, dispatcher.dispatch());
String result2 = new String(stream.toByteArray());
assertEquals("foo" + newLine + "bar" + newLine, result2);
}
@Test
public void testMaxMessagesPerTaskSameAsMessageCount() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
dispatcherPolicy.setMaxMessagesPerTask(2);
SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(adapter);
channel.send(new StringMessage("foo"), 0);
channel.send(new StringMessage("bar"), 0);
assertEquals(2, dispatcher.dispatch());
String result = new String(stream.toByteArray());
assertEquals("foobar", result);
}
@Test
public void testMaxMessagesPerTaskExceedsMessageCountWithAppendedNewLines() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
dispatcherPolicy.setMaxMessagesPerTask(10);
dispatcherPolicy.setReceiveTimeout(0);
SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
adapter.setShouldAppendNewLine(true);
dispatcher.addHandler(adapter);
channel.send(new StringMessage("foo"), 0);
channel.send(new StringMessage("bar"), 0);
assertEquals(2, dispatcher.dispatch());
String result = new String(stream.toByteArray());
String newLine = System.getProperty("line.separator");
assertEquals("foo" + newLine + "bar" + newLine, result);
}
@Test
public void testSingleNonStringObject() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
MessageChannel channel = new SimpleChannel();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(adapter);
TestObject testObject = new TestObject("foo");
channel.send(new GenericMessage<TestObject>(testObject));
int count = dispatcher.dispatch();
assertEquals(1, count);
String result = new String(stream.toByteArray());
assertEquals("foo", result);
}
@Test
public void testTwoNonStringObjectWithOutNewLines() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
dispatcherPolicy.setReceiveTimeout(0);
dispatcherPolicy.setMaxMessagesPerTask(2);
SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(adapter);
TestObject testObject1 = new TestObject("foo");
TestObject testObject2 = new TestObject("bar");
channel.send(new GenericMessage<TestObject>(testObject1), 0);
channel.send(new GenericMessage<TestObject>(testObject2), 0);
assertEquals(2, dispatcher.dispatch());
String result = new String(stream.toByteArray());
assertEquals("foobar", result);
}
@Test
public void testTwoNonStringObjectWithNewLines() {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
dispatcherPolicy.setReceiveTimeout(0);
dispatcherPolicy.setMaxMessagesPerTask(2);
SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
adapter.setShouldAppendNewLine(true);
dispatcher.addHandler(adapter);
TestObject testObject1 = new TestObject("foo");
TestObject testObject2 = new TestObject("bar");
channel.send(new GenericMessage<TestObject>(testObject1), 0);
channel.send(new GenericMessage<TestObject>(testObject2), 0);
dispatcher.dispatch();
String result = new String(stream.toByteArray());
String newLine = System.getProperty("line.separator");
assertEquals("foo" + newLine + "bar" + newLine, result);
}
private static class TestObject {
private String text;
TestObject(String text) {
this.text = text;
}
public String toString() {
return this.text;
}
}
}