extend observer with onFileStage

Signed-off-by: Robrecht Cannoodt <[email protected]>

Author: rcannood

modules/nextflow/src/main/groovy/nextflow/Session.groovy

@@ -1149,6 +1149,18 @@ class Session implements ISession {
         }
     }
 
+    void notifyFileStage(Path destination, Path source=null) {
+        def copy = new ArrayList<TraceObserver>(observers)
+        for( TraceObserver observer : copy  ) {
+            try {
+                observer.onFileStage(destination, source)
+            }
+            catch( Exception e ) {
+                log.error "Failed to invoke observer on file stage: $observer", e
+            }
+        }
+    }
+
     void notifyFlowComplete() {
         def copy = new ArrayList<TraceObserver>(observers)
         for( TraceObserver observer : copy  ) {

modules/nextflow/src/main/groovy/nextflow/file/FilePorter.groovy

@@ -105,7 +105,7 @@ class FilePorter {
     }
 
     protected FileTransfer createFileTransfer(Path source, Path target) {
-        return new FileTransfer(source, target, maxRetries, semaphore)
+        return new FileTransfer(source, target, maxRetries, semaphore, session)
     }
 
     protected FileTransfer getOrSubmit(FileCopy copy) {
@@ -281,15 +281,17 @@ class FilePorter {
         volatile Future result
         private String message
         private int debugDelay
+        final private Session session
 
-        FileTransfer(Path foreignPath, Path stagePath, int maxRetries, Semaphore semaphore) {
+        FileTransfer(Path foreignPath, Path stagePath, int maxRetries, Semaphore semaphore, Session session) {
             this.semaphore = semaphore
             this.source = foreignPath
             this.target = stagePath
             this.maxRetries = maxRetries
             this.message = "Staging foreign file: ${source.toUriString()}"
             this.refCount = new AtomicInteger(0)
             this.debugDelay = System.getProperty('filePorter.debugDelay') as Integer ?: 0
+            this.session = session
         }
 
         @Override
@@ -325,7 +327,9 @@ class FilePorter {
             int count = 0
             while( true ) {
                 try {
-                    return stageForeignFile0(filePath, stagePath)
+                    def output = stageForeignFile0(filePath, stagePath)
+                    this.session.notifyFileStage(stagePath, filePath)
+                    return output
                 }
                 catch( IOException e ) {
                     // remove the target file that could be have partially downloaded

modules/nextflow/src/main/groovy/nextflow/trace/TraceObserver.groovy

@@ -150,4 +150,15 @@ trait TraceObserver {
     void onFilePublish(Path destination, Path source){
         onFilePublish(destination)
     }
+
+    /**
+     * Method that is invoke when an output file is staged
+     * into the work directory.
+     *
+     * @param destination
+     *      The destination path at staging folder.
+     * @param source
+     *      The source remote source path.
+     */
+    void onFileStage(Path destination, Path source){}
 }