Function for mapcombinemultimap

lib/WorkflowTaxprofiler.groovy

@@ -4,6 +4,7 @@
 
 import nextflow.Nextflow
 import groovy.text.SimpleTemplateEngine
+import groovyx.gpars.dataflow.DataflowBroadcast
 
 class WorkflowTaxprofiler {
 
@@ -242,4 +243,31 @@ class WorkflowTaxprofiler {
             Nextflow.error(error_string)
         }
     }
+
+
+    /**
+     * Join output from a channel back to the original database the output was associated with
+     *
+     * The channel elements are assumed to be tuples one of [ meta, profile ], and the
+     * database to be of [db_key, meta, database_file].
+     *
+     * @param ch_profile A channel containing a meta and the profilign report of a given profiler
+     * @param ch_database A channel containing a key, the database meta, and the database file/folders itself
+     * @return An multiMap'ed output channel with two sub channels, one with the profile and the other with the db
+     */
+
+    static DataflowBroadcast mapCombineMultimap(DataflowBroadcast ch_profile, DataflowBroadcast ch_database) {
+
+        DataflowBroadcast ch_prepped_channel = ch_profile
+            .map { meta, profile -> [meta.db_name, meta, profile] }
+            .combine(ch_database, by: 0)
+            .multiMap {
+                key, meta, profile, db_meta, db ->
+                    profile: [meta, profile]
+                    db: db
+            }
+
+        return ch_prepped_channel
+    }
+
 }

subworkflows/local/profiling.nf

@@ -239,14 +239,7 @@ workflow PROFILING {
                                                 .filter { meta, db -> meta.tool == 'centrifuge' }
                                                 .map { meta, db -> [meta.db_name, meta, db] }
 
-        ch_input_for_centrifuge_kreport = CENTRIFUGE_CENTRIFUGE.out.report
-                                            .map { meta, profile -> [meta.db_name, meta, profile] }
-                                            .combine(ch_database_for_centrifugekreport, by: 0)
-                                            .multiMap {
-                                                key, meta, profile, db_meta, db ->
-                                                    profile: [meta, profile]
-                                                    db: db
-                                            }
+        ch_input_for_centrifuge_kreport = WorkflowTaxprofiler.mapCombineMultimap(CENTRIFUGE_CENTRIFUGE.out.report, ch_database_for_centrifugekreport)
 
         // Generate profile
         CENTRIFUGE_KREPORT (ch_input_for_centrifuge_kreport.profile, ch_input_for_centrifuge_kreport.db)