<commit_before>package org.mitallast.queue.transport.netty.codec;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import org.mitallast.queue.Version;
import org.mitallast.queue.transport.TransportFrame;

import java.io.IOException;
import java.util.List;

public class TransportFrameDecoder extends ByteToMessageDecoder {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
        while (true) {
            if (buffer.readableBytes() < TransportFrame.HEADER_SIZE) {
                break;
            }
            int readerIndex = buffer.readerIndex();
            if (buffer.getByte(readerIndex) != 'E' && buffer.getByte(readerIndex + 1) != 'Q') {
                throw new IOException("Invalid header");
            }

            final Version version = Version.fromId(buffer.getInt(readerIndex + 2));
            final long request = buffer.getLong(readerIndex + 2 + 4);
            final int size = buffer.getInt(readerIndex + 2 + 4 + 8);

            final ByteBuf content;
            if (size <= 0) {
                // ping request
                content = null;
                buffer.skipBytes(TransportFrame.HEADER_SIZE);
            } else {
                // standard request<commit_msg>fix NPE in transport frame decoder<commit_after>package org.mitallast.queue.transport.netty.codec;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import org.mitallast.queue.Version;
import org.mitallast.queue.transport.TransportFrame;

import java.io.IOException;
import java.util.List;

public class TransportFrameDecoder extends ByteToMessageDecoder {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
        while (true) {
            if (buffer.readableBytes() < TransportFrame.HEADER_SIZE) {
                break;
            }
            int readerIndex = buffer.readerIndex();
            if (buffer.getByte(readerIndex) != 'E' && buffer.getByte(readerIndex + 1) != 'Q') {
                throw new IOException("Invalid header");
            }

            final Version version = Version.fromId(buffer.getInt(readerIndex + 2));
            final long request = buffer.getLong(readerIndex + 2 + 4);
            final int size = buffer.getInt(readerIndex + 2 + 4 + 8);

            final ByteBuf content;
            if (size <= 0) {
                // ping request
                content = Unpooled.EMPTY_BUFFER;
                buffer.skipBytes(TransportFrame.HEADER_SIZE);
            } else {
                // standard request
{'fraction_non_alphanumeric': 0.09709781440343963, 'fraction_numerical': 0.00573271228950197, 'mean_word_length': 2.4215686274509802, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 5, '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': '4048431', 'n_tokens_mistral': 757, 'n_tokens_neox': 722, 'n_words': 172}