<commit_before>/*
 * Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and limitations under the License.
 */

package org.wso2.carbon.transport.http.netty.sender.channel.pool;


import com.lmax.disruptor.RingBuffer;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.EventLoopGroup;
import io.netty.handler.codec.http.HttpRequest;
import org.apache.commons.pool.impl.GenericObjectPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.wso2.carbon.messaging.CarbonCallback;
import org.wso2.carbon.messaging.CarbonMessage;
import org.wso2.carbon.transport.http.netty.common.HttpRoute;
import org.wso2.carbon.transport.http.netty.listener.SourceHandler;
import org.wso2.carbon.transport.http.netty.sender.ClientRequestWorker;
import org.wso2.carbon.transport.http.netty.sender.NettyClientInitializer;
import org.wso2.carbon.transport.http.netty.sender.channel.TargetChannel;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * A class which handles connection pool management.
 */
public class ConnectionManager {

    private static final Logger log = LoggerFactory.getLogger(ConnectionManager.class);

    private static volatile ConnectionManager connectionManager;

    private PoolConfiguration poolConfiguration;

    private int poolCount;

    private final List<Map<String, GenericObjectPool>> mapList;

    private PoolManagementPolicy poolManagementPolicy;

    private AtomicInteger index = new AtomicInteger(1);

    private ExecutorService executorService;


    private ConnectionManager(PoolConfiguration poolConfiguration) {
        this.poolConfiguration = poolConfiguration;
        this.poolCount = poolConfiguration.getNumberOfPools();
        this.executorService = Executors.newFixedThreadPool(poolConfiguration.getExecutorServiceThreads());
        if (poolConfiguration.getNumberOfPools() == 0) {
            this.poolManagementPolicy = PoolManagementPolicy.PER_SERVER_CHANNEL_ENDPOINT_CONNECTION_CACHING;
        } else {
            this.poolManagementPolicy = PoolManagementPolicy.GLOBAL_ENDPOINT_CONNECTION_CACHING;
        }

        mapList = new ArrayList<>();
        for (int i = 0; i < poolCount; i++) {
            Map<String, GenericObjectPool> map = new ConcurrentHashMap<>();
            mapList.add(map);
        }

    }


    private GenericObjectPool createPoolForRoute(HttpRoute httpRoute, EventLoopGroup eventLoopGroup,
                                                 Class eventLoopClass, NettyClientInitializer nettyClientInitializer) {
        GenericObjectPool.Config config = new GenericObjectPool.Config();
        config.maxActive = poolConfiguration.getMaxActivePerPool();
        config.maxIdle = poolConfiguration.getMaxIdlePerPool();
        config.minIdle = poolConfiguration.getMinIdlePerPool();
        config.testOnBorrow = poolConfiguration.isTestOnBorrow();
        config.testWhileIdle = poolConfiguration.isTestWhileIdle();
        config.timeBetweenEvictionRunsMillis = poolConfiguration.getTimeBetweenEvictionRuns();
        config.minEvictableIdleTimeMillis = poolConfiguration.getMinEvictableIdleTime();
        config.whenExhaustedAction = poolConfiguration.getExhaustedAction();
        return new GenericObjectPool(new PoolableTargetChannelFactory(httpRoute, eventLoopGroup,
                                                                      eventLoopClass, nettyClientInitializer),
                                     config);


    }


    public static ConnectionManager getInstance() {
        if (connectionManager == null) {
            synchronized (ConnectionManager.class) {
                if (connectionManager == null) {
                    PoolConfiguration poolConfiguration = PoolConfiguration.getInstance();
                    connectionManager = new ConnectionManager(poolConfiguration);
                }
            }

        }
        return connectionManager;
    }

    /**
     * Provide target channel for given http route.
     *
     * @param httpRoute BE address
     * @param sourceHandler Incoming channel
     * @param nettyClientInitializer netty initializer
     * @param httpRequest http request
     * @param carbonMessage carbon message
     * @param carbonCallback carbon call back
     * @param ringBuffer ring buffer
     * @return TargetChannel
     * @throws Exception to notify any errors occur during retrieving the target channel
     */
    public TargetChannel getTargetChannel(HttpRoute httpRoute, SourceHandler sourceHandler,
                                          NettyClientInitializer nettyClientInitializer,
                                          HttpRequest httpRequest, CarbonMessage carbonMessage,
                                          CarbonCallback carbonCallback, RingBuffer ringBuffer)
               throws Exception {
        Channel channel = null;
        TargetChannel targetChannel = null;
        ChannelHandlerContext ctx = sourceHandler.getInboundChannelContext();
        EventLoopGroup group = ctx.channel().eventLoop();
        Class cl = ctx.channel().getClass();

        // Take connections from Global connection pool
        if (poolManagementPolicy == PoolManagementPolicy.GLOBAL_ENDPOINT_CONNECTION_CACHING) {
            Map<String, GenericObjectPool> objectPoolMap = sourceHandler.getTargetChannelPool();
            GenericObjectPool pool = objectPoolMap.get(httpRoute.toString());
            if (pool == null) {
                pool = createPoolForRoute(httpRoute, group, cl, nettyClientInitializer);
                objectPoolMap.put(httpRoute.toString(), pool);
            }
            try {
                executorService.submit(new ClientRequestWorker(httpRoute, sourceHandler, nettyClientInitializer,
                                                               httpRequest, carbonMessage,
                                                               carbonCallback, true,
                                                               pool, this, ringBuffer));
            } catch (Exception e) {
                String msg = "Cannot borrow free channel from pool ";
                log.error(msg, e);
                throw new Exception(msg, e);
            }
        } else {
            // manage connections according to per inbound channel caching method
            if (!isRouteExists(httpRoute, sourceHandler)) {
               executorService.
                        execute(new ClientRequestWorker(httpRoute, sourceHandler, nettyClientInitializer,
                                                        httpRequest, carbonMessage,
                                                        carbonCallback, false,
                                                        null, this, ringBuffer));
            } else {
                targetChannel = sourceHandler.getChannel(httpRoute);
                Channel tempc = targetChannel.getChannel();
                if (!tempc.isActive()) {
                    executorService.
                               execute(new ClientRequestWorker(httpRoute, sourceHandler, nettyClientInitializer,
                                                               httpRequest, carbonMessage,
                                                               carbonCallback, false, null, this, ringBuffer));
                    targetChannel=null;
                }<commit_msg>style fixed<commit_after>/*
 * Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and limitations under the License.
 */

package org.wso2.carbon.transport.http.netty.sender.channel.pool;


import com.lmax.disruptor.RingBuffer;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.EventLoopGroup;
import io.netty.handler.codec.http.HttpRequest;
import org.apache.commons.pool.impl.GenericObjectPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.wso2.carbon.messaging.CarbonCallback;
import org.wso2.carbon.messaging.CarbonMessage;
import org.wso2.carbon.transport.http.netty.common.HttpRoute;
import org.wso2.carbon.transport.http.netty.listener.SourceHandler;
import org.wso2.carbon.transport.http.netty.sender.ClientRequestWorker;
import org.wso2.carbon.transport.http.netty.sender.NettyClientInitializer;
import org.wso2.carbon.transport.http.netty.sender.channel.TargetChannel;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * A class which handles connection pool management.
 */
public class ConnectionManager {

    private static final Logger log = LoggerFactory.getLogger(ConnectionManager.class);

    private static volatile ConnectionManager connectionManager;

    private PoolConfiguration poolConfiguration;

    private int poolCount;

    private final List<Map<String, GenericObjectPool>> mapList;

    private PoolManagementPolicy poolManagementPolicy;

    private AtomicInteger index = new AtomicInteger(1);

    private ExecutorService executorService;


    private ConnectionManager(PoolConfiguration poolConfiguration) {
        this.poolConfiguration = poolConfiguration;
        this.poolCount = poolConfiguration.getNumberOfPools();
        this.executorService = Executors.newFixedThreadPool(poolConfiguration.getExecutorServiceThreads());
        if (poolConfiguration.getNumberOfPools() == 0) {
            this.poolManagementPolicy = PoolManagementPolicy.PER_SERVER_CHANNEL_ENDPOINT_CONNECTION_CACHING;
        } else {
            this.poolManagementPolicy = PoolManagementPolicy.GLOBAL_ENDPOINT_CONNECTION_CACHING;
        }

        mapList = new ArrayList<>();
        for (int i = 0; i < poolCount; i++) {
            Map<String, GenericObjectPool> map = new ConcurrentHashMap<>();
            mapList.add(map);
        }

    }


    private GenericObjectPool createPoolForRoute(HttpRoute httpRoute, EventLoopGroup eventLoopGroup,
                                                 Class eventLoopClass, NettyClientInitializer nettyClientInitializer) {
        GenericObjectPool.Config config = new GenericObjectPool.Config();
        config.maxActive = poolConfiguration.getMaxActivePerPool();
        config.maxIdle = poolConfiguration.getMaxIdlePerPool();
        config.minIdle = poolConfiguration.getMinIdlePerPool();
        config.testOnBorrow = poolConfiguration.isTestOnBorrow();
        config.testWhileIdle = poolConfiguration.isTestWhileIdle();
        config.timeBetweenEvictionRunsMillis = poolConfiguration.getTimeBetweenEvictionRuns();
        config.minEvictableIdleTimeMillis = poolConfiguration.getMinEvictableIdleTime();
        config.whenExhaustedAction = poolConfiguration.getExhaustedAction();
        return new GenericObjectPool(new PoolableTargetChannelFactory(httpRoute, eventLoopGroup,
                                                                      eventLoopClass, nettyClientInitializer),
                                     config);


    }


    public static ConnectionManager getInstance() {
        if (connectionManager == null) {
            synchronized (ConnectionManager.class) {
                if (connectionManager == null) {
                    PoolConfiguration poolConfiguration = PoolConfiguration.getInstance();
                    connectionManager = new ConnectionManager(poolConfiguration);
                }
            }

        }
        return connectionManager;
    }

    /**
     * Provide target channel for given http route.
     *
     * @param httpRoute BE address
     * @param sourceHandler Incoming channel
     * @param nettyClientInitializer netty initializer
     * @param httpRequest http request
     * @param carbonMessage carbon message
     * @param carbonCallback carbon call back
     * @param ringBuffer ring buffer
     * @return TargetChannel
     * @throws Exception to notify any errors occur during retrieving the target channel
     */
    public TargetChannel getTargetChannel(HttpRoute httpRoute, SourceHandler sourceHandler,
                                          NettyClientInitializer nettyClientInitializer,
                                          HttpRequest httpRequest, CarbonMessage carbonMessage,
                                          CarbonCallback carbonCallback, RingBuffer ringBuffer)
               throws Exception {
        Channel channel = null;
        TargetChannel targetChannel = null;
        ChannelHandlerContext ctx = sourceHandler.getInboundChannelContext();
        EventLoopGroup group = ctx.channel().eventLoop();
        Class cl = ctx.channel().getClass();

        // Take connections from Global connection pool
        if (poolManagementPolicy == PoolManagementPolicy.GLOBAL_ENDPOINT_CONNECTION_CACHING) {
            Map<String, GenericObjectPool> objectPoolMap = sourceHandler.getTargetChannelPool();
            GenericObjectPool pool = objectPoolMap.get(httpRoute.toString());
            if (pool == null) {
                pool = createPoolForRoute(httpRoute, group, cl, nettyClientInitializer);
                objectPoolMap.put(httpRoute.toString(), pool);
            }
            try {
                executorService.submit(new ClientRequestWorker(httpRoute, sourceHandler, nettyClientInitializer,
                                                               httpRequest, carbonMessage,
                                                               carbonCallback, true,
                                                               pool, this, ringBuffer));
            } catch (Exception e) {
                String msg = "Cannot borrow free channel from pool ";
                log.error(msg, e);
                throw new Exception(msg, e);
            }
        } else {
            // manage connections according to per inbound channel caching method
            if (!isRouteExists(httpRoute, sourceHandler)) {
               executorService.
                        execute(new ClientRequestWorker(httpRoute, sourceHandler, nettyClientInitializer,
                                                        httpRequest, carbonMessage,
                                                        carbonCallback, false,
                                                        null, this, ringBuffer));
            } else {
                targetChannel = sourceHandler.getChannel(httpRoute);
                Channel tempc = targetChannel.getChannel();
                if (!tempc.isActive()) {
                    executorService.
                               execute(new ClientRequestWorker(httpRoute, sourceHandler, nettyClientInitializer,
                                                               httpRequest, carbonMessage,
                                                               carbonCallback, false, null, this, ringBuffer));
                    targetChannel = null;
                }
{'fraction_non_alphanumeric': 0.07106786550067742, 'fraction_numerical': 0.0028328611898017, 'mean_word_length': 2.3073319755600816, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 4, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 56, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7666390', 'n_tokens_mistral': 3726, 'n_tokens_neox': 3610, 'n_words': 908}