Extend observer with onFileStage

docs/developer/plugins.md

@@ -351,14 +351,19 @@ class MyObserverFactory implements TraceObserverFactory {
 // MyObserver.groovy
 import java.nio.file.Path
 
+import nextflow.Session
 import nextflow.processor.TaskHandler
 import nextflow.trace.TraceObserver
 import nextflow.trace.TraceRecord
 
 class MyObserver implements TraceObserver {
+    private Session session
 
     @Override
-    void onFlowBegin() {
+    void onFlowCreate(Session session) {
+        // store the session for later use
+        this.session = session
+
         println "Okay, let's begin!"
     }
 
@@ -372,9 +377,14 @@ class MyObserver implements TraceObserver {
         println "I found a task in the cache! It's name is '${handler.task.name}'"
     }
 
+    @Override
+    void onFileStage(Path destination, Path source) {
+        println("I staged a file from '${source.toUriString()}' to '${destination.toUriString()}'")
+    }
+
     @Override
     void onFilePublish(Path destination, Path source) {
-        println "I published a file! It's located at ${path.toUriString()}"
+        println "I published a file! It's located at '${destination.toUriString()}'"
     }
 
     @Override

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){}
 }

modules/nextflow/src/test/groovy/nextflow/file/FilePorterTest.groovy

@@ -133,18 +133,6 @@ class FilePorterTest extends Specification {
 
 
 
-    static class ErrorStage extends FilePorter.FileTransfer {
-
-        ErrorStage(Path path, Path stagePath, int maxRetries) {
-            super(path, stagePath, maxRetries, new Semaphore(100))
-        }
-
-        @Override
-        void run() throws Exception {
-            throw new ProcessStageException('Cannot stage file')
-        }
-    }
-
     def 'should submit actions' () {
 
         given:
@@ -283,11 +271,14 @@ class FilePorterTest extends Specification {
     }
     def 'should stage a file' () {
         given:
+        def sess = Mock(Session) {
+            getConfig() >> [:]
+        }
         def folder = Files.createTempDirectory('test')
         def local1 = folder.resolve('hola.text')
         def foreign1 = TestHelper.createInMemTempFile('hola.txt', 'hola mundo!')
         and:
-        def porter = new FilePorter.FileTransfer(foreign1, local1, 0, Mock(Semaphore))
+        def porter = new FilePorter.FileTransfer(foreign1, local1, 0, Mock(Semaphore), sess)
 
         when:
         porter.stageForeignFile(foreign1, local1)
@@ -309,13 +300,16 @@ class FilePorterTest extends Specification {
 
     def 'should check valid files' () {
         given:
+        def sess = Mock(Session) {
+            getConfig() >> [:]
+        }
         def CONTENT = 'hola mundo!'
         def foreign1 = TestHelper.createInMemTempFile('hola.txt', CONTENT)
         and:
         def folder = Files.createTempDirectory('test')
         def local1 = folder.resolve('hola.text'); local1.text = CONTENT
         and:
-        def porter = new FilePorter.FileTransfer(foreign1, local1, 0, Mock(Semaphore))
+        def porter = new FilePorter.FileTransfer(foreign1, local1, 0, Mock(Semaphore), sess)
 
         when:
         def equals = porter.checkPathIntegrity(foreign1, local1)