title: Non-blocking way to offer + drainTo for a BlockingQueue (or something else)
body: <p>I am looking for something seemingly simple, a collection with a non-blocking version of "add" and "drain". Something like this:</p>
  
  <pre><code>List itemsToProcess = queue.addOrDrainAndAdd( item );
  if ( itemsToProcess != null )
      process( items );
  </code></pre>
  
  <p>It seems to me that if I do these as separate "offer" and "drainTo" calls that I could have offer called twice before I get to the first call to drain to. I would also need a loop on something like "while ( !queue.offer( item ) )" so that after it is drained the offer would work, which I think would require me to also check if drain returned an empty collection (because two might call drain). My naive implementation was like this but it doesn't seem optimal:</p>
  
  <pre><code>void addBatchItem( T item ) {
     while ( !batch.offer( item ) ) {
         List&lt;T&gt; batched = new ArrayList&lt;&gt;( batchSize );
         batch.drainTo( batched );
         process( batched );
     }
  }
  </code></pre>
  
  <p>Then I thought maybe there is a better way and I just don't know it. Thanks!</p>
  
  <p>EDIT:</p>
  
  <p>Okay, so here's a solution (that is blocking based on ArrayBlockingQueue):</p>
  
  <pre><code>public void add( T batchItem ) {
      while ( !batch.offer( batchItem ) ) {
          flush();
      }
  }
  
  public void flush() {
      List&lt;T&gt; batched = new ArrayList&lt;&gt;( batchSize );
      batch.drainTo( batched, batchSize );
      if ( !batched.isEmpty() )
          executor.execute( new PhasedRunnable( batched ) );
  }
  </code></pre>
  
  <p>I guess my question is, would the above be more optimal for this purpose than a solution based on ConcurrentLinkedQueue since the latter requires an object allocation for each node?</p>
  
  <p>EXAMPLE CLASS WITH USAGE:</p>
  
  <pre><code>public abstract class Batcher&lt;T&gt; {
  
      private final int batchSize;
      private ArrayBlockingQueue&lt;T&gt; batch;
      private ExecutorService executor;
      private final Phaser phaser = new Phaser( 1 );
  
      public Batcher( int batchSize, ExecutorService executor ) {
          this.batchSize = batchSize;
          this.executor = executor;
          this.batch = new ArrayBlockingQueue&lt;&gt;( batchSize );
      }
  
      public void add( T batchItem ) {
          while ( !batch.offer( batchItem ) ) {
              flush();
          }
      }
  
      public void flush() {
          List&lt;T&gt; batched = new ArrayList&lt;&gt;( batchSize );
          batch.drainTo( batched, batchSize );
          if ( !batched.isEmpty() )
              executor.execute( new PhasedRunnable( batched ) );
      }
  
      public abstract void onFlush( List&lt;T&gt; batch );
  
      public void awaitDone() {
          phaser.arriveAndAwaitAdvance();
      }
  
      public void awaitDone( long duration, TimeUnit unit ) throws TimeoutException {
          try {
              phaser.awaitAdvanceInterruptibly( phaser.arrive(), duration, unit );
          }
          catch ( InterruptedException e ) {
              Thread.currentThread().interrupt();
          }
      }
  
      private class PhasedRunnable implements Runnable {
  
          private final List&lt;T&gt; batch;
  
          private PhasedRunnable( List&lt;T&gt; batch ) {
              this.batch = batch;
  
              phaser.register();
          }
  
          @Override
          public void run() {
              try {
                  onFlush( batch );
              }
              finally {
                  phaser.arrive();
              }
          }
      }
  }
  </code></pre>
  
  <p>This is a simple example, a more complete example may be JPA entity updates or inserts. Also, I would like it to be possible for #add to be called concurrently.</p>
  
  <pre><code>@Test
  public void testOddNumber() {
      Batcher&lt;Integer&gt; batcher = new Batcher&lt;Integer&gt;( 10, executor ) {
          @Override
          public void onFlush( List&lt;Integer&gt; batch ) {
              count.addAndGet( batch.size() );
          }
      };
  
      for ( int at = 0; at != 21; ++at ) {
          batcher.add( at );
      }
  
      batcher.flush();
      batcher.awaitDone();
  
      assertEquals( count.get(), 21 );
  }
  </code></pre>