/**
 * Copyright 2016 Palantir Technologies
 *
 * Licensed under the BSD-3 License (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://opensource.org/licenses/BSD-3-Clause
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.palantir.atlasdb.stream;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

import org.apache.commons.lang3.ArrayUtils;

public final class BlockConsumingInputStream extends InputStream {
    private final BlockGetter blockGetter;
    private final long numBlocks;
    private final int blocksInMemory;

    private int nextBlockToRead;
    private Iterator<Byte> buffer;

    public static BlockConsumingInputStream create(
            BlockGetter blockGetter,
            long numBlocks,
            int blocksInMemory) throws IOException {
        BlockConsumingInputStream stream = new BlockConsumingInputStream(blockGetter, numBlocks, blocksInMemory);
        stream.getNextBlock();
        return stream;
    }

    private BlockConsumingInputStream(BlockGetter blockGetter, long numBlocks, int blocksInMemory) {
        this.blockGetter = blockGetter;
        this.numBlocks = numBlocks;
        this.blocksInMemory = blocksInMemory;
        this.nextBlockToRead = 0;
    }

    @Override
    public int read() throws IOException {
        if (buffer.hasNext()) {
            return buffer.next() & 0xff;
        }

        if (nextBlockToRead < numBlocks) {
            getNextBlock();
            return buffer.next() & 0xff;
        }

        return -1;
    }

    private void getNextBlock() throws IOException {
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        int numBlocksToGet = Math.min(blocksLeft(), blocksInMemory);
        blockGetter.get(nextBlockToRead, numBlocksToGet, outputStream);
        nextBlockToRead += numBlocksToGet;
        byte[] bytes = outputStream.toByteArray();
        List<Byte> list = Arrays.asList(ArrayUtils.toObject(bytes));
        buffer = list.iterator();
        outputStream.close();
    }

    private int blocksLeft() {
        return (int) numBlocks - nextBlockToRead;
    }
}

@@ -717,128 +717,8 @@
 eam;
-%0Aimport java.util.Arrays;%0Aimport java.util.Iterator;%0Aimport java.util.List;%0A%0Aimport org.apache.commons.lang3.ArrayUtils;
 %0A%0Apu
@@ -927,24 +927,25 @@
 lockToRead;%0A
+%0A
     private 
@@ -948,24 +948,50 @@
 ate 
-Iterator%3CByte%3E b
+byte%5B%5D buffer;%0A    private int positionInB
 uffe
@@ -1596,16 +1596,51 @@
 ad = 0;%0A
+        this.positionInBuffer = 0;%0A
     %7D%0A%0A 
@@ -1711,24 +1711,40 @@
 if (
-buffer.hasNext()
+positionInBuffer %3C buffer.length
 ) %7B%0A
@@ -1760,39 +1760,52 @@
    return buffer
-.next()
+%5BpositionInBuffer++%5D
  & 0xff;%0A       
@@ -1908,15 +1908,28 @@
 ffer
-.next()
+%5BpositionInBuffer++%5D
  & 0
@@ -2294,19 +2294,13 @@
    b
-yte%5B%5D bytes
+uffer
  = o
@@ -2338,101 +2338,28 @@
     
-List%3CByte%3E list = Arrays.asList(ArrayUtils.toObject(bytes));%0A        buffer = list.iterator()
+positionInBuffer = 0
 ;%0A  

use byte array instead of iterator