<commit_before>/*
 * Created: Apr 15, 2013
 */
package org.xerial.snappy;

import static java.lang.Math.min;
import static org.xerial.snappy.SnappyFramed.COMPRESSED_DATA_FLAG;
import static org.xerial.snappy.SnappyFramed.HEADER_BYTES;
import static org.xerial.snappy.SnappyFramed.STREAM_IDENTIFIER_FLAG;
import static org.xerial.snappy.SnappyFramed.UNCOMPRESSED_DATA_FLAG;
import static org.xerial.snappy.SnappyFramed.readBytes;
import static org.xerial.snappy.SnappyFramedOutputStream.MAX_BLOCK_SIZE;

import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;
import java.util.Arrays;

/**
 * Implements the <a
 * href="http://snappy.googlecode.com/svn/trunk/framing_format.txt"
 * >x-snappy-framed</a> as an {@link InputStream} and
 * {@link ReadableByteChannel}.
 * 
 * @author <NAME>
 * @since 1.1.0
 */
public final class SnappyFramedInputStream extends InputStream implements
        ReadableByteChannel {

    private final ReadableByteChannel rbc;
    private final ByteBuffer frameHeader;
    private final boolean verifyChecksums;

    /**
     * A single frame read from the underlying {@link InputStream}.
     */
    private ByteBuffer input;

    /**
     * The decompressed data from {@link #input}.
     */
    private ByteBuffer uncompressedDirect;

    /**
     * Indicates if this instance has been closed.
     */
    private boolean closed;

    /**
     * Indicates if we have reached the EOF on {@link #in}.
     */
    private boolean eof;

    /**
     * The position in {@link buffer} to read to.
     */
    private int valid;

    /**
     * The next position to read from {@link #buffer}.
     */
    private int position;

    /**
     * Buffer contains a copy of the uncompressed data for the block.
     */
    private byte[] buffer;

    /**
     * Creates a Snappy input stream to read data from the specified underlying
     * input stream.
     * 
     * @param in
     *            the underlying input stream. Must not be {@code null}.
     */
    public SnappyFramedInputStream(InputStream in) throws IOException {
        this(in, true);
    }

    /**
     * Creates a Snappy input stream to read data from the specified underlying
     * input stream.
     * 
     * @param in
     *            the underlying input stream. Must not be {@code null}.
     * @param verifyChecksums
     *            if true, checksums in input stream will be verified
     */
    public SnappyFramedInputStream(InputStream in, boolean verifyChecksums)
            throws IOException {
        this(Channels.newChannel(in), verifyChecksums);
    }

    /**
     * Creates a Snappy input stream to read data from the specified underlying
     * channel.
     * 
     * @param in
     *            the underlying readable channel. Must not be {@code null}.
     */
    public SnappyFramedInputStream(ReadableByteChannel in)
            throws IOException {
        this(in, true);
    }

    /**
     * Creates a Snappy input stream to read data from the specified underlying
     * channel.
     * 
     * @param in
     *            the underlying readable channel. Must not be {@code null}.
     * @param verifyChecksums
     *            if true, checksums in input stream will be verified
     */
    public SnappyFramedInputStream(ReadableByteChannel in,
            boolean verifyChecksums) throws IOException {
        if (in == null) {
            throw new NullPointerException("in is null");
        }

        this.rbc = in;
        this.verifyChecksums = verifyChecksums;

        allocateBuffersBasedOnSize(MAX_BLOCK_SIZE + 5);
        this.frameHeader = ByteBuffer.allocate(4);

        // stream must begin with stream header
        final byte[] expectedHeader = HEADER_BYTES;
        final byte[] actualHeader = new byte[expectedHeader.length];
        final ByteBuffer actualBuffer = ByteBuffer.wrap(actualHeader);

        // assume that if the input cannot read 4 bytes that something is
        // wrong.
        final int read = in.read(actualBuffer);
        if (read < expectedHeader.length) {
            throw new EOFException(
                    "encountered EOF while reading stream header");
        }
        if (!Arrays.equals(expectedHeader, actualHeader)) {
            throw new IOException("invalid stream header");
        }
    }

    /**
     * @param size
     */
    private void allocateBuffersBasedOnSize(int size) {

        input = ByteBuffer.allocateDirect(size);
        final int maxCompressedLength = Snappy.maxCompressedLength(size);
        uncompressedDirect = ByteBuffer.allocateDirect(maxCompressedLength);
        buffer = new byte[maxCompressedLength];
    }

    @Override
    public int read() throws IOException {
        if (closed) {
            return -1;
        }
        if (!ensureBuffer()) {
            return -1;
        }
        return buffer[position++] & 0xFF;
    }<commit_msg>Read until all bytes needed are read or EOF https://github.com/xerial/snappy-java/issues/62<commit_after>/*
 * Created: Apr 15, 2013
 */
package org.xerial.snappy;

import static java.lang.Math.min;
import static org.xerial.snappy.SnappyFramed.COMPRESSED_DATA_FLAG;
import static org.xerial.snappy.SnappyFramed.HEADER_BYTES;
import static org.xerial.snappy.SnappyFramed.STREAM_IDENTIFIER_FLAG;
import static org.xerial.snappy.SnappyFramed.UNCOMPRESSED_DATA_FLAG;
import static org.xerial.snappy.SnappyFramed.readBytes;
import static org.xerial.snappy.SnappyFramedOutputStream.MAX_BLOCK_SIZE;

import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;
import java.util.Arrays;

/**
 * Implements the <a
 * href="http://snappy.googlecode.com/svn/trunk/framing_format.txt"
 * >x-snappy-framed</a> as an {@link InputStream} and
 * {@link ReadableByteChannel}.
 * 
 * @author <NAME>
 * @since 1.1.0
 */
public final class SnappyFramedInputStream extends InputStream implements
        ReadableByteChannel {

    private final ReadableByteChannel rbc;
    private final ByteBuffer frameHeader;
    private final boolean verifyChecksums;

    /**
     * A single frame read from the underlying {@link InputStream}.
     */
    private ByteBuffer input;

    /**
     * The decompressed data from {@link #input}.
     */
    private ByteBuffer uncompressedDirect;

    /**
     * Indicates if this instance has been closed.
     */
    private boolean closed;

    /**
     * Indicates if we have reached the EOF on {@link #in}.
     */
    private boolean eof;

    /**
     * The position in {@link buffer} to read to.
     */
    private int valid;

    /**
     * The next position to read from {@link #buffer}.
     */
    private int position;

    /**
     * Buffer contains a copy of the uncompressed data for the block.
     */
    private byte[] buffer;

    /**
     * Creates a Snappy input stream to read data from the specified underlying
     * input stream.
     * 
     * @param in
     *            the underlying input stream. Must not be {@code null}.
     */
    public SnappyFramedInputStream(InputStream in) throws IOException {
        this(in, true);
    }

    /**
     * Creates a Snappy input stream to read data from the specified underlying
     * input stream.
     * 
     * @param in
     *            the underlying input stream. Must not be {@code null}.
     * @param verifyChecksums
     *            if true, checksums in input stream will be verified
     */
    public SnappyFramedInputStream(InputStream in, boolean verifyChecksums)
            throws IOException {
        this(Channels.newChannel(in), verifyChecksums);
    }

    /**
     * Creates a Snappy input stream to read data from the specified underlying
     * channel.
     * 
     * @param in
     *            the underlying readable channel. Must not be {@code null}.
     */
    public SnappyFramedInputStream(ReadableByteChannel in)
            throws IOException {
        this(in, true);
    }

    /**
     * Creates a Snappy input stream to read data from the specified underlying
     * channel.
     * 
     * @param in
     *            the underlying readable channel. Must not be {@code null}.
     * @param verifyChecksums
     *            if true, checksums in input stream will be verified
     */
    public SnappyFramedInputStream(ReadableByteChannel in,
            boolean verifyChecksums) throws IOException {
        if (in == null) {
            throw new NullPointerException("in is null");
        }

        this.rbc = in;
        this.verifyChecksums = verifyChecksums;

        allocateBuffersBasedOnSize(MAX_BLOCK_SIZE + 5);
        this.frameHeader = ByteBuffer.allocate(4);

        // stream must begin with stream header
        final byte[] expectedHeader = HEADER_BYTES;
        final byte[] actualHeader = new byte[expectedHeader.length];
        final ByteBuffer actualBuffer = ByteBuffer.wrap(actualHeader);

        final int read = SnappyFramed.readBytes(in, actualBuffer);
        if (read < expectedHeader.length) {
            throw new EOFException(
                    "encountered EOF while reading stream header");
        }
        if (!Arrays.equals(expectedHeader, actualHeader)) {
            throw new IOException("invalid stream header");
        }
    }

    /**
     * @param size
     */
    private void allocateBuffersBasedOnSize(int size) {

        input = ByteBuffer.allocateDirect(size);
        final int maxCompressedLength = Snappy.maxCompressedLength(size);
        uncompressedDirect = ByteBuffer.allocateDirect(maxCompressedLength);
        buffer = new byte[maxCompressedLength];
    }

    @Override
    public int read() throws IOException {
        if (closed) {
            return -1;
        }
        if (!ensureBuffer()) {
            return -1;
        }
        return buffer[position++] & 0xFF;
    }
{'fraction_non_alphanumeric': 0.08811216045175738, 'fraction_numerical': 0.0030182066011099212, 'mean_word_length': 2.485578554462165, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 9, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '13572', 'n_tokens_mistral': 2851, 'n_tokens_neox': 2763, 'n_words': 838}