adds a directory option to obidistribute

pkg/obiformats/dispatcher.go

@@ -1,7 +1,10 @@
 package obiformats
 
 import (
+	"encoding/json"
 	"fmt"
+	"os"
+	"path/filepath"
 	"strings"
 	"sync"
 
@@ -33,11 +36,35 @@ func WriterDispatcher(prototypename string,
 					log.Fatalf("Cannot retreive the new chanel : %v", err)
 				}
 
-				name:=fmt.Sprintf(prototypename, dispatcher.Classifier().Value(newflux))
-				if opt.CompressedFile() && ! strings.HasSuffix(name,".gz") {
+				key := dispatcher.Classifier().Value(newflux)
+				directory := ""
+				if dispatcher.Classifier().Type == "DualAnnotationClassifier" {
+					var keys [2]string
+					err := json.Unmarshal([]byte(key), &keys)
+					if err != nil {
+						log.Fatalf("Error in parsing dispatch key %s", key)
+					}
+					key = keys[0]
+					directory = keys[1]
+				}
+
+				name := fmt.Sprintf(prototypename, key)
+				if opt.CompressedFile() && !strings.HasSuffix(name, ".gz") {
 					name = name + ".gz"
 				}
-				
+
+				if directory != "" {
+					info, err := os.Stat(directory)
+					switch {
+					case !os.IsNotExist(err) && !info.IsDir():
+						log.Fatalln("Cannot Create the directory %s", directory)
+					case os.IsNotExist(err):
+						os.Mkdir(directory, 0755)
+					}
+
+					name = filepath.Join(directory, name)
+				}
+
 				out, err := formater(data,
 					name,
 					options...)

pkg/obiformats/fastseq_write_fasta.go

@@ -89,8 +89,6 @@ func WriteFasta(iterator obiiter.IBioSequence,
 		}
 		close(chunkchan)
 		waitWriter.Wait()
-		obiiter.UnregisterPipe()
-		log.Debugln("End of the fasta file writing")
 	}()
 
 	ff := func(iterator obiiter.IBioSequence) {
@@ -143,6 +141,9 @@ func WriteFasta(iterator obiiter.IBioSequence,
 				file.Close()
 			}
 		}
+
+		log.Debugln("End of the fasta file writing")
+		obiiter.UnregisterPipe()
 		waitWriter.Done()
 
 	}()

pkg/obiformats/fastseq_write_fastq.go

@@ -80,8 +80,6 @@ func WriteFastq(iterator obiiter.IBioSequence,
 		}
 		close(chunkchan)
 		waitWriter.Wait()
-		obiiter.UnregisterPipe()
-		log.Debugln("End of the fastq file writing")
 	}()
 
 	ff := func(iterator obiiter.IBioSequence) {
@@ -134,6 +132,8 @@ func WriteFastq(iterator obiiter.IBioSequence,
 			}
 		}
 
+		log.Debugln("End of the fastq file writing")
+		obiiter.UnregisterPipe()
 		waitWriter.Done()
 	}()
 

pkg/obiformats/universal_read.go

@@ -78,7 +78,7 @@ func ReadSequencesFromFile(filename string,
 	}
 
 	filetype := GuessSeqFileType(string(tag))
-	log.Debug("File guessed format : %s (tag: %s)",
+	log.Debugf("File guessed format : %s (tag: %s)",
 		filetype, (strings.Split(string(tag), "\n"))[0])
 	reader = breader