Added file source adapter and supporting classes.
This commit is contained in:
@@ -0,0 +1,101 @@
|
||||
/*
|
||||
* 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.MessageHandlingException;
|
||||
import org.springframework.integration.message.AbstractMessageMapper;
|
||||
import org.springframework.integration.message.GenericMessage;
|
||||
import org.springframework.integration.message.Message;
|
||||
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.getUidGenerator().generateUid(), 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;
|
||||
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
/*
|
||||
* 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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
/*
|
||||
* 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;
|
||||
|
||||
/**
|
||||
* 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) {
|
||||
return message.getId() + "-" + System.currentTimeMillis() + ".msg";
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
/*
|
||||
* 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 generateFileName(Message<?> message);
|
||||
|
||||
}
|
||||
@@ -0,0 +1,80 @@
|
||||
/*
|
||||
* 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.MessageHandlingException;
|
||||
import org.springframework.integration.adapter.PollableSource;
|
||||
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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
/*
|
||||
* 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 pollInterval) {
|
||||
this(directory, channel, pollInterval, true);
|
||||
}
|
||||
|
||||
public FileSourceAdapter(File directory, MessageChannel channel, int pollInterval, boolean isTextBased) {
|
||||
super(new FileSource(directory), channel, pollInterval);
|
||||
if (isTextBased) {
|
||||
this.setMessageMapper(new TextFileMapper(directory));
|
||||
}
|
||||
else {
|
||||
this.setMessageMapper(new ByteArrayFileMapper(directory));
|
||||
}
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
/*
|
||||
* Copyright 2002-2007 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.integration.adapter.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));
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user