/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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.apache.shardingsphere.proxy.backend.communication.jdbc.execute;

import lombok.RequiredArgsConstructor;
import org.apache.shardingsphere.infra.config.DataSourceConfiguration;
import org.apache.shardingsphere.infra.config.RuleConfiguration;
import org.apache.shardingsphere.infra.executor.sql.context.ExecutionContext;
import org.apache.shardingsphere.infra.yaml.swapper.YamlRuleConfigurationSwapperEngine;
import org.apache.shardingsphere.proxy.backend.communication.jdbc.execute.generator.YamlDataSourceConfigurationGenerator;
import org.apache.shardingsphere.proxy.backend.communication.jdbc.execute.generator.YamlShardingRuleConfigurationGenerator;
import org.apache.shardingsphere.proxy.backend.response.BackendResponse;
import org.apache.shardingsphere.proxy.backend.response.update.UpdateResponse;
import org.apache.shardingsphere.proxy.config.util.DataSourceConverter;
import org.apache.shardingsphere.proxy.config.yaml.YamlDataSourceParameter;
import org.apache.shardingsphere.sharding.yaml.config.YamlShardingRuleConfiguration;
import org.apache.shardingsphere.sql.parser.binder.statement.CommonSQLStatementContext;
import org.apache.shardingsphere.sql.parser.binder.statement.ddl.CreateDataSourcesStatementContext;
import org.apache.shardingsphere.sql.parser.binder.statement.ddl.CreateShardingRuleStatementContext;
import org.apache.shardingsphere.sql.parser.sql.statement.SQLStatement;

import java.util.Collection;
import java.util.Collections;
import java.util.LinkedList;
import java.util.Map;

/**
 * Registry center execute engine.
 */
@RequiredArgsConstructor
public class RegistryCenterExecuteEngine implements SQLExecuteEngine {
    
    private final String schemaName;
    
    private final SQLStatement sqlStatement;
    
    @Override
    public final ExecutionContext execute(final String sql) {
        return new ExecutionContext(new CommonSQLStatementContext(sqlStatement), new LinkedList<>());
    }
    
    @Override
    public final BackendResponse execute(final ExecutionContext executionContext) {
        if (executionContext.getSqlStatementContext() instanceof CreateDataSourcesStatementContext) {
            return execute((CreateDataSourcesStatementContext) executionContext.getSqlStatementContext());
        }
        return execute((CreateShardingRuleStatementContext) executionContext.getSqlStatementContext());
    }
    
    private BackendResponse execute(final CreateDataSourcesStatementContext context) {
        Map<String, YamlDataSourceParameter> parameters = new YamlDataSourceConfigurationGenerator().generate(context);
        Map<String, DataSourceConfiguration> dataSources = DataSourceConverter.getDataSourceConfigurationMap(DataSourceConverter.getDataSourceParameterMap2(parameters));
        // TODO Check whether registry center exists.
        // TODO Load dataSources from registry center, merge them and persist them.
        UpdateResponse result = new UpdateResponse();
        result.setType("CREATE");
        return result;
    }
    
    private BackendResponse execute(final CreateShardingRuleStatementContext context) {
        YamlShardingRuleConfiguration configurations = new YamlShardingRuleConfigurationGenerator().generate(context);
        Collection<RuleConfiguration> rules = new YamlRuleConfigurationSwapperEngine().swapToRuleConfigurations(Collections.singleton(configurations));
        // TODO Check whether registry center exists.
        // TODO Persist it.
        UpdateResponse result = new UpdateResponse();
        result.setType("CREATE");
        return result;
    }
}

@@ -908,24 +908,182 @@
 onstructor;%0A
+import org.apache.shardingsphere.infra.callback.orchestration.DataSourceCallback;%0Aimport org.apache.shardingsphere.infra.callback.orchestration.RuleCallback;%0A
 import org.a
@@ -1372,16 +1372,80 @@
 Engine;%0A
+import org.apache.shardingsphere.kernel.context.SchemaContexts;%0A
 import o
@@ -1747,32 +1747,109 @@
 ackendResponse;%0A
+import org.apache.shardingsphere.proxy.backend.response.error.ErrorResponse;%0A
 import org.apach
@@ -1903,32 +1903,107 @@
 UpdateResponse;%0A
+import org.apache.shardingsphere.proxy.backend.schema.ProxySchemaContexts;%0A
 import org.apach
@@ -2581,16 +2581,46 @@
 ement;%0A%0A
+import java.sql.SQLException;%0A
 import j
@@ -3658,32 +3658,218 @@
 text context) %7B%0A
+        if (!isRegistryCenterExisted()) %7B%0A            return new ErrorResponse(new SQLException(%22No Registry center to execute %60%25s%60 SQL%22, context.getClass().getSimpleName()));%0A        %7D%0A
         Map%3CStri
@@ -4162,129 +4162,148 @@
 ODO 
-Check whether registry center exists.%0A        // TODO Load dataSources from registry center, merge them and persist them.
+Need to get the executed feedback from registry center for returning.%0A        DataSourceCallback.getInstance().run(schemaName, dataSources);
 %0A   
@@ -4501,32 +4501,218 @@
 text context) %7B%0A
+        if (!isRegistryCenterExisted()) %7B%0A            return new ErrorResponse(new SQLException(%22No Registry center to execute %60%25s%60 SQL%22, context.getClass().getSimpleName()));%0A        %7D%0A
         YamlShar
@@ -4986,183 +4986,404 @@
 ODO 
-Check whether registry center exists.%0A        // TODO Persist it.%0A        UpdateResponse result = new UpdateResponse();%0A        result.setType(%22CREATE%22);%0A        return result
+Need to get the executed feedback from registry center for returning.%0A        RuleCallback.getInstance().run(schemaName, rules);%0A        UpdateResponse result = new UpdateResponse();%0A        result.setType(%22CREATE%22);%0A        return result;%0A    %7D%0A    %0A    private boolean isRegistryCenterExisted() %7B%0A        return !(ProxySchemaContexts.getInstance().getSchemaContexts() instanceof SchemaContexts)
 ;%0A  

Use registry center to run SQL for proxy