/*
 * 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;

import com.google.common.base.Optional;
import io.shardingsphere.core.constant.DatabaseType;
import io.shardingsphere.core.constant.SQLType;
import io.shardingsphere.core.parsing.SQLJudgeEngine;
import io.shardingsphere.core.parsing.parser.dialect.mysql.statement.SetStatement;
import io.shardingsphere.core.parsing.parser.dialect.mysql.statement.ShowColumnsStatement;
import io.shardingsphere.core.parsing.parser.dialect.mysql.statement.ShowDatabasesStatement;
import io.shardingsphere.core.parsing.parser.dialect.mysql.statement.ShowIndexStatement;
import io.shardingsphere.core.parsing.parser.dialect.mysql.statement.ShowOtherStatement;
import io.shardingsphere.core.parsing.parser.dialect.mysql.statement.ShowTableStatusStatement;
import io.shardingsphere.core.parsing.parser.dialect.mysql.statement.ShowTablesStatement;
import io.shardingsphere.core.parsing.parser.dialect.mysql.statement.UseStatement;
import io.shardingsphere.core.parsing.parser.sql.SQLStatement;
import io.shardingsphere.core.parsing.parser.token.SQLToken;
import io.shardingsphere.core.parsing.parser.token.SchemaToken;
import io.shardingsphere.shardingproxy.backend.jdbc.JDBCBackendHandler;
import io.shardingsphere.shardingproxy.backend.jdbc.connection.BackendConnection;
import io.shardingsphere.shardingproxy.backend.jdbc.execute.JDBCExecuteEngine;
import io.shardingsphere.shardingproxy.backend.jdbc.wrapper.PreparedStatementExecutorWrapper;
import io.shardingsphere.shardingproxy.backend.jdbc.wrapper.StatementExecutorWrapper;
import io.shardingsphere.shardingproxy.backend.netty.NettyBackendHandler;
import io.shardingsphere.shardingproxy.frontend.common.FrontendHandler;
import io.shardingsphere.shardingproxy.runtime.GlobalRegistry;
import io.shardingsphere.shardingproxy.runtime.ShardingSchema;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;

import java.util.LinkedList;
import java.util.List;

/**
 * Backend handler factory.
 *
 * @author zhangliang
 * @author panjuan
 */
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public final class BackendHandlerFactory {
    
    private static final GlobalRegistry GLOBAL_REGISTRY = GlobalRegistry.getInstance();
    
    /**
     * Create new instance of text protocol backend handler.
     *
     * @param connectionId connection ID of database connected
     * @param sequenceId sequence ID of SQL packet
     * @param sql SQL to be executed
     * @param backendConnection backend connection
     * @param databaseType database type
     * @param schema schema
     * @return instance of text protocol backend handler
     */
    public static BackendHandler newTextProtocolInstance(
            final int connectionId, final int sequenceId, final String sql, final BackendConnection backendConnection, final DatabaseType databaseType, final String schema) {
        ShardingSchema shardingSchema = GLOBAL_REGISTRY.getShardingSchema(schema);
        backendConnection.setShardingSchema(shardingSchema);
        return GLOBAL_REGISTRY.isUseNIO()
                ? new NettyBackendHandler(shardingSchema, connectionId, sequenceId, sql, databaseType)
                : new JDBCBackendHandler(shardingSchema, sql, new JDBCExecuteEngine(backendConnection, new StatementExecutorWrapper(shardingSchema)));
    }
    
    /**
     * Create new instance of text protocol backend handler.
     *
     * @param connectionId connection ID of database connected
     * @param sequenceId sequence ID of SQL packet
     * @param sql SQL to be executed
     * @param parameters SQL parameters
     * @param backendConnection backend connection
     * @param databaseType database type
     * @param schema schema
     * @return instance of text protocol backend handler
     */
    public static BackendHandler newBinaryProtocolInstance(final int connectionId, final int sequenceId, final String sql, final List<Object> parameters,
                                                           final BackendConnection backendConnection, final DatabaseType databaseType, final String schema) {
        ShardingSchema shardingSchema = GLOBAL_REGISTRY.getShardingSchema(schema);
        backendConnection.setShardingSchema(shardingSchema);
        return GLOBAL_REGISTRY.isUseNIO() ? new NettyBackendHandler(shardingSchema, connectionId, sequenceId, sql, databaseType)
                : new JDBCBackendHandler(shardingSchema, sql, new JDBCExecuteEngine(backendConnection, new PreparedStatementExecutorWrapper(shardingSchema, parameters)));
    }
    
    /**
     * Create new instance by sql judge.
     * 
     * @param connectionId connection ID of database connected
     * @param sequenceId sequence ID of SQL packet
     * @param sql SQL to be executed
     * @param backendConnection backend connection
     * @param databaseType database type
     * @param frontendHandler frontend handler
     * @return instance of backend handler
     */
    public static BackendHandler createBackendHandler(
            final int connectionId, final int sequenceId, final String sql, final BackendConnection backendConnection, final DatabaseType databaseType, final FrontendHandler frontendHandler) {
        SQLStatement sqlStatement = new SQLJudgeEngine(sql).judge();
        if (SQLType.DCL == sqlStatement.getType() || sqlStatement instanceof SetStatement) {
            return new SchemaBroadcastBackendHandler(connectionId, sequenceId, sql, databaseType);
        }
        
        if (sqlStatement instanceof UseStatement || sqlStatement instanceof ShowDatabasesStatement) {
            return new SchemaIgnoreBackendHandler(sqlStatement, frontendHandler);
        }
        
        if (sqlStatement instanceof ShowOtherStatement) {
            return new SchemaUnicastBackendHandler(connectionId, sequenceId, sql, backendConnection, DatabaseType.MySQL);
        }
        
        String schema = getSchema(sqlStatement).isPresent() ? getSchema(sqlStatement).get() : frontendHandler.getCurrentSchema();
        return newTextProtocolInstance(connectionId, sequenceId, sql, backendConnection, DatabaseType.MySQL, schema);
    }
    
    private static Optional<String> getSchema(final SQLStatement sqlStatement) {
        List<SQLToken> sqlTokens = sqlStatement.getSQLTokens();
        if (!sqlTokens.isEmpty()
                && (sqlStatement instanceof ShowTablesStatement || sqlStatement instanceof ShowColumnsStatement
                || sqlStatement instanceof ShowIndexStatement || sqlStatement instanceof ShowTableStatusStatement)) {
            return Optional.of(((SchemaToken) new LinkedList<>(sqlTokens).getLast()).getSchemaName());
        }
        return Optional.absent();
    }
}

@@ -2381,16 +2381,76 @@
 gistry;%0A
+import io.shardingsphere.shardingproxy.runtime.LogicSchema;%0A
 import i
@@ -3541,39 +3541,33 @@
         
-ShardingSchema sharding
+LogicSchema logic
 Schema =
@@ -3582,32 +3582,29 @@
 REGISTRY.get
-Sharding
+Logic
 Schema(schem
@@ -3636,39 +3636,33 @@
 tion.set
-ShardingSchema(sharding
+LogicSchema(logic
 Schema);
@@ -3742,32 +3742,29 @@
 kendHandler(
-sharding
+logic
 Schema, conn
@@ -3841,32 +3841,29 @@
 kendHandler(
-sharding
+logic
 Schema, sql,
@@ -3929,32 +3929,29 @@
 utorWrapper(
-sharding
+logic
 Schema)));%0A 
@@ -4736,31 +4736,25 @@
     
-ShardingSchema sharding
+LogicSchema logic
 Sche
@@ -4777,24 +4777,21 @@
 STRY.get
-Sharding
+Logic
 Schema(s
@@ -4831,31 +4831,25 @@
 .set
-ShardingSchema(sharding
+LogicSchema(logic
 Sche
@@ -4917,32 +4917,29 @@
 kendHandler(
-sharding
+logic
 Schema, conn
@@ -5020,24 +5020,21 @@
 Handler(
-sharding
+logic
 Schema, 
@@ -5116,24 +5116,21 @@
 Wrapper(
-sharding
+logic
 Schema, 

use logicSchema