import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.spark.IgniteContext;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;

public class StockSimulationsCacheWriter {

	private IgniteContext<StockSimulationReturnsKey, StockSimulationReturnsVal> jic;
	private static final Logger log = Logger.getLogger(StockSimulationsCacheWriter.class.getName());
	public StockSimulationsCacheWriter(JavaSparkContext jsc)
	{
		this();
		String config = null;
		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 IgniteContext<StockSimulationReturnsKey,StockSimulationReturnsVal>(jsc.sc(), config,true);
	}
	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);
		stockSimulationsDataCacheCfg.setName(PARTITIONED_CACHE_NAME);
		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(stockSimulationsDataCacheCfg).savePairs(rdd.rdd(),true);
	}
	
	public void putMktDataInCache(StockSimulationReturnsKey stockSimDataKey,StockSimulationReturnsVal stockSimData){
		stockSimulationsDataCache.put(stockSimDataKey,stockSimData);
	}
	
}