fixed alternative destination issues for file transfer plugins

src/python/dm/common/processing/fileProcessingThread.py

@@ -45,7 +45,7 @@ class FileProcessingThread(threading.Thread):
                     for processorKey in self.fileProcessorKeyList: 
                         processorNumber += 1
                         processor = self.fileProcessorDict.get(processorKey)
-                        processorName = processor.__class__.__name__
+                        processorName = '%s-%s' % (processor.__class__.__name__,processorNumber)
                         fileProcessedByDict = fileInfo.get('processedByDict', {})
                         fileInfo['processedByDict'] = fileProcessedByDict
 

src/python/dm/common/processing/plugins/fileTransferPlugin.py

@@ -30,7 +30,7 @@ class FileTransferPlugin(FileProcessor):
         storageDirectory = fileInfo.get('storageDirectory')
         destUrl = self.getDestUrl(filePath, dataDirectory, storageHost, storageDirectory)
         srcUrl = self.getSrcUrl(filePath, dataDirectory)
-        self.start(srcUrl, destUrl)
+        self.start(srcUrl, destUrl, fileInfo)
 
     def getSrcUrl(self, filePath, dataDirectory):
         # Use relative path with respect to data directory as a source
@@ -51,17 +51,17 @@ class FileTransferPlugin(FileProcessor):
     def setDest(self, dest):
         self.dest = dest
 
-    def start(self, src=None, dest=None):
+    def start(self, src=None, dest=None, fileInfo={}):
         # Use preconfigured source if provided source is None
         fileSrc = src
         if src is None:
             fileSrc = self.src
-        # Use provided destination only if preconfigured destination is None
-        # Plugins may have desired destination preconfigured for all files
-        fileDest = self.dest
-        if self.dest is None:
-            fileDest = dest
+        fileDest = dest
+        if dest is None:
+            fileDest = self.dest
 
+        fileSrc = self.replaceTemplates(fileSrc, fileInfo)
+        fileDest = self.replaceTemplates(fileDest, fileInfo)
         if not fileSrc or not fileDest:
             raise InvalidRequest('Both source and destination must be non-empty strings.')
         self.subprocess = DmSubprocess.getSubprocess(self.getFullCommand(fileSrc, fileDest))

src/python/dm/common/processing/plugins/gridftpFileTransferPlugin.py

@@ -37,7 +37,10 @@ class GridftpFileTransferPlugin(FileTransferPlugin):
         (scheme, host, port, dirPath) = FtpUtility.parseFtpUrl(dataDirectory, defaultPort=self.DEFAULT_PORT)
         dirName = os.path.dirname(os.path.relpath(filePath, dirPath)).strip()
         fileName = os.path.basename(filePath)
-        destUrl = 'sshftp://%s/%s/%s/%s' % (storageHost, storageDirectory, dirName, fileName)
+        if self.dest:
+            destUrl = '%s/%s/%s' % (self.dest, dirName, fileName)
+        else:
+            destUrl = 'sshftp://%s/%s/%s/%s' % (storageHost, storageDirectory, dirName, fileName)
         return destUrl
 
     def checkUploadFilesForProcessing(self, filePathsDict, uploadInfo):
@@ -97,10 +100,8 @@ class GridftpFileTransferPlugin(FileTransferPlugin):
             ftpUtility.getMd5Sum(filePath, fileInfo)
 
         # Transfer file
-        srcUrl = self.replaceTemplates(srcUrl, fileInfo)
-        destUrl = self.replaceTemplates(destUrl, fileInfo)
-        self.logger.debug('Starting transfer: %s' % fileInfo)
-        self.start(srcUrl, destUrl)
+        self.logger.debug('Starting transfer: %s -> %s (fileInfo: %s)' % (srcUrl, destUrl, fileInfo))
+        self.start(srcUrl, destUrl, fileInfo)
 
         # Get remote checksum
         if self.remoteMd5Sum:

src/python/dm/common/processing/plugins/rsyncFileTransferPlugin.py

@@ -60,10 +60,8 @@ class RsyncFileTransferPlugin(FileTransferPlugin):
             FileUtility.getMd5Sum(filePath, fileInfo)
 
         # Transfer file
-        srcUrl = self.replaceTemplates(srcUrl, fileInfo)
-        destUrl = self.replaceTemplates(destUrl, fileInfo)
         self.logger.debug('Starting transfer: %s' % fileInfo)
-        self.start(srcUrl, destUrl)
+        self.start(srcUrl, destUrl, fileInfo)
 
         # Get remote checksum
         if self.remoteMd5Sum:

src/python/dm/common/processing/plugins/rsyncWithChecksumAndDeleteFileTransferPlugin.py

@@ -35,7 +35,7 @@ class RsyncWithChecksumAndDeleteFileTransferPlugin(FileTransferPlugin):
         self.logger.debug('File info before transfer: %s' % fileInfo)
 
         # Transfer file
-        self.start(srcUrl, destUrl)
+        self.start(srcUrl, destUrl, fileInfo)
 
         # Get remote checksum
         fileInfo2 = {}