package com.sapient.gm.riskAnalytics.simulations.cache;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.Ignition;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.query.SqlQuery;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.spark.JavaIgniteContext;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;

import com.sapient.gm.riskAnalytics.ignite.CacheConnection;
import com.sapient.gm.riskAnalytics.simulations.IStockSimulationsReader;


public class StockSimulationsCacheWriter {

	private JavaSparkContext jsc;
	private JavaIgniteContext<StockSimulationReturnsKey, StockSimulationReturnsVal> jic;
	private static final Logger log = Logger.getLogger(StockSimulationsCacheWriter.class.getName());
	public StockSimulationsCacheWriter(JavaSparkContext jsc)
	{
		String config = null;
		this.jsc = jsc; 
		String env = System.getProperty("env", "aws");
		if(env.equals("aws")){
			log.log(Level.INFO, ">>> RUNNING IN AWS ENVIRONMENT <<<");
			  config = "config/aws-cloud/risk-analytics-cache-client-aws.xml";
		}else{
			log.log(Level.INFO, ">>> RUNNING IN LOCAL ENVIRONMENT <<<");
			config = "config/risk-analytics-cache-client.xml";
		}
		
		this.jic = new JavaIgniteContext<StockSimulationReturnsKey,StockSimulationReturnsVal>(jsc, config);
	}
	private IgniteCache<StockSimulationReturnsKey, StockSimulationReturnsVal> stockSimulationsDataCache = null;
	private static final String PARTITIONED_CACHE_NAME = StockSimulationsCacheWriter.class.getSimpleName() + "Partitioned";
	private CacheConfiguration<StockSimulationReturnsKey, StockSimulationReturnsVal> stockSimulationsDataCacheCfg = new CacheConfiguration<>(PARTITIONED_CACHE_NAME);
	
	public StockSimulationsCacheWriter(){
		stockSimulationsDataCacheCfg.setBackups(1);
		stockSimulationsDataCacheCfg.setCacheMode(CacheMode.PARTITIONED);
		stockSimulationsDataCacheCfg.setIndexedTypes(StockSimulationReturnsKey.class,StockSimulationReturnsVal.class);
		stockSimulationsDataCache = Ignition.ignite("riskAnalyticsGrid").getOrCreateCache(stockSimulationsDataCacheCfg);
		
	}
	
	
	
	public void putMktDataInCache(Map<StockSimulationReturnsKey,StockSimulationReturnsVal> stockSimulationsMap){
		stockSimulationsDataCache.putAll(stockSimulationsMap);
		
	}
	
	public void putMktDataInCache(JavaPairRDD<StockSimulationReturnsKey,StockSimulationReturnsVal> rdd){
		jic.fromCache(PARTITIONED_CACHE_NAME).savePairs(rdd);
	}
	
	public void putMktDataInCache(StockSimulationReturnsKey stockSimDataKey,StockSimulationReturnsVal stockSimData){
		stockSimulationsDataCache.put(stockSimDataKey,stockSimData);
	}
	
}