<commit_before>package edu.uw.zookeeper.net.intravm;

import static com.google.common.base.Preconditions.checkState;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;

import java.io.IOException;
import java.net.SocketAddress;
import java.util.Queue;
import java.util.concurrent.Executor;

import net.engio.mbassy.PubSubSupport;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.google.common.base.Optional;
import com.google.common.util.concurrent.ListenableFuture;

import edu.uw.zookeeper.common.LoggingPromise;
import edu.uw.zookeeper.common.Pair;
import edu.uw.zookeeper.common.Promise;
import edu.uw.zookeeper.common.PromiseTask;
import edu.uw.zookeeper.common.SettableFuturePromise;
import edu.uw.zookeeper.net.Connection;
import edu.uw.zookeeper.net.LoggingMarker;
import edu.uw.zookeeper.protocol.Codec;

public class IntraVmCodecEndpoint<I, T extends Codec<? super I, ? extends Optional<?>>> extends IntraVmEndpoint<ByteBuf> {

    public static <I, T extends Codec<? super I, ? extends Optional<?>>> IntraVmCodecEndpoint<I,T> create(
            ByteBufAllocator allocator,
            Pair<Class<I>, T> codec,
            SocketAddress address,
            PubSubSupport<Object> publisher,
            Executor executor) {
        return IntraVmCodecEndpoint.<I,T>builder(
                allocator,
                address, 
                publisher,
                executor).setCodec(codec).build();
    }

    public static <I, T extends Codec<? super I, ? extends Optional<?>>> Builder<I,T> builder(
            ByteBufAllocator allocator,
            SocketAddress address,
            PubSubSupport<Object> publisher,
            Executor executor) {
        return new Builder<I,T>(allocator, address, publisher, executor);
    }
    
    public static class Builder<I, T extends Codec<? super I, ? extends Optional<?>>> extends IntraVmEndpoint.Builder<ByteBuf> {

        protected final ByteBufAllocator allocator;
        protected Pair<Class<I>, ? extends T> codec;
        
        public Builder(
                ByteBufAllocator allocator,
                SocketAddress address,
                PubSubSupport<Object> publisher,
                Executor executor) {
            this(allocator, address, publisher, executor, LogManager.getLogger(IntraVmCodecEndpoint.class));
        }
        
        public Builder(
                ByteBufAllocator allocator,
                SocketAddress address,
                PubSubSupport<Object> publisher,
                Executor executor,
                Logger logger) {
            super(address, publisher, executor, logger);
            this.allocator = allocator;
            this.codec = null;
        }
        
        public Builder<I,T> setCodec(Pair<Class<I>, ? extends T> codec) {
            this.codec = codec;
            return this;
        }
        
        @Override
        public IntraVmCodecEndpoint<I,T> build() {
            checkState(codec != null);
            
            return new IntraVmCodecEndpoint<I,T>(
                    allocator, codec, address, logger, executor, publisher, mailbox, stopped);
        }
    }
    
    protected final ByteBufAllocator allocator;
    protected final Class<I> type;
    protected final T codec;
    
    protected IntraVmCodecEndpoint(
            ByteBufAllocator allocator,
            Pair<Class<I>, ? extends T> codec,
            SocketAddress address,
            Logger logger,
            Executor executor,
            IntraVmPublisher publisher,
            Queue<Optional<? extends ByteBuf>> mailbox,
            Promise<IntraVmEndpoint<ByteBuf>> stopped) {
        super(address, logger, executor, publisher, mailbox, stopped);
        this.allocator = allocator;
        this.type = codec.first();
        this.codec = codec.second();
    }
    
    public Pair<Class<I>, T> getCodec() {
        return Pair.create(type, codec);
    }
    
    @Override
    public <U> ListenableFuture<U> write(Optional<U> message, IntraVmEndpoint<?> remote) {
        @SuppressWarnings("unchecked")
        EncodingSendTask<U> task = new EncodingSendTask<U>((IntraVmEndpoint<? super ByteBuf>) remote, message, LoggingPromise.create(logger, SettableFuturePromise.<U>create()));
        execute(task);
        return task;
    }

    @Override
    protected void doPublish(ByteBuf input) {
        Optional<?> output;
        try { 
            output = codec.decode(input);
        } catch (IOException e) {
            logger.warn(LoggingMarker.NET_MARKER.get(), "{}", input, e);
            stop();
            return;
        }
        if (output.isPresent()) {
            Object value = output.get();
            logger.trace(LoggingMarker.NET_MARKER.get(), "Decoded: {}", value);
            publisher.publish(value);
        }
    }

    protected class EncodingSendTask<U> extends PromiseTask<Optional<U>, U> implements Runnable {

        protected final IntraVmEndpoint<? super ByteBuf> remote;
        
        public EncodingSendTask(
                IntraVmEndpoint<? super ByteBuf> remote,
                Optional<U> task, 
                Promise<U> promise) {
            super(task, promise);
            this.remote = remote;
        }
        
        @Override
        public synchronized void run() {
            if(!isDone()) {
                ByteBuf output;
                if (task().isPresent()) {
                    @SuppressWarnings("unchecked")
                    I input = (I) task().get();
                    output = allocator.buffer();
                    logger.trace(LoggingMarker.NET_MARKER.get(), "Encoding: {}", input);
                    try {
                        codec.encode(input, output);
                    } catch (IOException e) {
                        logger.warn(LoggingMarker.NET_MARKER.get(), "{}", e);
                        setException(e);
                        return;
                    }
                } else {
                    output = null;
                }
                if (remote.send(Optional.fromNullable(output))) {
                    set(task().orNull());
                } else {
                    setException(new IllegalStateException(Connection.State.CONNECTION_CLOSING.toString()));
                }
            }
        }
    }
}<commit_msg>minor<commit_after>package edu.uw.zookeeper.net.intravm;

import static com.google.common.base.Preconditions.checkState;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;

import java.io.IOException;
import java.net.SocketAddress;
import java.util.Queue;
import java.util.concurrent.Executor;

import net.engio.mbassy.PubSubSupport;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.google.common.base.Optional;
import com.google.common.util.concurrent.ListenableFuture;

import edu.uw.zookeeper.common.LoggingPromise;
import edu.uw.zookeeper.common.Pair;
import edu.uw.zookeeper.common.Promise;
import edu.uw.zookeeper.common.PromiseTask;
import edu.uw.zookeeper.common.SettableFuturePromise;
import edu.uw.zookeeper.net.Connection;
import edu.uw.zookeeper.net.LoggingMarker;
import edu.uw.zookeeper.protocol.Codec;

public class IntraVmCodecEndpoint<I, T extends Codec<? super I, ? extends Optional<?>>> extends IntraVmEndpoint<ByteBuf> {

    public static <I, T extends Codec<? super I, ? extends Optional<?>>> IntraVmCodecEndpoint<I,T> create(
            ByteBufAllocator allocator,
            Pair<Class<I>, T> codec,
            SocketAddress address,
            PubSubSupport<Object> publisher,
            Executor executor) {
        return IntraVmCodecEndpoint.<I,T>builder(
                allocator,
                address, 
                publisher,
                executor).setCodec(codec).build();
    }

    public static <I, T extends Codec<? super I, ? extends Optional<?>>> Builder<I,T> builder(
            ByteBufAllocator allocator,
            SocketAddress address,
            PubSubSupport<Object> publisher,
            Executor executor) {
        return new Builder<I,T>(allocator, address, publisher, executor);
    }
    
    public static class Builder<I, T extends Codec<? super I, ? extends Optional<?>>> extends IntraVmEndpoint.Builder<ByteBuf> {

        protected final ByteBufAllocator allocator;
        protected Pair<Class<I>, ? extends T> codec;
        
        public Builder(
                ByteBufAllocator allocator,
                SocketAddress address,
                PubSubSupport<Object> publisher,
                Executor executor) {
            this(allocator, address, publisher, executor, LogManager.getLogger(IntraVmCodecEndpoint.class));
        }
        
        public Builder(
                ByteBufAllocator allocator,
                SocketAddress address,
                PubSubSupport<Object> publisher,
                Executor executor,
                Logger logger) {
            super(address, publisher, executor, logger);
            this.allocator = allocator;
            this.codec = null;
        }
        
        public Builder<I,T> setCodec(Pair<Class<I>, ? extends T> codec) {
            this.codec = codec;
            return this;
        }
        
        @Override
        public IntraVmCodecEndpoint<I,T> build() {
            checkState(codec != null);
            
            return new IntraVmCodecEndpoint<I,T>(
                    allocator, codec, address, logger, executor, publisher, mailbox, stopped);
        }
    }
    
    protected final ByteBufAllocator allocator;
    protected final Class<I> type;
    protected final T codec;
    
    protected IntraVmCodecEndpoint(
            ByteBufAllocator allocator,
            Pair<Class<I>, ? extends T> codec,
            SocketAddress address,
            Logger logger,
            Executor executor,
            IntraVmPublisher publisher,
            Queue<Optional<? extends ByteBuf>> mailbox,
            Promise<IntraVmEndpoint<ByteBuf>> stopped) {
        super(address, logger, executor, publisher, mailbox, stopped);
        this.allocator = allocator;
        this.type = codec.first();
        this.codec = codec.second();
    }
    
    public Pair<Class<I>, T> getCodec() {
        return Pair.create(type, codec);
    }
    
    @Override
    public <U> ListenableFuture<U> write(Optional<U> message, IntraVmEndpoint<?> remote) {
        @SuppressWarnings("unchecked")
        EncodingSendTask<U> task = new EncodingSendTask<U>((IntraVmEndpoint<? super ByteBuf>) remote, message, LoggingPromise.create(logger, SettableFuturePromise.<U>create()));
        execute(task);
        return task;
    }

    @Override
    protected void doPublish(ByteBuf input) {
        Optional<?> output;
        try { 
            output = codec.decode(input);
        } catch (IOException e) {
            logger.warn(LoggingMarker.NET_MARKER.get(), "{}", input, e);
            stop();
            return;
        }
        if (output.isPresent()) {
            Object value = output.get();
            logger.trace(LoggingMarker.NET_MARKER.get(), "Decoded: {}", value);
            publisher.publish(value);
        }
    }

    protected class EncodingSendTask<U> extends PromiseTask<Optional<U>, U> implements Runnable {

        protected final IntraVmEndpoint<? super ByteBuf> remote;
        
        public EncodingSendTask(
                IntraVmEndpoint<? super ByteBuf> remote,
                Optional<U> task, 
                Promise<U> promise) {
            super(task, promise);
            this.remote = remote;
        }
        
        @Override
        public synchronized void run() {
            if(!isDone()) {
                ByteBuf output;
                if (task().isPresent()) {
                    @SuppressWarnings("unchecked")
                    I input = (I) task().get();
                    output = allocator.buffer();
                    logger.trace(LoggingMarker.NET_MARKER.get(), "Encoding: {} ({})", input, IntraVmCodecEndpoint.this);
                    try {
                        codec.encode(input, output);
                    } catch (IOException e) {
                        logger.warn(LoggingMarker.NET_MARKER.get(), "{}", IntraVmCodecEndpoint.this, e);
                        setException(e);
                        return;
                    }
                } else {
                    output = null;
                }
                if (remote.send(Optional.fromNullable(output))) {
                    set(task().orNull());
                } else {
                    setException(new IllegalStateException(Connection.State.CONNECTION_CLOSING.toString()));
                }
            }
        }
    }
}
{'fraction_non_alphanumeric': 0.10079159808762442, 'fraction_numerical': 0.00031350419311858296, 'mean_word_length': 1.9813084112149533, 'pattern_counts': {'":': 0, '<': 115, '<?xml version=': 0, '>': 115, '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': '3438945', 'n_tokens_mistral': 3497, 'n_tokens_neox': 3359, 'n_words': 728}