// Path: src/main/java/com/stratio/connector/deep/configuration/DeepConnectorConstants.java
// public final class DeepConnectorConstants {
// 	private DeepConnectorConstants(){
// 
// 	}
// 
// 	public static final String HDFS = "hdfs";
// 	public static final String SPARK_MASTER = "spark.master";
// 	public static final String SPARK_HOME = "spark.home";
// 	public static final String SPARK_JARS = "spark.jars";
// 	public static final String CLUSTER_PREFIX_CONSTANT = "cluster";
// 	public static final String IMPL_CLASS_SUFIX_CONSTANT = ".implClass";
// 	public static final String EXTRACTOR_IMPL_CLASS = "extractorImplClass";
// 	public static final String DEEP_CONNECTOR_JOB_CONSTANT = "DeepConnectorJob";
// 
// 	public static final int DEFAULT_RESULT_SIZE = 10000;
// 	public static final String PROPERTY_DEFAULT_LIMIT = "DefaultLimit";
//     public static final String SPARK_EXECUTOR_MEMORY = "spark.executor.memory";
//     public static final String SPARK_DRIVER_MEMORY = "spark.driver.memory";
//     public static final String  SPARK_TASK_CPUS = "spark.task.cpus";
//     public static final String  SPARK_DEFAULT_PARALELISM = "spark.default.parallelism";
//     public static final String  SPARK_CORES_MAX = "spark.cores.max";
//     public static final String  SPARK_DRIVER_RESULTSIZE = "spark.driver.maxResultSize";
// 
// 
// 
// 
// }


import com.stratio.deep.commons.entity.Cells;
import com.stratio.deep.commons.extractor.utils.ExtractorConstants;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import com.stratio.connector.commons.connection.Connection;
import com.stratio.connector.commons.util.ConnectorParser;
import com.stratio.connector.deep.configuration.DeepConnectorConstants;
import com.stratio.crossdata.common.connector.ConnectorClusterConfig;
import com.stratio.crossdata.common.exceptions.ConnectionException;
import com.stratio.crossdata.common.security.ICredentials;
import com.stratio.deep.commons.config.ExtractorConfig;
/*
 * Licensed to STRATIO (C) under one or more contributor license agreements.
 * See the NOTICE file distributed with this work for additional information
 * regarding copyright ownership.  The STRATIO (C) 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 com.stratio.connector.deep.connection;



/**
 * .Connection object exist in the ConnectionHandler and contains all the connection info & config.
 * {@link com.stratio.connector.commons.connection.Connection}
 *
 */
public class DeepConnection extends Connection<Object> {

	private boolean isConnect = false;

	private final ExtractorConfig<Cells> extractorConfig;

	/**
	 * Constructor using credentials and cluster config.
	 *
	 * @param credentials
	 *            the credentials.
	 * @param config
	 *            The cluster configuration.
	 */
	public DeepConnection(ICredentials credentials, ConnectorClusterConfig config)
			throws ConnectionException {

		if (credentials != null) {
			// TODO check the credentials
		}

		Map<String, String> clusterOptions = config.getClusterOptions();
		Map<String, String> connectorOptions = config.getConnectorOptions();
		// Creating a configuration for the Extractor and initialize it
		ExtractorConfig<Cells> extractorconfig = new ExtractorConfig<>(Cells.class);
		String extractorImplClassName = clusterOptions.get(DeepConnectorConstants.EXTRACTOR_IMPL_CLASS);

		if (extractorImplClassName == null) {
			throw new ConnectionException("Unknown data source, please add it to the configuration.");
		}
		Map<String,Serializable> values = returnConfig(clusterOptions);
// Path: src/main/java/com/stratio/connector/deep/configuration/DeepConnectorConstants.java
// public final class DeepConnectorConstants {
// 	private DeepConnectorConstants(){
// 
// 	}
// 
// 	public static final String HDFS = "hdfs";
// 	public static final String SPARK_MASTER = "spark.master";
// 	public static final String SPARK_HOME = "spark.home";
// 	public static final String SPARK_JARS = "spark.jars";
// 	public static final String CLUSTER_PREFIX_CONSTANT = "cluster";
// 	public static final String IMPL_CLASS_SUFIX_CONSTANT = ".implClass";
// 	public static final String EXTRACTOR_IMPL_CLASS = "extractorImplClass";
// 	public static final String DEEP_CONNECTOR_JOB_CONSTANT = "DeepConnectorJob";
// 
// 	public static final int DEFAULT_RESULT_SIZE = 10000;
// 	public static final String PROPERTY_DEFAULT_LIMIT = "DefaultLimit";
//     public static final String SPARK_EXECUTOR_MEMORY = "spark.executor.memory";
//     public static final String SPARK_DRIVER_MEMORY = "spark.driver.memory";
//     public static final String  SPARK_TASK_CPUS = "spark.task.cpus";
//     public static final String  SPARK_DEFAULT_PARALELISM = "spark.default.parallelism";
//     public static final String  SPARK_CORES_MAX = "spark.cores.max";
//     public static final String  SPARK_DRIVER_RESULTSIZE = "spark.driver.maxResultSize";
// 
// 
// 
// 
// }


// Path: src/main/java/com/stratio/connector/deep/connection/DeepConnection.java
import com.stratio.deep.commons.entity.Cells;
import com.stratio.deep.commons.extractor.utils.ExtractorConstants;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import com.stratio.connector.commons.connection.Connection;
import com.stratio.connector.commons.util.ConnectorParser;
import com.stratio.connector.deep.configuration.DeepConnectorConstants;
import com.stratio.crossdata.common.connector.ConnectorClusterConfig;
import com.stratio.crossdata.common.exceptions.ConnectionException;
import com.stratio.crossdata.common.security.ICredentials;
import com.stratio.deep.commons.config.ExtractorConfig;

/*
 * Licensed to STRATIO (C) under one or more contributor license agreements.
 * See the NOTICE file distributed with this work for additional information
 * regarding copyright ownership.  The STRATIO (C) 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 com.stratio.connector.deep.connection;



/**
 * .Connection object exist in the ConnectionHandler and contains all the connection info & config.
 * {@link com.stratio.connector.commons.connection.Connection}
 *
 */
public class DeepConnection extends Connection<Object> {

	private boolean isConnect = false;

	private final ExtractorConfig<Cells> extractorConfig;

	/**
	 * Constructor using credentials and cluster config.
	 *
	 * @param credentials
	 *            the credentials.
	 * @param config
	 *            The cluster configuration.
	 */
	public DeepConnection(ICredentials credentials, ConnectorClusterConfig config)
			throws ConnectionException {

		if (credentials != null) {
			// TODO check the credentials
		}

		Map<String, String> clusterOptions = config.getClusterOptions();
		Map<String, String> connectorOptions = config.getConnectorOptions();
		// Creating a configuration for the Extractor and initialize it
		ExtractorConfig<Cells> extractorconfig = new ExtractorConfig<>(Cells.class);
		String extractorImplClassName = clusterOptions.get(DeepConnectorConstants.EXTRACTOR_IMPL_CLASS);

		if (extractorImplClassName == null) {
			throw new ConnectionException("Unknown data source, please add it to the configuration.");
		}
		Map<String,Serializable> values = returnConfig(clusterOptions);
		values.put(DeepConnectorConstants.PROPERTY_DEFAULT_LIMIT,connectorOptions.get(DeepConnectorConstants