public void consume(Inbox inbox) {
        ensureNotDone();
        if (limit <= 0) {
            done.compareAndSet(null, new ResultLimitReachedException());
            ensureNotDone();
        }

        while (offset > 0 && inbox.poll() != null) {
            offset--;
        }

        for (JetSqlRow row; (row = (JetSqlRow) inbox.peek()) != null && rows.offer(row); ) {
            inbox.remove();
            if (limit != Long.MAX_VALUE) {
                limit -= 1;
                if (limit < 1) {
                    done.compareAndSet(null, new ResultLimitReachedException());
                    ensureNotDone();
                }
            }
        }
    }
@Test
    public void when_nextItemWhileWaiting_then_hasNextReturns() throws Exception {
        initProducer(false);
        Future<?> future = spawn(() -> {
            assertThat(iterator.hasNext(1, DAYS)).isEqualTo(YES);
            assertThat((int) iterator.next().get(0)).isEqualTo(42);
        });
        sleepMillis(50); // sleep so that the thread starts blocking in `hasNext`

        inbox.queue().add(jetRow(42));
        producer.consume(inbox);
        assertThat(inbox).isEmpty();
        future.get();
    }