Resolve Fusion symlinks when publishing files

modules/nextflow/src/main/groovy/nextflow/processor/PublishDir.groovy

@@ -29,6 +29,7 @@ import java.util.concurrent.ExecutorService
 import groovy.transform.CompileDynamic
 import groovy.transform.CompileStatic
 import groovy.transform.EqualsAndHashCode
+import groovy.transform.Memoized
 import groovy.transform.PackageScope
 import groovy.transform.ToString
 import groovy.util.logging.Slf4j
@@ -38,6 +39,7 @@ import nextflow.Session
 import nextflow.extension.FilesEx
 import nextflow.file.FileHelper
 import nextflow.file.TagAwareFile
+import nextflow.fusion.FusionHelper
 import nextflow.util.PathTrie
 /**
  * Implements the {@code publishDir} directory. It create links or copies the output
@@ -119,6 +121,8 @@ class PublishDir {
 
     private String taskName
 
+    private Map<String,Path> taskInputs
+
     @Lazy
     private ExecutorService threadPool = { def sess = Global.session as Session; sess.publishDirExecutorService() }()
 
@@ -282,6 +286,7 @@ class PublishDir {
         this.sourceFileSystem = sourceDir.fileSystem
         this.stageInMode = task.config.stageInMode
         this.taskName = task.name
+        this.taskInputs = task.getInputFilesMap()
 
         apply0(files)
     }
@@ -364,6 +369,11 @@ class PublishDir {
     @CompileStatic
     protected void processFile( Path source, Path destination ) {
 
+        // resolve Fusion symlink if applicable
+        if( FusionHelper.isFusionEnabled(Global.session as Session) )
+            if( source.name in taskInputs )
+                source = resolveFusionLink(taskInputs[source.name])
+
         // create target dirs if required
         makeDirs(destination.parent)
 
@@ -387,6 +397,31 @@ class PublishDir {
         notifyFilePublish(destination, source)
     }
 
+    /**
+     * Resolve a Fusion symlink by following the .fusion.symlinks
+     * file in the task directory until the original file is reached.
+     *
+     * @param file
+     */
+    @CompileStatic
+    protected Path resolveFusionLink(Path file) {
+        while( file.name in getFusionLinks(file.parent) )
+            file = file.text.replace('/fusion/s3/', 's3://') as Path
+        return file
+    }
+
+    @CompileStatic
+    @Memoized
+    protected List<String> getFusionLinks(Path workDir) {
+        try {
+            final file = workDir.resolve('.fusion.symlinks')
+            return file.text.tokenize('\n')
+        }
+        catch( NoSuchFileException ) {
+            return []
+        }
+    }
+
     private String real0(Path p) {
         try {
             // resolve symlink if it's file in the default (posix) file system