/*
 * Copyright 2016-2018 shardingsphere.io.
 * <p>
 * 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.
 * </p>
 */

package io.shardingsphere.shardingproxy.backend.netty.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.epoll.EpollChannelOption;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.pool.AbstractChannelPoolMap;
import io.netty.channel.pool.ChannelPoolMap;
import io.netty.channel.pool.FixedChannelPool;
import io.netty.channel.pool.SimpleChannelPool;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.shardingsphere.core.metadata.datasource.DataSourceMetaData;
import io.shardingsphere.shardingproxy.runtime.GlobalRegistry;
import io.shardingsphere.shardingproxy.runtime.ShardingSchema;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * Backend connection client for netty.
 *
 * @author wangkai
 * @author linjiaqi
 */
@RequiredArgsConstructor
@Slf4j
public final class BackendNettyClient {
    
    private static final int WORKER_MAX_THREADS = Runtime.getRuntime().availableProcessors();
    
    private static final GlobalRegistry GLOBAL_REGISTRY = GlobalRegistry.getInstance();
    
    private final ShardingSchema shardingSchema;
    
    private final int maxConnections;
    
    private final int connectionTimeoutSeconds;
    
    private EventLoopGroup workerGroup;
    
    @Getter
    private ChannelPoolMap<String, SimpleChannelPool> poolMap;
    
    public BackendNettyClient(final ShardingSchema shardingSchema) {
        this.shardingSchema = shardingSchema;
        maxConnections = GLOBAL_REGISTRY.getBackendNIOConfig().getMaxConnections();
        connectionTimeoutSeconds = GLOBAL_REGISTRY.getBackendNIOConfig().getConnectionTimeoutSeconds();
    }
    
    /**
     * Start backend connection client for netty.
     *
     * @throws InterruptedException interrupted exception
     */
    public void start() throws InterruptedException {
        Bootstrap bootstrap = new Bootstrap();
        // TODO :jiaqi where to init workerGroup?
        if (workerGroup instanceof EpollEventLoopGroup) {
            groupsEpoll(bootstrap);
        } else {
            groupsNio(bootstrap);
        }
        initPoolMap(bootstrap);
    }
    
    /**
     * Stop backend connection client for netty.
     */
    public void stop() {
        if (null != workerGroup) {
            workerGroup.shutdownGracefully();
        }
    }
    
    private void groupsEpoll(final Bootstrap bootstrap) {
        workerGroup = new EpollEventLoopGroup(WORKER_MAX_THREADS);
        bootstrap.group(workerGroup)
                .channel(EpollSocketChannel.class)
                .option(EpollChannelOption.TCP_CORK, true)
                .option(EpollChannelOption.SO_KEEPALIVE, true)
                .option(EpollChannelOption.SO_BACKLOG, 128)
                .option(EpollChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
    }
    
    private void groupsNio(final Bootstrap bootstrap) {
        workerGroup = new NioEventLoopGroup(WORKER_MAX_THREADS);
        bootstrap.group(workerGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_BACKLOG, 128)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 100)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
    }
    
    private void initPoolMap(final Bootstrap bootstrap) throws InterruptedException {
        poolMap = new AbstractChannelPoolMap<String, SimpleChannelPool>() {
            
            @Override
            protected SimpleChannelPool newPool(final String dataSourceName) {
                DataSourceMetaData dataSourceMetaData = shardingSchema.getMetaData().getDataSource().getActualDataSourceMetaData(dataSourceName);
                return new FixedChannelPool(
                        bootstrap.remoteAddress(dataSourceMetaData.getHostName(), dataSourceMetaData.getPort()), 
                        new BackendNettyClientChannelPoolHandler(dataSourceName, shardingSchema.getName()), maxConnections);
            }
        };
        for (String each : shardingSchema.getDataSources().keySet()) {
            SimpleChannelPool pool = poolMap.get(each);
            Channel[] channels = new Channel[maxConnections];
            for (int i = 0; i < maxConnections; i++) {
                try {
                    channels[i] = pool.acquire().get(connectionTimeoutSeconds, TimeUnit.SECONDS);
                } catch (final ExecutionException | TimeoutException ex) {
                    log.error(ex.getMessage(), ex);
                }
            }
            for (int i = 0; i < maxConnections; i++) {
                pool.release(channels[i]);
            }
        }
    }
}

@@ -1449,16 +1449,76 @@
 gistry;%0A
+import io.shardingsphere.shardingproxy.runtime.LogicSchema;%0A
 import i
@@ -2178,39 +2178,33 @@
 e final 
-ShardingSchema sharding
+LogicSchema logic
 Schema;%0A
@@ -2480,24 +2480,21 @@
 gSchema 
-sharding
+logic
 Schema) 
@@ -2508,24 +2508,21 @@
    this.
-sharding
+logic
 Schema =
@@ -2518,32 +2518,29 @@
 gicSchema = 
-sharding
+logic
 Schema;%0A    
@@ -4768,24 +4768,21 @@
 aData = 
-sharding
+logic
 Schema.g
@@ -5095,24 +5095,21 @@
 ceName, 
-sharding
+logic
 Schema.g
@@ -5188,24 +5188,21 @@
  each : 
-sharding
+logic
 Schema.g

use LogicSchema