<commit_before>/* 
 * Copyright (c) 2011-2016, PCJ Library, <NAME>
 * All rights reserved.
 *
 * Licensed under New BSD License (3-clause license).
 *
 * See the file "LICENSE" for the full license governing this code.
 */
package org.pcj.internal.network;

import java.io.IOException;
import java.io.InputStream;
import java.io.UncheckedIOException;
import java.nio.ByteBuffer;
import java.util.LinkedList;
import java.util.Queue;
import org.pcj.internal.message.Message;
import org.pcj.internal.message.MessageType;

/**
 *
 * @author <NAME> (<EMAIL>)
 */
public class MessageBytesInputStream {

    private static final int HEADER_SIZE = Short.BYTES;
    private final ByteBuffer header;
    private final MessageInputStream messageInputStream;
    private final MessageDataInputStream messageDataInputStream;
    private ByteBuffer currentByteBuffer;

    public MessageBytesInputStream() {
        this.header = ByteBuffer.allocateDirect(HEADER_SIZE);
        this.messageInputStream = new MessageInputStream();
        this.messageDataInputStream = new MessageDataInputStream(messageInputStream);
    }

    public void offerNextBytes(ByteBuffer byteBuffer) {
        while (byteBuffer.hasRemaining()) {
            if (currentByteBuffer == null) {
                while (header.hasRemaining() && byteBuffer.hasRemaining()) {
                    header.put(byteBuffer.get());
                }
                if (header.hasRemaining() == false) {
                    short length = header.getShort(0);

                    if ((length & 0x8000) != 0) {
                        messageInputStream.close();
                    }

//                    int size = length & 0x7FFF;
//                    if (size > 0) {
//                        currentByteBuffer = ByteBuffer.allocate(size);
//                    }
                    currentByteBuffer = ByteBuffer.allocate(length & 0x7FFF);
                    header.clear();
                }
            } else {
                while (currentByteBuffer.hasRemaining() && byteBuffer.hasRemaining()) {
                    currentByteBuffer.put(byteBuffer.get());
                }

                if (currentByteBuffer.hasRemaining() == false) {
                    currentByteBuffer.flip();
                    messageInputStream.offerByteBuffer(currentByteBuffer);
                    currentByteBuffer = null;
                    if (messageInputStream.isClosed()) {
                        return;
                    }
                }
            }
        }
    }

    public Message readMessage() {
        try {
            byte messageType = messageDataInputStream.readByte();

            Message message = MessageType.valueOf(messageType).create();

            return message;
        } catch (IOException ex) {
            throw new UncheckedIOException(ex);
        }
    }
<commit_msg>Finally fix?<commit_after>/* 
 * Copyright (c) 2011-2016, PCJ Library, <NAME>
 * All rights reserved.
 *
 * Licensed under New BSD License (3-clause license).
 *
 * See the file "LICENSE" for the full license governing this code.
 */
package org.pcj.internal.network;

import java.io.IOException;
import java.io.InputStream;
import java.io.UncheckedIOException;
import java.nio.ByteBuffer;
import java.util.LinkedList;
import java.util.Queue;
import org.pcj.internal.message.Message;
import org.pcj.internal.message.MessageType;

/**
 *
 * @author <NAME> (<EMAIL>)
 */
public class MessageBytesInputStream {

    private static final int HEADER_SIZE = Short.BYTES;
    private final ByteBuffer header;
    private final MessageInputStream messageInputStream;
    private final MessageDataInputStream messageDataInputStream;
    private ByteBuffer currentByteBuffer;

    public MessageBytesInputStream() {
        this.header = ByteBuffer.allocateDirect(HEADER_SIZE);
        this.messageInputStream = new MessageInputStream();
        this.messageDataInputStream = new MessageDataInputStream(messageInputStream);
    }

    public void offerNextBytes(ByteBuffer byteBuffer) {
        while (byteBuffer.hasRemaining()) {
            if (currentByteBuffer == null) {
                while (header.hasRemaining() && byteBuffer.hasRemaining()) {
                    header.put(byteBuffer.get());
                }
                if (header.hasRemaining() == false) {
                    short length = header.getShort(0);
                    int size = length & 0x7FFF;

                    header.clear();

                    if ((length & 0x8000) != 0) {
                        messageInputStream.close();
                        if (size == 0) {
                            return;
                        }
                    }

                    currentByteBuffer = ByteBuffer.allocate(size);
                }
            } else {
                while (currentByteBuffer.hasRemaining() && byteBuffer.hasRemaining()) {
                    currentByteBuffer.put(byteBuffer.get());
                }

                if (currentByteBuffer.hasRemaining() == false) {
                    currentByteBuffer.flip();
                    messageInputStream.offerByteBuffer(currentByteBuffer);
                    currentByteBuffer = null;
                    if (messageInputStream.isClosed()) {
                        return;
                    }
                }
            }
        }
    }

    public Message readMessage() {
        try {
            byte messageType = messageDataInputStream.readByte();

            Message message = MessageType.valueOf(messageType).create();

            return message;
        } catch (IOException ex) {
            throw new UncheckedIOException(ex);
        }
    }

{'fraction_non_alphanumeric': 0.09350237717908082, 'fraction_numerical': 0.007043493572812115, 'mean_word_length': 1.845691382765531, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 10, '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': '1694471', 'n_tokens_mistral': 1438, 'n_tokens_neox': 1320, 'n_words': 305}