accounting-lib/src/main/java/org/gcube/accounting/persistence/Persistence.java

/**
 * 
 */
package org.gcube.accounting.persistence;

import java.io.File;
import java.util.ServiceLoader;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.gcube.accounting.aggregation.scheduler.AggregationScheduler;
import org.gcube.accounting.datamodel.SingleUsageRecord;
import org.gcube.accounting.datamodel.UsageRecord;
import org.gcube.accounting.exception.InvalidValueException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author Luca Frosini (ISTI - CNR) http://www.lucafrosini.com/
 */
public abstract class Persistence {

	private static final Logger logger = LoggerFactory.getLogger(Persistence.class);
	
	private static final String ACCOUTING_FALLBACK_FILENAME = "accountingFallback.log";
	
	/**
	 * The singleton instance of persistence
	 */
	protected static Persistence persistence;
	protected static FallbackPersistence fallback;
	protected static AggregationScheduler aggregationScheduler;
	
	private static File file(File file) throws IllegalArgumentException {
		
		if (file.isDirectory())
			throw new IllegalArgumentException(file.getAbsolutePath() + " cannot be used in write mode because it's folder");

		//create folder structure it does not exist
		if (!file.getParentFile().exists())
			file.getParentFile().mkdirs();
		
		return file;
		
	}
	
	public synchronized static void setFallbackLocation(String path){
		if(fallback == null){
			if(path==null){
				path = ".";
			}
			File file =  file(new File(path, ACCOUTING_FALLBACK_FILENAME));
			fallback = new FallbackPersistence(file);
		}
	}
	
	protected static void init() {
		setFallbackLocation(null);
		try {
			ServiceLoader<Persistence> serviceLoader = ServiceLoader.load(Persistence.class);
			for (Persistence foundPersistence : serviceLoader) {
				if(foundPersistence.getClass().isInstance(FallbackPersistence.class)){
					continue;
				}
				try {
					String foundPersistenceClassName = foundPersistence.getClass().getSimpleName();
					logger.debug("Testing {}", foundPersistenceClassName);
					String scope = null; // TODO
					PersistenceConfiguration configuration = PersistenceConfiguration.getPersistenceConfiguration(scope, foundPersistenceClassName);
					foundPersistence.prepareConnection(configuration);
					/*
					 * Uncomment the following line of code if you want to try 
					 * to create a test UsageRecord before setting the
					 * foundPersistence as default
					 * 
					 * foundPersistence.accountWithFallback(TestUsageRecord.createTestServiceUsageRecord());
					 */
					persistence = foundPersistence;
					break;
				} catch (Exception e) {
					logger.debug(String.format("%s not initialized correctly. It will not be used", foundPersistence.getClass().getSimpleName()));
					
				}
			}
			if(persistence==null){
				persistence = fallback;
			}
		} catch(Exception e){
			logger.error("Unable to instance a Persistence Implementation. Using fallback as default", 
					e.getCause());
			persistence = fallback;
		}
		aggregationScheduler = AggregationScheduler.getInstance();
	}
	
	/**
	 * Pool for thread execution
	 */
	private ExecutorService pool;
	
	/**
	 * @return the singleton instance of persistence
	 * @throws Exception  if fails
	 */
	public static Persistence getInstance() {
		if(persistence==null){
			init();
		}
		return persistence;
	}
	
	protected Persistence() {
		pool = Executors.newCachedThreadPool();
	}
	
	/**
	 * Prepare the connection to persistence.
	 * This method must be used by implementation class to open 
	 * the connection with the persistence storage, DB, file etc.
	 * @param configuration The configuration to create the connection
	 * @throws Exception if fails
	 */
	protected abstract void prepareConnection(PersistenceConfiguration configuration) throws Exception;
	
	/* *
	 * Prepare the connection and try to write a test record on default 
	 * persistence and fallback persistence.
	 * This method should not be re-implemented from subclass.
	 * @throws Exception if fails
	 * /
	public void connect() throws Exception {
		persistence.prepareConnection();
		persistence.account(createTestUsageRecord());
	}
	*/
	
	/**
	 * This method contains the code to save the {@link #UsageRecord}
	 * 
	 */
	protected abstract void reallyAccount(UsageRecord usageRecords) throws Exception;
	
	private void accountWithFallback(UsageRecord... usageRecords) {
		String persistenceName = getInstance().getClass().getSimpleName();
		for(UsageRecord usageRecord : usageRecords){
			try {
				//logger.debug("Going to account {} using {}", usageRecord, persistenceName);
				persistence.reallyAccount(usageRecord);
				logger.debug("{} accounted succesfully from {}.", usageRecord.toString(), persistenceName);
			} catch (Exception e) {
				String fallabackPersistenceName = fallback.getClass().getSimpleName();
				try {
					logger.error("{} was not accounted succesfully from {}. Trying to use {}.", 
							usageRecord.toString(), persistenceName, fallabackPersistenceName);
					fallback.reallyAccount(usageRecord);
					logger.debug("{} accounted succesfully from {}", 
							usageRecord.toString(), fallabackPersistenceName);
				}catch(Exception ex){
					logger.error("{} was not accounted at all", usageRecord.toString());
				}
			}
		}
	}
	
	
	protected void validateAccountAggregate(final SingleUsageRecord usageRecord, boolean validate, boolean aggregate){
		try {
			if(validate){
				usageRecord.validate();
			}
			if(aggregate){
				aggregationScheduler.aggregate(usageRecord, new PersistenceExecutor(){
	
					@Override
					public void persist(UsageRecord... usageRecords) throws Exception {
						persistence.accountWithFallback(usageRecords);	
					}
					
				});
			}else{
				persistence.accountWithFallback(usageRecord);	
			}
			
		} catch (InvalidValueException e) {
			logger.error("Error validating UsageRecord", e.getCause());
		} catch (Exception e) {
			logger.error("Error accounting UsageRecord", e.getCause());
		}
	}
	
	/**
	 * Persist the {@link #UsageRecord}.
	 * The Record is validated first, then accounted, in a separated thread. 
	 * So that the program can continue the execution.
	 * If the persistence fails the class write that the record in a local file
	 * so that the {@link #UsageRecord} can be recorder later.
	 * @param usageRecord the {@link #UsageRecord} to persist
	 * @throws InvalidValueException if the Record Validation Fails
	 */
	public void account(final SingleUsageRecord usageRecord) throws InvalidValueException{
		Runnable runnable = new Runnable(){
			@Override
			public void run(){
				validateAccountAggregate(usageRecord, true, true);
			}
		};
		pool.execute(runnable);
		
	}
	
	public void flush() throws Exception {
		aggregationScheduler.flush(new PersistenceExecutor(){

			@Override
			public void persist(UsageRecord... usageRecords) throws Exception {
				persistence.accountWithFallback(usageRecords);	
			}
			
		});
	}
	
	
	public abstract void close() throws Exception;
	
}