@Test public void skipTest() throws InterruptedException { DummyProcessor dp = new DummyProcessor(false); for (int i = 0; i < 10000; i += 5) { Integer res = dp.nextInterruptable(); assertEquals(i, (int) res); dp.skip(4); } }
@Override public long skip(long count) { long skipped = 0; while (skipped < count) { try { if (!ensureBuffer()) { return skipped; } } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new UncheckedInterruptedException(e); } int toSkip = (int) Math.min(count - skipped, _currentResult._length - _currentResultIndex); Arrays.fill(_currentResult._data, _currentResultIndex, _currentResultIndex + toSkip, null); _currentResultIndex += toSkip; skipped += toSkip; } assert skipped == count; return skipped; }
ParallelProcessor implements AutoCloseable, BatchIterator<R> { @Override public long skip(long count) { long skipped = 0; while (skipped < count) { try { if (!ensureBuffer()) { return skipped; } } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new UncheckedInterruptedException(e); } int toSkip = (int) Math.min(count - skipped, _currentResult._length - _currentResultIndex); Arrays.fill(_currentResult._data, _currentResultIndex, _currentResultIndex + toSkip, null); _currentResultIndex += toSkip; skipped += toSkip; } assert skipped == count; return skipped; } }
ParallelProcessor implements AutoCloseable, BatchIterator<R> { @Override public long skip(long count) { long skipped = 0; while (skipped < count) { try { if (!ensureBuffer()) { return skipped; } } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new UncheckedInterruptedException(e); } int toSkip = (int) Math.min(count - skipped, _currentResult._length - _currentResultIndex); Arrays.fill(_currentResult._data, _currentResultIndex, _currentResultIndex + toSkip, null); _currentResultIndex += toSkip; skipped += toSkip; } assert skipped == count; return skipped; } ParallelProcessor(Config config); private  ParallelProcessor(int readerThreadPriority, int processorThreads, int processorThreadPriority,
      int batchSize, int inputBufferSize, int outputBufferSize); }
ParallelProcessor implements AutoCloseable, BatchIterator<R> { @Override public long skip(long count) { long skipped = 0; while (skipped < count) { try { if (!ensureBuffer()) { return skipped; } } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new UncheckedInterruptedException(e); } int toSkip = (int) Math.min(count - skipped, _currentResult._length - _currentResultIndex); Arrays.fill(_currentResult._data, _currentResultIndex, _currentResultIndex + toSkip, null); _currentResultIndex += toSkip; skipped += toSkip; } assert skipped == count; return skipped; } ParallelProcessor(Config config); private  ParallelProcessor(int readerThreadPriority, int processorThreads, int processorThreadPriority,
      int batchSize, int inputBufferSize, int outputBufferSize); R nextInterruptable(); @Override boolean hasNext(); @Override R next(); @Override int next(R[] destination, int offset, int count); @Override long skip(long count); BlockingStatus getBlockingStatus(); @Override long available(); @Override void close(); void start(); }
ParallelProcessor implements AutoCloseable, BatchIterator<R> { @Override public long skip(long count) { long skipped = 0; while (skipped < count) { try { if (!ensureBuffer()) { return skipped; } } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new UncheckedInterruptedException(e); } int toSkip = (int) Math.min(count - skipped, _currentResult._length - _currentResultIndex); Arrays.fill(_currentResult._data, _currentResultIndex, _currentResultIndex + toSkip, null); _currentResultIndex += toSkip; skipped += toSkip; } assert skipped == count; return skipped; } ParallelProcessor(Config config); private  ParallelProcessor(int readerThreadPriority, int processorThreads, int processorThreadPriority,
      int batchSize, int inputBufferSize, int outputBufferSize); R nextInterruptable(); @Override boolean hasNext(); @Override R next(); @Override int next(R[] destination, int offset, int count); @Override long skip(long count); BlockingStatus getBlockingStatus(); @Override long available(); @Override void close(); void start();  }