<commit_before>package net.openhft.chronicle.network;

import net.openhft.chronicle.bytes.Bytes;
import net.openhft.chronicle.core.tcp.ISocketChannel;

import java.io.IOException;
import java.nio.ByteBuffer;

import static java.util.Objects.requireNonNull;
import static net.openhft.chronicle.wire.Wires.lengthOf;

/**
 * provides flow control and back pressure to the the socket writer, so that only just one message is read at a time from the TCP/IP socket This class
 * is only used in data grid
 */
public class ReadWithFlowControl implements TcpEventHandler.SocketReader {

    int len = 0;
    boolean hasReadLen = false;
    int position = 0;
    int limit = 4;
    private int rawLen;

    /**
     * reads just a single message from the socket
     */
    public int read(ISocketChannel sc, Bytes<ByteBuffer> inBBB) throws IOException {
        ByteBuffer bb = requireNonNull(inBBB.underlyingObject());
        bb.limit(limit);
        bb.position(position);
        if (hasReadLen) {
            // write the len
            inBBB.writeInt(0, rawLen);
        } else {
            // read the len
            sc.read(bb);

            if (bb.position() < bb.limit())
                return 0;
            rawLen = inBBB.readInt(this.len);
            len = lengthOf(rawLen);
            inBBB.ensureCapacity(this.len + 4);
            limit(this.len + 8);
            position(4);
            limit(len + 4);
            bb.limit(limit);
            hasReadLen = true;
        }

        sc.read(bb);

        // we can read the message, now read the len of the next message
        if (bb.position() == len + 4) {
            position(0);
            limit(4);
            int result = len;
            inBBB.readPositionRemaining(0, len);
            hasReadLen = false;
            len = 0;
            // read all the data from the buffer
            return result;
        }

        // we can read the message and the len
        if (bb.position() == len + 8) {
            position(4);
            int result = len;
            rawLen = inBBB.readInt(this.len);
            this.len = lengthOf(rawLen);
            limit(this.len + 4);
            inBBB.ensureCapacity(this.len + 4);
            bb.position(len + 4);
            inBBB.readPositionRemaining(0, result);
            hasReadLen = true;

            return result;
        }

        if (bb.position() > len + 4)
            throw new IllegalStateException("an error has occurred, position=" + bb.position());

        return 0;
    }

    private void limit(final int limit) {
        this.limit = limit;
    }

    private void position(final int position) {
        this.position = position;
    }

}<commit_msg>if we call ensureCapacity, then we shoudl re-acquire the byte buffer.<commit_after>package net.openhft.chronicle.network;

import net.openhft.chronicle.bytes.Bytes;
import net.openhft.chronicle.core.tcp.ISocketChannel;

import java.io.IOException;
import java.nio.ByteBuffer;

import static java.util.Objects.requireNonNull;
import static net.openhft.chronicle.wire.Wires.lengthOf;

/**
 * provides flow control and back pressure to the the socket writer, so that only just one message is read at a time from the TCP/IP socket This class
 * is only used in data grid
 */
public class ReadWithFlowControl implements TcpEventHandler.SocketReader {

    int len = 0;
    boolean hasReadLen = false;
    int position = 0;
    int limit = 4;
    private int rawLen;

    /**
     * reads just a single message from the socket
     */
    public int read(ISocketChannel sc, Bytes<ByteBuffer> inBBB) throws IOException {
        ByteBuffer bb = requireNonNull(inBBB.underlyingObject());
        bb.limit(limit);
        bb.position(position);
        if (hasReadLen) {
            // write the len
            inBBB.writeInt(0, rawLen);
        } else {
            // read the len
            sc.read(bb);

            if (bb.position() < bb.limit())
                return 0;
            rawLen = inBBB.readInt(this.len);
            len = lengthOf(rawLen);
            inBBB.ensureCapacity(this.len + 4);
            bb = requireNonNull(inBBB.underlyingObject());
            limit(this.len + 8);
            position(4);
            limit(len + 4);
            bb.limit(limit);
            hasReadLen = true;
        }

        sc.read(bb);

        // we can read the message, now read the len of the next message
        if (bb.position() == len + 4) {
            position(0);
            limit(4);
            int result = len;
            inBBB.readPositionRemaining(0, len);
            hasReadLen = false;
            len = 0;
            // read all the data from the buffer
            return result;
        }

        // we can read the message and the len
        if (bb.position() == len + 8) {
            position(4);
            int result = len;
            rawLen = inBBB.readInt(this.len);
            this.len = lengthOf(rawLen);
            limit(this.len + 4);
            inBBB.ensureCapacity(this.len + 4);
            bb = requireNonNull(inBBB.underlyingObject());
            bb.position(len + 4);
            inBBB.readPositionRemaining(0, result);
            hasReadLen = true;

            return result;
        }

        if (bb.position() > len + 4)
            throw new IllegalStateException("an error has occurred, position=" + bb.position());

        return 0;
    }

    private void limit(final int limit) {
        this.limit = limit;
    }

    private void position(final int position) {
        this.position = position;
    }

}
{'fraction_non_alphanumeric': 0.09935321595400647, 'fraction_numerical': 0.007905138339920948, 'mean_word_length': 1.8330788804071247, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, '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': '3403538', 'n_tokens_mistral': 1624, 'n_tokens_neox': 1552, 'n_words': 454}