<commit_before>package org.nohope.protobuf.core.net;

import com.google.protobuf.ExtensionRegistry;
import com.google.protobuf.Message;
import com.google.protobuf.MessageLite;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.ChannelUpstreamHandler;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.handler.codec.frame.LengthFieldBasedFrameDecoder;
import org.jboss.netty.handler.codec.frame.LengthFieldPrepender;
import org.jboss.netty.handler.codec.protobuf.ProtobufDecoder;
import org.jboss.netty.handler.codec.protobuf.ProtobufEncoder;

import static org.nohope.protobuf.core.MessageUtils.getExtensionRegistry;

/**
 * Set of decoders/encoders to convert packet format from/to
 * {@link org.nohope.rpc.protocol.RPC.RpcRequest RpcRequest} or
 * {@link org.nohope.rpc.protocol.RPC.RpcResponse RpcResponse}.
 * <p/>
 * <b>Packet format</b>:
 * <pre>
 * +----------------------+----------------+
 * | Serialized RpcObject |   Serialized   |
 * |   Length (4 bytes)   | RpcObject body |
 * +----------------------+----------------+
 * </pre>
 *
 * @author <a href="mailto:<EMAIL>"><NAME></a>
 * @since 8/19/13 6:16 PM
 */
public final class PipelineFactory implements ChannelPipelineFactory {
    private static final int MAX_FRAME_BYTES_LENGTH = Integer.MAX_VALUE;
    private static final int HEADER_BYTES = 4;

    private final ChannelUpstreamHandler handler;
    private final MessageLite prototype;

    public PipelineFactory(final ChannelUpstreamHandler handler, final MessageLite prototype) {
        this.handler = handler;
        this.prototype = prototype;
    }

    @Override
    public ChannelPipeline getPipeline() throws Exception {
        final ChannelPipeline p = Channels.pipeline();
        p.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(MAX_FRAME_BYTES_LENGTH, 0, HEADER_BYTES, 0, HEADER_BYTES));

        final ExtensionRegistry extensionRegistry;
        if (prototype instanceof Message) {
            extensionRegistry = getExtensionRegistry(((Message) prototype).getDescriptorForType().getFile());
        } else {
            extensionRegistry = null;
        }

        p.addLast("protobufDecoder", new ProtobufDecoder(prototype, extensionRegistry));

        p.addLast("frameEncoder", new LengthFieldPrepender(HEADER_BYTES));
        p.addLast("protobufEncoder", new ProtobufEncoder());
        p.addLast("handler", handler);
        return p;
    }
}
<commit_msg>protobuf-rpc: OrderedMemoryAwareThreadPoolExecutor to avoid waiting in IO threads<commit_after>package org.nohope.protobuf.core.net;

import com.google.protobuf.ExtensionRegistry;
import com.google.protobuf.Message;
import com.google.protobuf.MessageLite;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.ChannelUpstreamHandler;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.handler.codec.frame.LengthFieldBasedFrameDecoder;
import org.jboss.netty.handler.codec.frame.LengthFieldPrepender;
import org.jboss.netty.handler.codec.protobuf.ProtobufDecoder;
import org.jboss.netty.handler.codec.protobuf.ProtobufEncoder;
import org.jboss.netty.handler.execution.ExecutionHandler;
import org.jboss.netty.handler.execution.OrderedMemoryAwareThreadPoolExecutor;

import java.util.concurrent.Executor;

import static java.util.concurrent.Executors.defaultThreadFactory;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.nohope.protobuf.core.MessageUtils.getExtensionRegistry;

/**
 * Set of decoders/encoders to convert packet format from/to
 * {@link org.nohope.rpc.protocol.RPC.RpcRequest RpcRequest} or
 * {@link org.nohope.rpc.protocol.RPC.RpcResponse RpcResponse}.
 * <p/>
 * <b>Packet format</b>:
 * <pre>
 * +----------------------+----------------+
 * | Serialized RpcObject |   Serialized   |
 * |   Length (4 bytes)   | RpcObject body |
 * +----------------------+----------------+
 * </pre>
 *
 * @author <a href="mailto:<EMAIL>"><NAME></a>
 * @since 8/19/13 6:16 PM
 */
public final class PipelineFactory implements ChannelPipelineFactory {
    private static final int MAX_FRAME_BYTES_LENGTH = Integer.MAX_VALUE;
    private static final int HEADER_BYTES = 4;
    private static final long MEMORY_SIZE = 1048576L;

    private final ChannelUpstreamHandler handler;
    private final MessageLite prototype;

    public PipelineFactory(final ChannelUpstreamHandler handler, final MessageLite prototype) {
        this.handler = handler;
        this.prototype = prototype;
    }

    @Override
    public ChannelPipeline getPipeline() throws Exception {
        final ChannelPipeline p = Channels.pipeline();
        p.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(MAX_FRAME_BYTES_LENGTH, 0, HEADER_BYTES, 0, HEADER_BYTES));

        final Executor eventExecutor =
                new OrderedMemoryAwareThreadPoolExecutor(16, MEMORY_SIZE, MEMORY_SIZE, 1000, MILLISECONDS, defaultThreadFactory());
        final ExecutionHandler executionHandler = new ExecutionHandler(eventExecutor);

        final ExtensionRegistry extensionRegistry;
        if (prototype instanceof Message) {
            extensionRegistry = getExtensionRegistry(((Message) prototype).getDescriptorForType().getFile());
        } else {
            extensionRegistry = null;
        }

        p.addLast("protobufDecoder", new ProtobufDecoder(prototype, extensionRegistry));

        p.addLast("frameEncoder", new LengthFieldPrepender(HEADER_BYTES));
        p.addLast("protobufEncoder", new ProtobufEncoder());
        p.addLast("executionHandler", executionHandler);
        p.addLast("handler", handler);
        return p;
    }
}

{'fraction_non_alphanumeric': 0.1258692628650904, 'fraction_numerical': 0.006432545201668985, 'mean_word_length': 3.7663628831814417, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 21, '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': '6444092', 'n_tokens_mistral': 1661, 'n_tokens_neox': 1596, 'n_words': 328}