Dealing with file globs in DSL2

[maxbates]

Hello! I'm ramping up with DSL2, and hit a roadblock dealing with processes outputting a glob of files. I wonder if there is a better way to deal with these lists of files, but hopefully my code may help others.

DSL2 modules are great, but the publishDir directive is unwieldy for processes imported across multiple workflows; I've found it more tenable to explicitly define a tree of results at the end of my workflow. This requires mapping single files, files in directories, and array lists globbed files. I've illustrated this below.

I've been confused by the changes to:

• paths (which may be a single file, a glob or array list of files, or a directory - each with their own behavior)
• files (file() appears to be supported, but the type is not recommended in process outputs in DSL2)
• channels, which mostly seem to have fallen out of favor in DSL2 processes, and I haven't had much need for, except apparently in dealing with file globs, as below.

Ultimately, I'm wondering if there is a better way to define copyFilesFromGlob - and think the docs would benefit from some further explanation or examples dealing with array lists of file paths:

nextflow.enable.dsl=2

process create_files {
	output:
		path '0.txt', emit: "zero"
		path '*.txt', emit: "files"
		path 'dir',   emit: "dir", type: "dir"
	script:
		'''
		echo 'zero' > 0.txt
		echo 'one' > 1.txt
		echo 'two' > 2.txt

		mkdir -p dir
		echo 'three' > dir/3.txt
		echo 'four' > dir/4.txt
		'''
}

def copyFilesFromGlob(files, target) {
	// files is an array of paths
	// appears to require an unintuitive subscribe() - e.g. each() does not work
	files.collect().flatten().subscribe { p -> 
		f = file(p)
		if ( f.isFile() ) { 
			f.copyTo("${target}/${f.getName()}")
		}
	}
}

def copyDirectoryFiles(dir, target) {
	dir.eachFile { f -> 
 		if ( f.isFile() ) { 
 			f.copyTo("${target}/${f.getName()}")
 		}
 	}
}

workflow {
	main:
		create_files()
		(zero, files, dir) = create_files.out

		// creates /tmp/0.txt
		// [expected behavior]
		zero.copyTo("/tmp/0.txt")

		// fails
		// [expected, but docs not clear on what to do instead]
		// files.copyTo("/tmp")

		// fails
		// [unexpected, and not obvious to debug]
		// files.eachFile(f -> f.copyTo("/tmp/${f.getName()}"))

		// fails
		// [was surprising to me that files.eachFile() does not work]
		// copyDirectoryFiles(files, "/tmp")

		// creates /tmp/{1,2}.txt
		// [was not obvious to me in the docs how to achieve this behavior]
		copyFilesFromGlob(files, "/tmp")

		// creates /tmp/{3,4}.txt
		// [expected]
		copyDirectoryFiles(dir, "/tmp")

		// creates /tmp/dir/{3,4}.txt
		// [reasonable, but I had expected the files to copy to /tmp directly]
		// dir.copyTo("/tmp") 
}

Thanks for any pointers!

[maxbates]

Empirically, it seems that copying file globs in this manner is not blocking (because it relies on subscribe()) and, especially when results are cached, some files are not present in my results directory. Guidance on how to ensure that file globs are copied would be greatly appreciated!

[pditommaso]

The process output is always a channel, so what you could do is apply subscribe and then use copyTo in the implementation, eg.

files.subscribe { file->file.copyTo('/some/path') }

[maxbates]

Hi @pditommaso - I use subscribe in my original code, in copyFilesFromGlob. However, it doesn't seem to be blocking, and the files are not always written before the run completes.

[pditommaso]

However, it doesn't seem to be blocking, and the files are not always written before the run completes

That's by design. You should only do this to make a copy of the result, not for inter-processes communication. In the latter case make no sense, because NF knows how to handle them.

[maxbates]

When I say blocking, I mean the nextflow program exits before the subscribe closure is called for all entries. It seems that subscribe closures are not guaranteed to execute. I'm not sure if that's an issue with dataflow or the implementation of subscribe in nextflow's DSL. Specifically, in the example, the files passed to copyFilesFromGlob are not copied to my final tree.

If there were a way to call each, that might solve the issue, but I couldn't figure out how the file type, file glob array lists, and dataflow variables relate enough to get that to work.

[pditommaso]

The correct implementation is

def copyFilesFromGlob(files, target) {
	files.flatten().subscribe { f -> 
			f.copyTo("${target}/")
	}
}

The collect is not needed because you want to get a file at time flatten is already doing that. The f = file(p) is not needed because the file is already a Path object. Same for the if( f.isFile() ) because the channel is only capturing *.txt files.

[stale[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.