<commit_before>package io.reactivex.netty.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.reactivex.netty.ChannelCloseListener;
import io.reactivex.netty.RxNetty;
import io.reactivex.netty.channel.ConnectionHandler;
import io.reactivex.netty.channel.ObservableConnection;
import io.reactivex.netty.pipeline.PipelineConfigurator;
import io.reactivex.netty.pipeline.PipelineConfiguratorComposite;
import io.reactivex.netty.pipeline.PipelineConfigurators;
import io.reactivex.netty.server.RxServer;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import rx.Observable;
import rx.Subscriber;

import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author <NAME>
 */
public class ConnectionPoolTest {

    public static final int MAX_IDLE_TIME_MILLIS = 1000;
    private ConnectionPoolImpl<String, String> pool;
    private RxClient.ServerInfo serverInfo;
    private Bootstrap clientBootstrap;
    private TrackableStateChangeListener stateChangeListener;
    private MaxConnectionsBasedStrategy strategy;
    private RxServer<String,String> server;
    private final ChannelCloseListener channelCloseListener = new ChannelCloseListener();
    private PoolStats stats;
    private ConnectionHandlerImpl serverConnHandler;
    private PipelineConfigurator<String,String> pipelineConfigurator;

    @Before
    public void setUp() throws Exception {
        serverInfo = new RxClient.ServerInfo("localhost", 9999);
        serverConnHandler = new ConnectionHandlerImpl();
        server = RxNetty.createTcpServer(9999, PipelineConfigurators.textOnlyConfigurator(),
                                         serverConnHandler).start();
        stateChangeListener = new TrackableStateChangeListener();
        strategy = new MaxConnectionsBasedStrategy(1);
        clientBootstrap = new Bootstrap().group(new NioEventLoopGroup(4))
                                         .channel(NioSocketChannel.class);
        pipelineConfigurator = new PipelineConfiguratorComposite<String, String>(PipelineConfigurators.textOnlyConfigurator(), new PipelineConfigurator() {
            @Override
            public void configureNewPipeline(ChannelPipeline pipeline) {
                channelCloseListener.reset();
                pipeline.addFirst(channelCloseListener);
            }
        });
        pool = new ConnectionPoolImpl<String, String>(new PoolConfig(MAX_IDLE_TIME_MILLIS), strategy, null);
        pool.setChannelFactory(new ClientChannelFactoryImpl<String, String>(clientBootstrap, pool, serverInfo));
        pool.poolStateChangeObservable().subscribe(stateChangeListener);
        stats = pool.getStats();
    }

    @After
    public void tearDown() throws Exception {
        if (null != pool) {
            pool.shutdown();
        }
        if (null != clientBootstrap) {
            clientBootstrap.group().shutdownGracefully();
        }
        if (null != server) {
            server.shutdown();
            server.waitTillShutdown();
        }
    }

    @Test
    public void testAcquireRelease() throws Exception {
        ObservableConnection<String, String> conn = acquireAndTestStats();
        conn.close();
        waitForClose();
        assertAllConnectionsReturned();
    }

    @Test
    public void testReleaseAfterClose() throws Exception {
        ObservableConnection<String, String> conn = acquireAndTestStats();
        waitForClose();
        conn.close();
        assertAllConnectionsReturned();
    }

    @Test
    public void testReuse() throws Exception {
        serverConnHandler.closeNewConnectionsOnReceive(false);
        ObservableConnection<String, String> connection = acquireAndTestStats();

        connection.close();

        ObservableConnection<String, String> reusedConn = acquireAndTestStats();

        Assert.assertEquals("Connection reuse callback not received.", 1, stateChangeListener.getReuseCount());
        Assert.assertEquals("Connection not reused.", connection, reusedConn);

        serverConnHandler.closeAllClientConnections();

        connection.close();
        waitForClose();<commit_msg>Removing redundant close of client connection in test<commit_after>package io.reactivex.netty.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.reactivex.netty.ChannelCloseListener;
import io.reactivex.netty.RxNetty;
import io.reactivex.netty.channel.ConnectionHandler;
import io.reactivex.netty.channel.ObservableConnection;
import io.reactivex.netty.pipeline.PipelineConfigurator;
import io.reactivex.netty.pipeline.PipelineConfiguratorComposite;
import io.reactivex.netty.pipeline.PipelineConfigurators;
import io.reactivex.netty.server.RxServer;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import rx.Observable;
import rx.Subscriber;

import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author <NAME>
 */
public class ConnectionPoolTest {

    public static final int MAX_IDLE_TIME_MILLIS = 1000;
    private ConnectionPoolImpl<String, String> pool;
    private RxClient.ServerInfo serverInfo;
    private Bootstrap clientBootstrap;
    private TrackableStateChangeListener stateChangeListener;
    private MaxConnectionsBasedStrategy strategy;
    private RxServer<String,String> server;
    private final ChannelCloseListener channelCloseListener = new ChannelCloseListener();
    private PoolStats stats;
    private ConnectionHandlerImpl serverConnHandler;
    private PipelineConfigurator<String,String> pipelineConfigurator;

    @Before
    public void setUp() throws Exception {
        serverInfo = new RxClient.ServerInfo("localhost", 9999);
        serverConnHandler = new ConnectionHandlerImpl();
        server = RxNetty.createTcpServer(9999, PipelineConfigurators.textOnlyConfigurator(),
                                         serverConnHandler).start();
        stateChangeListener = new TrackableStateChangeListener();
        strategy = new MaxConnectionsBasedStrategy(1);
        clientBootstrap = new Bootstrap().group(new NioEventLoopGroup(4))
                                         .channel(NioSocketChannel.class);
        pipelineConfigurator = new PipelineConfiguratorComposite<String, String>(PipelineConfigurators.textOnlyConfigurator(), new PipelineConfigurator() {
            @Override
            public void configureNewPipeline(ChannelPipeline pipeline) {
                channelCloseListener.reset();
                pipeline.addFirst(channelCloseListener);
            }
        });
        pool = new ConnectionPoolImpl<String, String>(new PoolConfig(MAX_IDLE_TIME_MILLIS), strategy, null);
        pool.setChannelFactory(new ClientChannelFactoryImpl<String, String>(clientBootstrap, pool, serverInfo));
        pool.poolStateChangeObservable().subscribe(stateChangeListener);
        stats = pool.getStats();
    }

    @After
    public void tearDown() throws Exception {
        if (null != pool) {
            pool.shutdown();
        }
        if (null != clientBootstrap) {
            clientBootstrap.group().shutdownGracefully();
        }
        if (null != server) {
            server.shutdown();
            server.waitTillShutdown();
        }
    }

    @Test
    public void testAcquireRelease() throws Exception {
        ObservableConnection<String, String> conn = acquireAndTestStats();
        conn.close();
        waitForClose();
        assertAllConnectionsReturned();
    }

    @Test
    public void testReleaseAfterClose() throws Exception {
        ObservableConnection<String, String> conn = acquireAndTestStats();
        waitForClose();
        conn.close();
        assertAllConnectionsReturned();
    }

    @Test
    public void testReuse() throws Exception {
        serverConnHandler.closeNewConnectionsOnReceive(false);
        ObservableConnection<String, String> connection = acquireAndTestStats();

        connection.close();

        ObservableConnection<String, String> reusedConn = acquireAndTestStats();

        Assert.assertEquals("Connection reuse callback not received.", 1, stateChangeListener.getReuseCount());
        Assert.assertEquals("Connection not reused.", connection, reusedConn);

        serverConnHandler.closeAllClientConnections();

        waitForClose();
{'fraction_non_alphanumeric': 0.08812085145342184, 'fraction_numerical': 0.003433279926756695, 'mean_word_length': 3.5610647181628394, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1695959', 'n_tokens_mistral': 2269, 'n_tokens_neox': 2204, 'n_words': 451}