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

polishing

Browse Source
This commit is contained in:
Mark Fisher
2010-11-18 22:36:31 -05:00
parent 11477bbe63
commit c06e2218b6
5 changed files with 137 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java
View File

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.sftp.config;
import org.w3c.dom.Element;
@@ -25,12 +26,12 @@ import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.util.StringUtils;
/**
* Parser for 'sftp:inbound-channel-adapter'
*
* @author Oleg Zhurakousky
* @since 2.0
*
*/
public class SftpInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
@@ -38,44 +39,35 @@ public class SftpInboundChannelAdapterParser extends AbstractPollingInboundChann
protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
String sessionFactoryName = element.getAttribute("session-factory");
String autoStartup = element.getAttribute("auto-startup");
String fileNamePattern = element.getAttribute("filename-pattern");
String filter = element.getAttribute("filter");
boolean hasFileNamePattern = StringUtils.hasText(fileNamePattern);
boolean hasFilter = StringUtils.hasText(filter);
if (hasFileNamePattern | hasFilter){
if (hasFileNamePattern || hasFilter) {
if (!(hasFileNamePattern ^ hasFilter)) {
throw new BeanDefinitionStoreException("exactly one of 'filename-pattern' or 'filter' " +
throw new BeanDefinitionStoreException("at most one of 'filename-pattern' or 'filter' " +
"is allowed on SFTP inbound adapter");
}
}
BeanDefinitionBuilder sessionPollBuilder =
BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.sftp.session.QueuedSftpSessionPool");
BeanDefinitionBuilder sessionPollBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.sftp.session.QueuedSftpSessionPool");
sessionPollBuilder.addConstructorArgReference(sessionFactoryName);
sessionPollBuilder.addPropertyValue("autoStartup", autoStartup);
String sessionPollName =
BeanDefinitionReaderUtils.registerWithGeneratedName(sessionPollBuilder.getBeanDefinition(), parserContext.getRegistry());
BeanDefinitionBuilder synchronizerBuilder =
BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.sftp.inbound.SftpInboundSynchronizer");
String sessionPollName = BeanDefinitionReaderUtils.registerWithGeneratedName(
sessionPollBuilder.getBeanDefinition(), parserContext.getRegistry());
BeanDefinitionBuilder synchronizerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.sftp.inbound.SftpInboundSynchronizer");
synchronizerBuilder.addConstructorArgReference(sessionPollName);
synchronizerBuilder.addPropertyValue("autoStartup", autoStartup);
IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "remote-directory", "remotePath");
IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "local-directory");
IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "auto-delete-remote-files-on-sync", "shouldDeleteSourceFile");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(synchronizerBuilder, element, "filter");
BeanDefinitionBuilder messageSourceBuilder =
BeanDefinitionBuilder.rootBeanDefinition("org.springframework.integration.sftp.inbound.SftpInboundSynchronizingMessageSource");
BeanDefinitionBuilder messageSourceBuilder = BeanDefinitionBuilder.rootBeanDefinition(
"org.springframework.integration.sftp.inbound.SftpInboundSynchronizingMessageSource");
messageSourceBuilder.addPropertyValue("synchronizer", synchronizerBuilder.getBeanDefinition());
if (hasFileNamePattern){
if (parserContext.getRegistry().containsBeanDefinition(fileNamePattern)){
if (hasFileNamePattern) {
if (parserContext.getRegistry().containsBeanDefinition(fileNamePattern)) {
messageSourceBuilder.addPropertyReference("filenamePattern", fileNamePattern);
}
else {
@@ -84,7 +76,7 @@ public class SftpInboundChannelAdapterParser extends AbstractPollingInboundChann
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(messageSourceBuilder, element, "auto-create-directories");
IntegrationNamespaceUtils.setValueIfAttributeDefined(messageSourceBuilder, element, "local-directory");
return messageSourceBuilder.getBeanDefinition();
}
}
}

59
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpOutboundChannelAdapterParser.java
View File

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.sftp.config;
import org.w3c.dom.Element;
@@ -27,8 +28,9 @@ import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.util.StringUtils;
/**
* Parser for 'sftp:outbound-channel-adapter'
* Parser for the 'sftp:outbound-channel-adapter' element.
*
* @author Oleg Zhurakousky
* @since 2.0
@@ -37,56 +39,51 @@ public class SftpOutboundChannelAdapterParser extends AbstractOutboundChannelAda
@Override
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
BeanDefinitionBuilder sessionPollBuilder =
BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.sftp.session.QueuedSftpSessionPool");
sessionPollBuilder.addConstructorArgReference(element.getAttribute("session-factory"));
String sessionPollName =
BeanDefinitionReaderUtils.registerWithGeneratedName(sessionPollBuilder.getBeanDefinition(), parserContext.getRegistry());
BeanDefinitionBuilder handlerBuilder =
BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.sftp.outbound.SftpSendingMessageHandler");
handlerBuilder.addConstructorArgReference(sessionPollName);
BeanDefinitionBuilder sessionPoolBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.sftp.session.QueuedSftpSessionPool");
sessionPoolBuilder.addConstructorArgReference(element.getAttribute("session-factory"));
String sessionPoolName = BeanDefinitionReaderUtils.registerWithGeneratedName(
sessionPoolBuilder.getBeanDefinition(), parserContext.getRegistry());
BeanDefinitionBuilder handlerBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.sftp.outbound.SftpSendingMessageHandler");
handlerBuilder.addConstructorArgReference(sessionPoolName);
IntegrationNamespaceUtils.setValueIfAttributeDefined(handlerBuilder, element, "charset");
String remoteDirectory = element.getAttribute("remote-directory");
String remoteDirectoryExpression = element.getAttribute("remote-directory-expression");
boolean hasDirectory = StringUtils.hasText(remoteDirectory);
boolean hasDirectoryExpression = StringUtils.hasText(remoteDirectoryExpression);
if (!(hasDirectory ^ hasDirectoryExpression)) {
throw new BeanDefinitionStoreException("exactly one of 'remote-directory' or 'remote-directory-expression' " +
"is required on SFTP outbound adapter");
"is required on the SFTP outbound adapter");
}
BeanDefinition expressionDef = null;
if (hasDirectory){
if (hasDirectory) {
expressionDef = new RootBeanDefinition("org.springframework.expression.common.LiteralExpression");
expressionDef.getConstructorArgumentValues().addGenericArgumentValue(remoteDirectory);
}
else if (hasDirectoryExpression){
else if (hasDirectoryExpression) {
expressionDef = new RootBeanDefinition("org.springframework.integration.config.ExpressionFactoryBean");
expressionDef.getConstructorArgumentValues().addGenericArgumentValue(remoteDirectoryExpression);
}
handlerBuilder.addPropertyValue("remoteDirectoryExpression", expressionDef);
String remoteFileExpression = element.getAttribute("remote-filename-generator-expression");
String fileNameGenerator = element.getAttribute("remote-filename-generator");
boolean hasRemoteFileExp = StringUtils.hasText(remoteFileExpression);
boolean hasFileNameGener = StringUtils.hasText(fileNameGenerator);
if (hasRemoteFileExp | hasFileNameGener){
if (!(hasRemoteFileExp ^ hasFileNameGener)) {
throw new BeanDefinitionStoreException("exactly one of 'remote-filename-generator-expression' or 'remote-filename-generator' " +
"is allowed on SFTP outbound adapter");
boolean hasRemoteFileExpression = StringUtils.hasText(remoteFileExpression);
boolean hasFileNameGenerator = StringUtils.hasText(fileNameGenerator);
if (hasRemoteFileExpression || hasFileNameGenerator) {
if (hasRemoteFileExpression && hasFileNameGenerator) {
throw new BeanDefinitionStoreException("at most one of 'remote-filename-generator-expression' or 'remote-filename-generator' " +
"is allowed on the SFTP outbound adapter");
}
if (StringUtils.hasText(remoteFileExpression)){
BeanDefinitionBuilder fNameGenerBuilder =
BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.file.DefaultFileNameGenerator");
fNameGenerBuilder.addPropertyValue("expression", remoteFileExpression);
handlerBuilder.addPropertyValue("filenameGenerator", fNameGenerBuilder.getBeanDefinition());
if (StringUtils.hasText(remoteFileExpression)) {
BeanDefinitionBuilder fileNameGeneratorBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.file.DefaultFileNameGenerator");
fileNameGeneratorBuilder.addPropertyValue("expression", remoteFileExpression);
handlerBuilder.addPropertyValue("fileNameGenerator", fileNameGeneratorBuilder.getBeanDefinition());
}
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(handlerBuilder, element, "remote-filename-generator", "filenameGenerator");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(handlerBuilder, element, "remote-filename-generator", "fileNameGenerator");
return handlerBuilder.getBeanDefinition();
}
}
}

40
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/outbound/SftpSendingMessageHandler.java
View File

@@ -53,7 +53,7 @@ import com.jcraft.jsch.ChannelSftp;
* @author Oleg Zhurakousky
* @since 2.0
*/
public class SftpSendingMessageHandler extends AbstractMessageHandler{
public class SftpSendingMessageHandler extends AbstractMessageHandler {
private static final String TEMPORARY_FILE_SUFFIX = ".writing";
@@ -63,25 +63,27 @@ public class SftpSendingMessageHandler extends AbstractMessageHandler{
private volatile Expression remoteDirectoryExpression;
private volatile FileNameGenerator filenameGenerator = new DefaultFileNameGenerator();
private volatile FileNameGenerator fileNameGenerator = new DefaultFileNameGenerator();
private volatile File temporaryBufferFolderFile;
private volatile Resource temporaryBufferFolder = new FileSystemResource(SystemUtils.getJavaIoTmpDir());
private volatile String charset = Charset.defaultCharset().name();
public SftpSendingMessageHandler(SftpSessionPool sessionPool) {
Assert.notNull(sessionPool, "'sessionPool' must not be null");
this.sessionPool = sessionPool;
}
public void setTemporaryBufferFolder(Resource temporaryBufferFolder) {
this.temporaryBufferFolder = temporaryBufferFolder;
}
public void setFilenameGenerator(FileNameGenerator filenameGenerator) {
this.filenameGenerator = filenameGenerator;
public void setFileNameGenerator(FileNameGenerator fileNameGenerator) {
this.fileNameGenerator = fileNameGenerator;
}
public void setRemoteDirectoryExpression(Expression remoteDirectoryExpression) {
@@ -92,11 +94,12 @@ public class SftpSendingMessageHandler extends AbstractMessageHandler{
this.charset = charset;
}
@Override
protected void onInit() throws Exception {
this.temporaryBufferFolderFile = this.temporaryBufferFolder.getFile();
if (remoteDirectoryExpression != null){
directoryExpressionProcesor =
new ExpressionEvaluatingMessageProcessor<String>(remoteDirectoryExpression, String.class);
if (this.remoteDirectoryExpression != null) {
this.directoryExpressionProcesor =
new ExpressionEvaluatingMessageProcessor<String>(this.remoteDirectoryExpression, String.class);
}
}
@@ -105,12 +108,12 @@ public class SftpSendingMessageHandler extends AbstractMessageHandler{
File inboundFilePayload = this.redeemForStorableFile(message);
try {
if ((inboundFilePayload != null) && inboundFilePayload.exists()) {
sendFileToRemoteEndpoint(message, inboundFilePayload);
this.sendFileToRemoteEndpoint(message, inboundFilePayload);
}
}
catch (Exception e) {
throw new MessageDeliveryException(message, "Failed to transfer '" + message.getPayload() + "' to" +
" " + this.remoteDirectoryExpression.getExpressionString(), e);
throw new MessageDeliveryException(message, "Failed to transfer '" + message.getPayload() + "' to " +
this.remoteDirectoryExpression.getExpressionString(), e);
}
finally {
if (inboundFilePayload != null && inboundFilePayload.exists()) {
@@ -119,7 +122,6 @@ public class SftpSendingMessageHandler extends AbstractMessageHandler{
}
}
private File handleFileMessage(File sourceFile, File tempFile, File resultFile) throws IOException {
FileCopyUtils.copy(sourceFile, tempFile);
tempFile.renameTo(resultFile);
@@ -142,7 +144,7 @@ public class SftpSendingMessageHandler extends AbstractMessageHandler{
private File redeemForStorableFile(Message<?> message) throws MessageDeliveryException {
try {
Object payload = message.getPayload();
String generateFileName = this.filenameGenerator.generateFileName(message);
String generateFileName = this.fileNameGenerator.generateFileName(message);
File tempFile = new File(this.temporaryBufferFolderFile, generateFileName + TEMPORARY_FILE_SUFFIX);
File resultFile = new File(this.temporaryBufferFolderFile, generateFileName);
File sendableFile = null;
@@ -157,8 +159,8 @@ public class SftpSendingMessageHandler extends AbstractMessageHandler{
}
return sendableFile;
}
catch (Throwable th) {
throw new MessageDeliveryException(message, "Failed to create sendable file.", th);
catch (Exception e) {
throw new MessageDeliveryException(message, "Failed to create sendable file.", e);
}
}
@@ -173,13 +175,12 @@ public class SftpSendingMessageHandler extends AbstractMessageHandler{
ChannelSftp sftp = session.getChannel();
fileInputStream = new FileInputStream(file);
String baseOfRemotePath = "";
if (directoryExpressionProcesor != null){
String result = directoryExpressionProcesor.processMessage(message);
if (StringUtils.hasText(result)){
if (this.directoryExpressionProcesor != null) {
String result = this.directoryExpressionProcesor.processMessage(message);
if (StringUtils.hasText(result)) {
baseOfRemotePath = result;
}
}
if (!StringUtils.endsWithIgnoreCase(baseOfRemotePath, "/")) {
baseOfRemotePath += "/";
}
@@ -191,4 +192,5 @@ public class SftpSendingMessageHandler extends AbstractMessageHandler{
this.sessionPool.release(session);
}
}
}

124
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/QueuedSftpSessionPool.java
View File

@@ -29,7 +29,7 @@ import com.jcraft.jsch.Session;
/**
* This approach - of having a SessionPool ({@link SftpSessionPool}) that has an
* implementation of Queued*SessionPool ({@link QueuedSftpSessionPool}) - was
* implementation of a queued SessionPool ({@link QueuedSftpSessionPool}) - was
* taken almost directly from the Spring Integration FTP adapter.
*
* @author Josh Long
@@ -37,23 +37,27 @@ import com.jcraft.jsch.Session;
* @since 2.0
*/
public class QueuedSftpSessionPool implements SftpSessionPool, SmartLifecycle {
private final ReentrantLock atomicOperationLock = new ReentrantLock();
private static Logger logger = Logger.getLogger(QueuedSftpSessionPool.class.getName());
public static final int DEFAULT_POOL_SIZE = 10;
private volatile Queue<SftpSession> queue;
private final SftpSessionFactory sftpSessionFactory;
private final int maxPoolSize;
private volatile boolean started;
private volatile boolean running;
private volatile boolean autoStartup;
private volatile int phase = 0;
private final ReentrantLock lock = new ReentrantLock();
public QueuedSftpSessionPool(SftpSessionFactory factory) {
this(DEFAULT_POOL_SIZE, factory);
}
@@ -62,14 +66,19 @@ public class QueuedSftpSessionPool implements SftpSessionPool, SmartLifecycle {
this.sftpSessionFactory = sessionFactory;
this.maxPoolSize = maxPoolSize;
}
public void setAutoStartup(boolean autoStartup) {
this.autoStartup = autoStartup;
}
public void setPhase(int phase) {
this.phase = phase;
}
public SftpSession getSession() throws Exception {
Assert.notNull(this.queue, "SftpSession is unavailable since component is not started");
this.atomicOperationLock.lock();
Assert.notNull(this.queue, "SftpSession is unavailable since the pool component is not started");
this.lock.lock();
try {
SftpSession session = this.queue.poll();
if (null == session) {
@@ -78,14 +87,14 @@ public class QueuedSftpSessionPool implements SftpSessionPool, SmartLifecycle {
return session;
}
finally {
this.atomicOperationLock.unlock();
this.lock.unlock();
}
}
public void release(SftpSession sftpSession) {
if (this.started){
this.atomicOperationLock.lock();
if (this.running) {
this.lock.lock();
try {
if (queue.size() < maxPoolSize && sftpSession != null) {
queue.add(sftpSession);
@@ -95,72 +104,79 @@ public class QueuedSftpSessionPool implements SftpSessionPool, SmartLifecycle {
}
}
finally {
this.atomicOperationLock.unlock();
this.lock.unlock();
}
}
else {
this.destroySftpSession(sftpSession);
}
}
public void start() {
Assert.isTrue(this.maxPoolSize > 0, "poolSize must be greater than 0");
this.atomicOperationLock.lock();
private void destroySftpSession(SftpSession sftpSession) {
try {
this.queue = new ArrayBlockingQueue<SftpSession>(this.maxPoolSize, true);
if (sftpSession != null) {
Channel channel = sftpSession.getChannel();
if (channel.isConnected()) {
channel.disconnect();
}
Session session = sftpSession.getSession();
if (session.isConnected()) {
session.disconnect();
}
}
}
finally {
this.atomicOperationLock.unlock();
}
this.started = true;
}
public void stop() {
for (SftpSession sftpSession : queue) {
this.destroySftpSession(sftpSession);
catch (Throwable e) {
// log and ignore
logger.warning("Exception was thrown while destroying SftpSession. " + e);
}
}
public boolean isRunning() {
return this.started;
}
public int getPhase() {
return 0;
}
// SmartLifeycle implementation
public boolean isAutoStartup() {
return this.autoStartup;
}
public int getPhase() {
return this.phase;
}
public boolean isRunning() {
return this.running;
}
public void start() {
Assert.isTrue(this.maxPoolSize > 0, "poolSize must be greater than 0");
this.lock.lock();
try {
this.queue = new ArrayBlockingQueue<SftpSession>(this.maxPoolSize, true);
this.running = true;
}
finally {
this.lock.unlock();
}
}
public void stop() {
if (this.queue != null) {
for (SftpSession sftpSession : this.queue) {
this.destroySftpSession(sftpSession);
}
}
}
public void stop(Runnable callback) {
this.atomicOperationLock.lock();
this.lock.lock();
try {
this.stop();
callback.run();
}
finally {
this.started = false;
this.atomicOperationLock.unlock();
this.running = false;
this.lock.unlock();
}
}
private void destroySftpSession(SftpSession sftpSession){
try {
if (sftpSession != null){
Channel channel = sftpSession.getChannel();
if (channel.isConnected()){
channel.disconnect();
}
Session session = sftpSession.getSession();
if (session.isConnected()){
session.disconnect();
}
}
} catch (Throwable e) {
// log and ignore
logger.warning("Exception was thrown during while destroying SftpSession. " + e);
}
}
}

4
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/OutboundChannelAdapaterParserTests.java
View File

@@ -52,7 +52,7 @@ public class OutboundChannelAdapaterParserTests {
Expression remoteDirectoryExpression = (Expression) TestUtils.getPropertyValue(handler, "remoteDirectoryExpression");
assertNotNull(remoteDirectoryExpression);
assertTrue(remoteDirectoryExpression instanceof LiteralExpression);
assertEquals(context.getBean("fileNameGenerator"), TestUtils.getPropertyValue(handler, "filenameGenerator"));
assertEquals(context.getBean("fileNameGenerator"), TestUtils.getPropertyValue(handler, "fileNameGenerator"));
assertEquals("UTF-8", TestUtils.getPropertyValue(handler, "charset"));
assertNotNull(TestUtils.getPropertyValue(handler, "temporaryBufferFolder"));
assertNotNull(TestUtils.getPropertyValue(handler, "temporaryBufferFolderFile"));
@@ -74,7 +74,7 @@ public class OutboundChannelAdapaterParserTests {
SpelExpression remoteDirectoryExpression = (SpelExpression) TestUtils.getPropertyValue(handler, "remoteDirectoryExpression");
assertNotNull(remoteDirectoryExpression);
assertEquals("'foo' + '/' + 'bar'", remoteDirectoryExpression.getExpressionString());
FileNameGenerator generator = (FileNameGenerator) TestUtils.getPropertyValue(handler, "filenameGenerator");
FileNameGenerator generator = (FileNameGenerator) TestUtils.getPropertyValue(handler, "fileNameGenerator");
String fileNameGeneratorExpression = (String) TestUtils.getPropertyValue(generator, "expression");
assertEquals("payload.getName() + '-foo'", fileNameGeneratorExpression);
assertEquals("UTF-8", TestUtils.getPropertyValue(handler, "charset"));