Removed flush() method from VCFWriter interface; added takeOwnershipOfInner parameter in constructor of wrapper VCFWriters to designate if the Writer should close the inner Writer it receives on construction

git-svn-id: file:///humgen/gsa-scr1/gsa-engineering/svn_contents/trunk@4546 348d0f76-0448-11de-a6fe-93d51630548a

java/src/org/broadinstitute/sting/gatk/walkers/phasing/MergePhasedSegregatingPolymorphismsToMNPvcfWriter.java

@@ -31,7 +31,6 @@ import org.broad.tribble.util.variantcontext.VariantContext;
 import org.broad.tribble.vcf.VCFHeader;
 import org.broad.tribble.vcf.VCFWriter;
 import org.broadinstitute.sting.gatk.contexts.variantcontext.VariantContextUtils;
-import org.broadinstitute.sting.utils.GenomeLoc;
 import org.broadinstitute.sting.utils.exceptions.ReviewedStingException;
 
 import java.io.File;
@@ -51,17 +50,10 @@ public class MergePhasedSegregatingPolymorphismsToMNPvcfWriter implements VCFWri
 
     private Logger logger;
 
-    private static class VCFRecord {
+    // Should we call innerWriter.close() in close()
-        public VariantContext vc;
+    private boolean takeOwnershipOfInner;
-        public byte refBase;
 
-        public VCFRecord(VariantContext vc, byte refBase) {
+    public MergePhasedSegregatingPolymorphismsToMNPvcfWriter(VCFWriter innerWriter, File referenceFile, int maxGenomicDistanceForMNP, Logger logger, boolean takeOwnershipOfInner) {
-            this.vc = vc;
-            this.refBase = refBase;
-        }
-    }
-
-    public MergePhasedSegregatingPolymorphismsToMNPvcfWriter(VCFWriter innerWriter, File referenceFile, int maxGenomicDistanceForMNP, Logger logger) {
         this.innerWriter = innerWriter;
         this.referenceFileForMNPmerging = new IndexedFastaSequenceFile(referenceFile);
         this.maxGenomicDistanceForMNP = maxGenomicDistanceForMNP;
@@ -69,6 +61,11 @@ public class MergePhasedSegregatingPolymorphismsToMNPvcfWriter implements VCFWri
         this.filteredVcfrList = new LinkedList<VCFRecord>();
         this.numMergedRecords = 0;
         this.logger = logger;
+        this.takeOwnershipOfInner = takeOwnershipOfInner;
+    }
+
+    public MergePhasedSegregatingPolymorphismsToMNPvcfWriter(VCFWriter innerWriter, File referenceFile, int maxGenomicDistanceForMNP, Logger logger) {
+        this(innerWriter, referenceFile, maxGenomicDistanceForMNP, logger, false); // by default, don't own inner
     }
 
     public void writeHeader(VCFHeader header) {
@@ -76,13 +73,10 @@ public class MergePhasedSegregatingPolymorphismsToMNPvcfWriter implements VCFWri
     }
 
     public void close() {
-        flush();
-        innerWriter.close();
-    }
-
-    public void flush() {
         stopWaitingToMerge();
-        innerWriter.flush();        
+
+        if (takeOwnershipOfInner)
+            innerWriter.close();
     }
 
     public void add(VariantContext vc, byte refBase) {
@@ -161,4 +155,14 @@ public class MergePhasedSegregatingPolymorphismsToMNPvcfWriter implements VCFWri
     public String toString() {
         return getClass().getName();
     }
+
+    private static class VCFRecord {
+        public VariantContext vc;
+        public byte refBase;
+
+        public VCFRecord(VariantContext vc, byte refBase) {
+            this.vc = vc;
+            this.refBase = refBase;
+        }
+    }
 }

java/src/org/broadinstitute/sting/gatk/walkers/phasing/MergeSegregatingPolymorphismsWalker.java

@@ -66,7 +66,8 @@ public class MergeSegregatingPolymorphismsWalker extends RodWalker<Integer, Inte
     }
 
     private void initializeVcfWriter() {
-        vcMergerWriter = new MergePhasedSegregatingPolymorphismsToMNPvcfWriter(writer, getToolkit().getArguments().referenceFile, maxGenomicDistanceForMNP, logger);
+        // false <-> don't take control of writer, since didn't create it:
+        vcMergerWriter = new MergePhasedSegregatingPolymorphismsToMNPvcfWriter(writer, getToolkit().getArguments().referenceFile, maxGenomicDistanceForMNP, logger, false);
         writer = null; // so it can't be accessed directly [i.e., not through vcMergerWriter]
 
         // setup the header fields:
@@ -132,7 +133,7 @@ public class MergeSegregatingPolymorphismsWalker extends RodWalker<Integer, Inte
      * @param result Empty for now...
      */
     public void onTraversalDone(Integer result) {
-        vcMergerWriter.flush();
+        vcMergerWriter.close();
         System.out.println("Number of records merged: " + vcMergerWriter.getNumMergedRecords());
     }
 }

java/src/org/broadinstitute/sting/gatk/walkers/phasing/ReadBackedPhasingWalker.java

@@ -124,8 +124,11 @@ public class ReadBackedPhasingWalker extends RodWalker<PhasingStatsAndOutput, Ph
     }
 
     private void initializeVcfWriter() {
+        // Wrapper VCFWriters will take ownership of inner writers iff: inner writer != origWriter [which wasn't created here]
+        VCFWriter origWriter = writer;
+
         if (enableMergePhasedSegregatingPolymorphismsToMNP)
-            writer = new MergePhasedSegregatingPolymorphismsToMNPvcfWriter(writer, getToolkit().getArguments().referenceFile, maxGenomicDistanceForMNP, logger);
+            writer = new MergePhasedSegregatingPolymorphismsToMNPvcfWriter(writer, getToolkit().getArguments().referenceFile, maxGenomicDistanceForMNP, logger, writer != origWriter);
 
         /* Due to discardIrrelevantPhasedSites(), the startDistance spanned by [partiallyPhasedSites.peek(), unphasedSiteQueue.peek()] is <= cacheWindow
            Due to processQueue(), the startDistance spanned by [unphasedSiteQueue.peek(), mostDownstreamLocusReached] is <= cacheWindow
@@ -136,7 +139,7 @@ public class ReadBackedPhasingWalker extends RodWalker<PhasingStatsAndOutput, Ph
            But, NOTE that map() is careful to pass out a list of records to be written that FIRST includes any records discarded due to having reached mostDownstreamLocusReached,
            and only THEN records located at mostDownstreamLocusReached.  The opposite order in map() would violate the startDistance limits imposed when contracting SortingVCFWriter with (2 * cacheWindow).
          */
-        writer = new SortingVCFWriter(writer, 2 * cacheWindow);
+        writer = new SortingVCFWriter(writer, 2 * cacheWindow, writer != origWriter);
 
         // setup the header fields:
         Set<VCFHeaderLine> hInfo = new HashSet<VCFHeaderLine>();
@@ -923,7 +926,7 @@ public class ReadBackedPhasingWalker extends RodWalker<PhasingStatsAndOutput, Ph
     public void onTraversalDone(PhasingStats result) {
         List<VariantContext> finalList = processQueue(result, true); // process all remaining data
         writeVcList(finalList);
-        writer.flush();
+        writer.close();
 
         if (statsWriter != null)
             statsWriter.close();