Make BlockingOperatorToIterator exert backpressure.

[vqvu]

The iterator created by BlockingOperator#getIterator() doesn't exert backpressure, which causes code like this to never terminate/run out of memory

Observable.from(new Iterable<Integer>() {

    @Override
    public Iterator<Integer> iterator() {
        return new Iterator<Integer>() {

            @Override
            public boolean hasNext() {
                return true;
            }

            @Override
            public Integer next() {
                return 1;
            }
        };
    }
}).toBlocking().getIterator().next();

This PR adds the appropriate request calls so that this works. I had to combine the implementations of Subscriber and Iterator into a single class to get access to request().

[akarnokd]

Setting variables to their default value on construction is a no-op semantically but still mean instructions executed which causes the construction of the object to take longer.

[vqvu]

I'm surprised the compiler doesn't just optimize that out...but I can remove it.

[vqvu]

Ok. I removed the unnecessary variable initialization and added batched requests. The initial batch is RxRingBuffer.SIZE, and subsequent ones are 3 * RxRingBuffer.SIZE / 4 each (once we've received that much).

[akarnokd]

Excellent! 👍

[abersnaze]

👍