/*
 * 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.proxy.backend.common;

import io.shardingsphere.core.parsing.parser.sql.SQLStatement;
import io.shardingsphere.core.routing.router.masterslave.MasterVisitedManager;
import io.shardingsphere.proxy.config.RuleRegistry;
import io.shardingsphere.proxy.transport.mysql.constant.ColumnType;
import io.shardingsphere.proxy.transport.mysql.constant.StatusFlag;
import io.shardingsphere.proxy.transport.mysql.packet.command.CommandResponsePackets;
import io.shardingsphere.proxy.transport.mysql.packet.command.text.query.ColumnDefinition41Packet;
import io.shardingsphere.proxy.transport.mysql.packet.command.text.query.FieldCountPacket;
import io.shardingsphere.proxy.transport.mysql.packet.command.text.query.TextResultSetRowPacket;
import io.shardingsphere.proxy.transport.mysql.packet.generic.EofPacket;
import io.shardingsphere.proxy.transport.mysql.packet.generic.ErrPacket;
import io.shardingsphere.proxy.transport.mysql.packet.generic.OKPacket;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;

/**
 * Abstract ExecuteWorker class, include SQL and PreparedStatement implement.
 *
 * @author zhaojun
 */
@AllArgsConstructor
@Getter
@Slf4j
public abstract class ExecuteWorker implements Callable<CommandResponsePackets> {
    
    private final ExecuteBackendHandler executeBackendHandler;
    
    private final SQLStatement sqlStatement;
    
    @Override
    public CommandResponsePackets call() {
        try {
            return execute();
        } catch (SQLException ex) {
            log.error("ExecuteWorker", ex);
            return new CommandResponsePackets(new ErrPacket(1, ex));
        } finally {
            MasterVisitedManager.clear();
        }
    }
    
    private CommandResponsePackets execute() throws SQLException {
        switch (sqlStatement.getType()) {
            case DQL: case DAL:
                return executeQuery();
            case DML: case DDL:
                return executeUpdate();
            default:
                return executeCommon();
        }
    }
    
    private CommandResponsePackets executeQuery() throws SQLException {
        if (ProxyMode.MEMORY_STRICTLY == RuleRegistry.getInstance().getProxyMode()) {
            return executeQueryWithStreamResultSet();
        } else if (ProxyMode.CONNECTION_STRICTLY == RuleRegistry.getInstance().getProxyMode()) {
            return executeQueryWithNonStreamResultSet();
        } else {
            return new CommandResponsePackets(new ErrPacket(1, 0, "", "Invalid proxy.mode"));
        }
    }
    
    protected abstract CommandResponsePackets executeQueryWithStreamResultSet() throws SQLException;
    
    protected abstract CommandResponsePackets executeQueryWithNonStreamResultSet() throws SQLException;
    
    protected abstract CommandResponsePackets executeUpdate() throws SQLException;
    
    protected abstract CommandResponsePackets executeCommon() throws SQLException;
    
    protected CommandResponsePackets getQueryDatabaseProtocolPackets(final ResultSet resultSet) throws SQLException {
        CommandResponsePackets result = new CommandResponsePackets();
        int currentSequenceId = 0;
        ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
        int columnCount = resultSetMetaData.getColumnCount();
        executeBackendHandler.setColumnCount(columnCount);
        if (0 == columnCount) {
            result.addPacket(new OKPacket(++currentSequenceId));
            return result;
        }
        result.addPacket(new FieldCountPacket(++currentSequenceId, columnCount));
        for (int i = 1; i <= columnCount; i++) {
            setColumnType(ColumnType.valueOfJDBCType(resultSetMetaData.getColumnType(i)));
            result.addPacket(new ColumnDefinition41Packet(++currentSequenceId, resultSetMetaData.getSchemaName(i), resultSetMetaData.getTableName(i),
                    resultSetMetaData.getTableName(i), resultSetMetaData.getColumnLabel(i), resultSetMetaData.getColumnName(i),
                    resultSetMetaData.getColumnDisplaySize(i), ColumnType.valueOfJDBCType(resultSetMetaData.getColumnType(i)), 0));
        }
        result.addPacket(new EofPacket(++currentSequenceId, 0, StatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue()));
        return result;
    }
    
    protected void setColumnType(final ColumnType columnType) {
    }
    
    protected CommandResponsePackets getCommonDatabaseProtocolPackets(final ResultSet resultSet) throws SQLException {
        CommandResponsePackets result = new CommandResponsePackets();
        int currentSequenceId = 0;
        ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
        int columnCount = resultSetMetaData.getColumnCount();
        executeBackendHandler.setColumnCount(columnCount);
        if (0 == columnCount) {
            result.addPacket(new OKPacket(++currentSequenceId));
            return result;
        }
        result.addPacket(new FieldCountPacket(++currentSequenceId, columnCount));
        for (int i = 1; i <= columnCount; i++) {
            result.addPacket(new ColumnDefinition41Packet(++currentSequenceId, resultSetMetaData.getSchemaName(i), resultSetMetaData.getTableName(i),
                    resultSetMetaData.getTableName(i), resultSetMetaData.getColumnLabel(i), resultSetMetaData.getColumnName(i),
                    resultSetMetaData.getColumnDisplaySize(i), ColumnType.valueOfJDBCType(resultSetMetaData.getColumnType(i)), 0));
        }
        result.addPacket(new EofPacket(++currentSequenceId, 0, StatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue()));
        while (resultSet.next()) {
            List<Object> data = new ArrayList<>(columnCount);
            for (int i = 1; i <= columnCount; i++) {
                data.add(resultSet.getObject(i));
            }
            result.addPacket(new TextResultSetRowPacket(++currentSequenceId, data));
        }
        result.addPacket(new EofPacket(++currentSequenceId, 0, StatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue()));
        return result;
    }
    
    protected long getGeneratedKey(final Statement statement) throws SQLException {
        long result = 0;
        ResultSet resultSet = statement.getGeneratedKeys();
        if (resultSet.next()) {
            result = resultSet.getLong(1);
        }
        return result;
    }
}

@@ -1893,25 +1893,16 @@
 *%0A *
- Abstract
  Execute
 Work
@@ -1901,23 +1901,28 @@
 cute
-W
+ w
 orker
- class, i
+.%0A * %0A * %3Cp%3EI
 nclu
@@ -1960,16 +1960,20 @@
 plement.
+%3C/p%3E
 %0A *%0A * @
@@ -2982,41 +2982,16 @@
     
-if (ProxyMode.MEMORY_STRICTLY == 
+switch (
 Rule
@@ -3023,32 +3023,70 @@
 tProxyMode()) %7B%0A
+            case MEMORY_STRICTLY:%0A    
             retu
@@ -3135,97 +3135,42 @@
     
-%7D else if (ProxyMode.CONNECTION_STRICTLY == RuleRegistry.getInstance().getProxyMode()) %7B%0A
+    case CONNECTION_STRICTLY:%0A    
     
@@ -3234,16 +3234,20 @@
     
-%7D else %7B
+    default:
 %0A   
@@ -3259,86 +3259,99 @@
     
-return new CommandResponsePackets(new ErrPacket(1, 0, %22%22, %22Invalid p
+    throw new UnsupportedOperationException(RuleRegistry.getInstance().getP
 roxy
-.mode%22
+Mode().name(
 ));%0A

throw exception if executeQuery use unsupported enum