ServiceObserver extends AbstractFuture<CachedService> implements Runnable { public Iterator<RowMetaAndData> rows() { return new Iterator<RowMetaAndData>() { int index = 0; @Override public boolean hasNext() { if ( rowMetaAndData.size() > index ) { return true; } if ( isRunning ) { latch = new CountDownLatch( 1 ); try { while ( !latch.await( 1, TimeUnit.SECONDS ) ) { if ( !isRunning ) { return rowMetaAndData.size() > index; } } return rowMetaAndData.size() > index; } catch ( InterruptedException e ) { return rowMetaAndData.size() > index; } } else { return rowMetaAndData.size() > index; } } @Override public RowMetaAndData next() { return rowMetaAndData.get( index++ ); } }; } ServiceObserver( DataServiceExecutor executor ); Iterator<RowMetaAndData> rows(); ListenableFuture<CachedService> install(); @Override void run();  }
@Test public void testRowIterator() throws Exception { CountDownLatch delayRowsLatch = new CountDownLatch( 1 ); CountDownLatch firstRowLatch = new CountDownLatch( 1 ); ServiceObserver serviceObserver = new ServiceObserver( executor ) { @Override public void run() { rowMetaAndData.add( new RowMetaAndData() ); firstRowLatch.countDown(); try { delayRowsLatch.await(); rowMetaAndData.add( new RowMetaAndData() ); latch.countDown(); rowMetaAndData.add( new RowMetaAndData() ); latch.countDown(); } catch ( InterruptedException e ) { throw new RuntimeException( e ); } isRunning = false; } }; Executors.newSingleThreadExecutor().submit( serviceObserver ); firstRowLatch.await(); Iterator<RowMetaAndData> rows = serviceObserver.rows(); assertTrue( rows.hasNext() ); rows.next(); Executors.newSingleThreadScheduledExecutor().schedule( delayRowsLatch::countDown, 1100, TimeUnit.MILLISECONDS ); assertTrue( rows.hasNext() ); rows.next(); assertTrue( rows.hasNext() ); rows.next(); assertFalse( rows.hasNext() ); delayRowsLatch.countDown(); }