package org.broadinstitute.sting.gatk.iterators;

import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * Created by IntelliJ IDEA.
 * User: depristo
 * Date: Feb 24, 2009
 * Time: 10:24:38 AM
 * To change this template use File | Settings | File Templates.
 */
public class ThreadedIterator<T> implements Iterator<T>, Runnable {
    private Iterator<T> it;
    private final BlockingQueue<T> queue;
    private int nOps = 0;
    private final int printStateFreq = -1;

    public void run() {
        try {
            while (it.hasNext()) {
                queue.put(it.next());
                printState("addNext");
            }
        } catch (InterruptedException ex) {
            // bail out
            ;
        }
    }

    public synchronized void printState(final String op) {
        if ( printStateFreq != -1 && nOps++ % printStateFreq == 0 )
            System.out.printf("  [%s] Queue has %d elements %d ops%n", op, queue.size(), nOps);
    }

    public ThreadedIterator(Iterator<T> it, int buffSize) {
        this.it = it;
        queue = new LinkedBlockingQueue<T>(buffSize);
        new Thread(this).start();
    }

    public boolean hasNext() {
        return queue.peek() != null || it.hasNext();
    }

    public T next() {
        printState("getNext");
        try {
            return queue.poll(10, TimeUnit.SECONDS);
        } catch (InterruptedException ex) {
            // bail out
            System.out.printf("ThreadedIterator next() timed out...%n");
            printState("getNext");
            return null;
        }
    }

    public void remove () {
        throw new UnsupportedOperationException();
    }
}
@@ -43,16 +43,49 @@
 ators;%0A%0A
+import org.apache.log4j.Logger;%0A%0A
 import j
@@ -605,16 +605,96 @@
  = -1;%0A%0A
+    protected static Logger logger = Logger.getLogger(ThreadedIterator.class);%0A%0A
     publ
@@ -754,24 +754,284 @@
 asNext()) %7B%0A
+                // Spin loop...if queue has remaining capacity, add more elements.%0A                // Otherwise, yield to the thread(s) waiting for the data.%0A                if(queue.remainingCapacity() %3E 0 ) %7B%0A                    synchronized(this) %7B%0A        
             
@@ -1056,16 +1056,24 @@
 ext());%0A
+        
         
@@ -1095,32 +1095,129 @@
 ate(%22addNext%22);%0A
+                    %7D%0A                %7D%0A                else%0A                    Thread.yield();%0A
             %7D%0A  
@@ -1452,33 +1452,41 @@
         
-System.out.printf
+logger.info(String.format
 (%22  %5B%25s%5D
@@ -1542,16 +1542,17 @@
 ), nOps)
+)
 ;%0A    %7D%0A
@@ -1740,16 +1740,29 @@
  public 
+synchronized 
 boolean 
@@ -2037,25 +2037,19 @@
     
-System.out.printf
+logger.info
 (%22Th
@@ -2082,18 +2082,16 @@
 d out...
-%25n
 %22);%0A    

Make the operation 'read from the iterator and place on the queue' atomic with respect to hasNext(), next().