title: how to terminate retrieval from a blocking queue
body: <p>I have some code where i execute a several tasks using Executors and a Blocking Queue. The results have to be returned as an iterator because that is what the application that i work on expects. However, there is a 1:N relationship between the task and the results added to the queue, so i cannot use the <a href="http://download.oracle.com/docs/cd/E17409_01/javase/6/docs/api/java/util/concurrent/ExecutorCompletionService.html" rel="nofollow noreferrer">ExecutorCompletionService</a>. While calling hasNext(), i need to know when all the tasks have finished and added all the results to the queue, so that i can stop the retrieval of results from the queue. Note, that once items are put on the queue, another thread should be ready to consume (<a href="http://download.oracle.com/docs/cd/E17409_01/javase/6/docs/api/java/util/concurrent/ExecutorService.html#invokeAll(java.util.Collection)" rel="nofollow noreferrer">Executor.invokeAll()</a>, blocks until all tasks have completed, which is not what i want, nor a timeout). This was my first attempt, i am using an AtomicInteger just to demonstrate the point even though it will not work. Could someone help me in undestanding how i can solve this issue?</p>
  
  <pre><code>public class ResultExecutor&lt;T&gt; implements Iterable&lt;T&gt; {
      private BlockingQueue&lt;T&gt; queue;
      private Executor executor;
      private AtomicInteger count;
  
      public ResultExecutor(Executor executor) {
          this.queue = new LinkedBlockingQueue&lt;T&gt;();
          this.executor = executor;
          count = new AtomicInteger();            
      }
  
      public void execute(ExecutorTask task) {
          executor.execute(task);
      }
  
      public Iterator&lt;T&gt; iterator() {
          return new MyIterator();
      }
  
      public class MyIterator implements Iterator&lt;T&gt; {
          private T current;          
          public boolean hasNext() {
              if (count.get() &gt; 0 &amp;&amp; current == null)
              {
                  try {
                      current = queue.take();
                      count.decrementAndGet();
                  } catch (InterruptedException e) {
                      e.printStackTrace();
                  }
              }
              return current != null;
          }
  
          public T next() {
              final T ret = current;
              current = null;
              return ret;
          }
  
          public void remove() {
              throw new UnsupportedOperationException();
          }
  
      }
  
      public class ExecutorTask implements Runnable{
          private String name;
  
          public ExecutorTask(String name) {
              this.name = name;
  
          }
  
           private int random(int n)
           {
             return (int) Math.round(n * Math.random());
           }
  
  
          @SuppressWarnings("unchecked")
          public void run() {
              try {
                  int random = random(500);
                  Thread.sleep(random);
                  queue.put((T) (name + ":" + random + ":1"));
                  queue.put((T) (name + ":" + random + ":2"));
                  queue.put((T) (name + ":" + random + ":3"));
                  queue.put((T) (name + ":" + random + ":4"));
                  queue.put((T) (name + ":" + random + ":5"));
  
                  count.addAndGet(5);
              } catch (InterruptedException e) {
                  e.printStackTrace();
              }                   
          }           
      }       
  
  }
  </code></pre>
  
  <p>And the calling code looks like: </p>
  
  <pre><code>    Executor e = Executors.newFixedThreadPool(2);
      ResultExecutor&lt;Result&gt; resultExecutor = new ResultExecutor&lt;Result&gt;(e);
  
      resultExecutor.execute(resultExecutor.new ExecutorTask("A"));
      resultExecutor.execute(resultExecutor.new ExecutorTask("B"));
  
      Iterator&lt;Result&gt; iter = resultExecutor.iterator();
      while (iter.hasNext()) {
          System.out.println(iter.next());
      }
  </code></pre>