/*
    * Copyright 2008 fastConnect.
    *    
    * This file is part of Composite BulkDataPersister.  
    *
    * This is free software; you can redistribute it and/or modify it
    * under the terms of the GNU Lesser General Public License as
    * published by the Free Software Foundation; either version 3 of
    * the License, or (at your option) any later version.
   *
   * This software is distributed in the hope that it will be useful,
   * but WITHOUT ANY WARRANTY; without even the implied warranty of
   * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
   * Lesser General Public License for more details.
   *
   * You should have received a copy of the GNU Lesser General Public
   * License along with this software. If not, see <http://www.gnu.org/licenses/>.
   */
  package fr.fastconnect.gigaspaces.datasource;
  
  import java.io.FileOutputStream;
  import java.io.IOException;
  import java.io.ObjectOutputStream;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Properties;
  import java.util.concurrent.Callable;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.Future;
  import java.util.concurrent.LinkedBlockingQueue;
  import java.util.concurrent.ThreadFactory;
  import java.util.concurrent.ThreadPoolExecutor;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.logging.Level;
  import java.util.logging.Logger;
  
  import com.gigaspaces.datasource.BulkDataPersister;
  import com.gigaspaces.datasource.BulkItem;
  import com.gigaspaces.datasource.DataIterator;
  import com.gigaspaces.datasource.DataSourceException;
  import com.gigaspaces.datasource.ManagedDataSource;
  
  /**
   *Provides the capacity to use several implementation of BulkDataPersister using a Mirror.
   *
   *@param <T>
   */
  public class CompositeBulkDataPersister<T> implements BulkDataPersister, ManagedDataSource<T> {
  	
  	/**
  	 * Simple implementation of {@link Callable} calling {@link BulkDataPersister#executeBulk(List)} on provided {@link BulkDataPersister}.
  	 */
  	private static class BulkDataPersisterExecutor implements Callable<Void> {
  
  		private final BulkDataPersister bulkDataPersister;
  		private final List<BulkItem> bulkItems;
  		
  		public BulkDataPersisterExecutor(final BulkDataPersister bulkDataPersister, final List<BulkItem> bulkItems, final ExecutionFailureStrategy executionFailureStrategy) {
  			this.bulkDataPersister = bulkDataPersister;
  			this.bulkItems = bulkItems;
  		}
  		
  		public Void call() throws Exception {
  			this.bulkDataPersister.executeBulk(this.bulkItems);
  			return null;
  		}
  		
  	}
  
  	private static final Logger LOGGER = Logger.getLogger(CompositeBulkDataPersister.class.getName());
  	
  	public static final String BULK_DATA_PERSISTER_CLASS_NAMES_PROPERTY = "delegate-classnames";
  	public static final String BULK_DATA_PERSISTER_CLASS_NAMES_DELIMITER_PROPERTY = "delegate-classname-delimiter";
  	public static final String BULK_DATA_PERSISTER_EXECUTION_EXCEPTION_STRATEGY_CLASS_NAME_PROPERTY = "executionfailurestrategy-classname";
  	
  	private List<BulkDataPersister> bulkDataPersisterDelegates = null;
  	private ExecutionFailureStrategy executionFailureStrategy;
  	
  	private ExecutorService executorService;
  	
  	public CompositeBulkDataPersister() {
  	}
  	
  	public CompositeBulkDataPersister(final BulkDataPersister[] bulkDataPersisterDelegates) {
  		this.bulkDataPersisterDelegates = Arrays.asList(bulkDataPersisterDelegates);
  	}
  	
  	/** 
  	 * {@inheritDoc}
  	 */
  	public void init(final Properties properties) throws DataSourceException {
  		if (this.bulkDataPersisterDelegates == null) {
  			//get configured BulkDataPersister class names
 			if (properties.containsKey(BULK_DATA_PERSISTER_CLASS_NAMES_PROPERTY)) {
 				final String bulkDataPersisterClassNames = properties.getProperty(BULK_DATA_PERSISTER_CLASS_NAMES_PROPERTY);
 				if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.CONFIG)) {
 					CompositeBulkDataPersister.LOGGER.config("Using <"+bulkDataPersisterClassNames+"> as "+BULK_DATA_PERSISTER_CLASS_NAMES_PROPERTY);
 				}
 				
 				final String bulkDataPersisterClassNamesDelimiter = properties.getProperty(BULK_DATA_PERSISTER_CLASS_NAMES_DELIMITER_PROPERTY, ";");
 				if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.CONFIG)) {
 					CompositeBulkDataPersister.LOGGER.config("Using <"+bulkDataPersisterClassNamesDelimiter+"> as "+BULK_DATA_PERSISTER_CLASS_NAMES_DELIMITER_PROPERTY);
 				}
 				
 				//try to instantiate ExecutionFailureStrategy
 				final String bulkDataPersisterExecutionExceptionStrategyClassName = properties.getProperty(BULK_DATA_PERSISTER_EXECUTION_EXCEPTION_STRATEGY_CLASS_NAME_PROPERTY);
 				if (bulkDataPersisterExecutionExceptionStrategyClassName != null) {
 					if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.CONFIG)) {
 						CompositeBulkDataPersister.LOGGER.config("Using <"+bulkDataPersisterExecutionExceptionStrategyClassName+"> as "+BULK_DATA_PERSISTER_EXECUTION_EXCEPTION_STRATEGY_CLASS_NAME_PROPERTY);
 					}
 					
 					try {
 						final Class<?> bulkDataPersisterExecutionExceptionStrategyClass = Class.forName(bulkDataPersisterExecutionExceptionStrategyClassName.trim());
 						try {
 							final Object bulkDataPersisterExecutionExceptionStrategy = bulkDataPersisterExecutionExceptionStrategyClass.newInstance();
 							if (bulkDataPersisterExecutionExceptionStrategy instanceof ExecutionFailureStrategy) {
 								this.executionFailureStrategy = ExecutionFailureStrategy.class.cast(bulkDataPersisterExecutionExceptionStrategy);
 								this.executionFailureStrategy.init(properties);
 							} else {
 								final String message = "<"+bulkDataPersisterExecutionExceptionStrategy+"> is not a "+BulkDataPersister.class.getSimpleName();
 								if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.SEVERE)) {
 									CompositeBulkDataPersister.LOGGER.log(Level.SEVERE, message);
 								}
 								throw new DataSourceException(message);
 							}
 						} catch (InstantiationException e) {
 							final String message = "<"+bulkDataPersisterExecutionExceptionStrategyClass.getName()+"> has no default constructor; skipping";
 							if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.SEVERE)) {
 								CompositeBulkDataPersister.LOGGER.log(Level.SEVERE, message, e);
 							}
 							throw new DataSourceException(message, e);
 						} catch (IllegalAccessException e) {
 							final String message = "<"+bulkDataPersisterExecutionExceptionStrategyClass.getName()+"> has no visible default constructor; skipping";
 							if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.SEVERE)) {
 								CompositeBulkDataPersister.LOGGER.log(Level.SEVERE, message, e);
 							}
 							throw new DataSourceException(message, e);
 						}
 					} catch (ClassNotFoundException e) {
 						final String message = "<"+bulkDataPersisterExecutionExceptionStrategyClassName+"> cannot be found";
 						if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.SEVERE)) {
 							CompositeBulkDataPersister.LOGGER.log(Level.SEVERE, message, e);
 						}
 						throw new DataSourceException(message, e);
 					}
 				}
 				
 				this.bulkDataPersisterDelegates = new ArrayList<BulkDataPersister>(bulkDataPersisterClassNames.length());
 				
 				//try to instantiate all defined BulkDataPersister classes
 				for (final String bulkDataPersisterClassName : bulkDataPersisterClassNames.split(bulkDataPersisterClassNamesDelimiter)) {
 					try {
 						final Class<?> bulkDataPersisterClass = Class.forName(bulkDataPersisterClassName.trim());
 						try {
 							final Object bulkDataPersister = bulkDataPersisterClass.newInstance();
 							if (bulkDataPersister instanceof BulkDataPersister) {
 								this.bulkDataPersisterDelegates.add(BulkDataPersister.class.cast(bulkDataPersister));
 							} else {
 								if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.WARNING)) {
 									CompositeBulkDataPersister.LOGGER.warning("<"+bulkDataPersister+"> is not a "+BulkDataPersister.class.getSimpleName()+"; skipping");
 								}
 							}
 						} catch (InstantiationException e) {
 							if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.WARNING)) {
 								CompositeBulkDataPersister.LOGGER.log(Level.WARNING, "<"+bulkDataPersisterClass.getName()+"> has no default constructor; skipping", e);
 							}
 						} catch (IllegalAccessException e) {
 							if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.WARNING)) {
 								CompositeBulkDataPersister.LOGGER.log(Level.WARNING, "<"+bulkDataPersisterClass.getName()+"> has no visible default constructor; skipping", e);
 							}
 						}
 					} catch (ClassNotFoundException e) {
 						if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.WARNING)) {
 							CompositeBulkDataPersister.LOGGER.log(Level.WARNING, "<"+bulkDataPersisterClassName+"> cannot be found; skipping", e);
 						}
 					}
 					
 				}
 			} else {
 				final String message = "No "+BULK_DATA_PERSISTER_CLASS_NAMES_PROPERTY+" defined";
 				if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.SEVERE)) {
 					CompositeBulkDataPersister.LOGGER.log(Level.SEVERE, message);
 				}
 				throw new DataSourceException(message);
 			}
 		} else {
 			if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.CONFIG)) {
 				CompositeBulkDataPersister.LOGGER.log(Level.CONFIG, CompositeBulkDataPersister.class.getSimpleName()+" configured manually; ignoring properties");
 			}
 		}
 		
 		if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.CONFIG)) {
 			CompositeBulkDataPersister.LOGGER.log(Level.CONFIG, "Delegating to <"+this.bulkDataPersisterDelegates+"> using <"+this.executionFailureStrategy+"> execution exception strategy");
 		}
 		if (this.executionFailureStrategy == null) {
 			if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.WARNING)) {
 				CompositeBulkDataPersister.LOGGER.warning("No "+ExecutionFailureStrategy.class.getSimpleName()+" defined; execution exception will be ignored");
 			}
 		}
 	}
 	
 	/** 
 	 * {@inheritDoc}
 	 */
 	public DataIterator<T> initialLoad() throws DataSourceException {
 		return null;
 	}
 	
 	/** 
 	 * {@inheritDoc}
 	 */
 	public void executeBulk(final List<BulkItem> bulkItems) throws DataSourceException {		
 		final ExecutorService executorService = getExecutorService();
 		final Map<Future<Void>, BulkDataPersister> bulkDataPersisterExecutions = new HashMap<Future<Void>, BulkDataPersister>(this.bulkDataPersisterDelegates.size());
 		for (final BulkDataPersister bulkDataPersister : this.bulkDataPersisterDelegates) {
 			bulkDataPersisterExecutions.put(executorService.submit(new BulkDataPersisterExecutor(bulkDataPersister, bulkItems, this.executionFailureStrategy)), bulkDataPersister);
 		}
 		
 		//wait for end of execution of all threads
 		for (final Map.Entry<Future<Void>, BulkDataPersister> result : bulkDataPersisterExecutions.entrySet()) {
 			try {
 				//TODO do we have to handle a timeout here? 
 				result.getKey().get();
 			} catch (InterruptedException e) {
 				if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.SEVERE)) {
 					CompositeBulkDataPersister.LOGGER.log(Level.SEVERE, "Thread <"+Thread.currentThread().getName()+"> has been interrupted", e);
 				}
 				
 				Thread.currentThread().interrupt();
 			} catch (ExecutionException e) {
 				final BulkDataPersister bulkDataPersister = result.getValue();
 				if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.SEVERE)) {
 					CompositeBulkDataPersister.LOGGER.log(Level.SEVERE, "Execution of <"+bulkDataPersister+"> has thrown an exception", e);
 				}
 
 				//if exception thrown is a DataSourceException and we have an executionFailureStrategy defined then delegate handling to it
 				if (e.getCause() != null && e.getCause() instanceof DataSourceException && this.executionFailureStrategy != null) {
 					if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.INFO)) {
 						CompositeBulkDataPersister.LOGGER.log(Level.INFO, "Handling execution exception using <"+this.executionFailureStrategy+">");
 					}
 					
 					try {
 						this.executionFailureStrategy.handleExecutionFailure(bulkDataPersister, bulkItems, DataSourceException.class.cast(e.getCause()));
 					} catch (Exception unrecoverableException) {
 						if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.SEVERE)) {
 							CompositeBulkDataPersister.LOGGER.log(Level.SEVERE, "Exception trying to handle execution exception", unrecoverableException);
 						}
 						
 						handleUnrecoverableException(bulkDataPersister, bulkItems, unrecoverableException);
 					}
 				} else {
 					handleUnrecoverableException(bulkDataPersister, bulkItems, e);
 				}
 			}
 		}
 	}
 	
 	/**
 	 * Constructs the {@link ExecutorService} implementation used to execute concurrently all {@link BulkDataPersister} delegates.
 	 * Called each time {@link CompositeBulkDataPersister#executeBulk(List)} is called.
 	 * <br />
 	 * <br />
 	 * Default implementation returns a {@link ThreadPoolExecutor} using one named thread (CompositeBulkDataPersisterThread-X) per {@link BulkDataPersister} delegate.
 	 *  
 	 * @return the executorService
 	 */
 	protected synchronized ExecutorService getExecutorService() {
 		if (this.executorService == null) {
 			this.executorService = new ThreadPoolExecutor(this.bulkDataPersisterDelegates.size(), this.bulkDataPersisterDelegates.size(), 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), new ThreadFactory() {
 
 				private final ThreadFactory defaultThreadFactory = Executors.defaultThreadFactory();
 				private final AtomicInteger threadCount = new AtomicInteger(0);
 				
 				public Thread newThread(final Runnable runnable) {
 					final Thread thread = this.defaultThreadFactory.newThread(runnable);
 					thread.setName(CompositeBulkDataPersister.class.getSimpleName()+"Thread-"+String.valueOf(this.threadCount.incrementAndGet()));
 					return thread;
 				}
 				
 			});
 		}
 		return this.executorService;
 	}
 	
 	/**
 	 * Last chance to handle an exception thrown by {@link BulkDataPersister#executeBulk(List)}.
 	 * <br />
 	 * <br />
 	 * Is called when:
 	 * -exception thrown is a {@link DataSourceException} and provided execution exception strategy throws an exception
 	 * -exception thrown is a {@link DataSourceException} and there is no provided execution exception strategy
 	 * -exception thrown is not a {@link DataSourceException}
 	 * <br />
 	 * Is NOT called when:
 	 * -exception thrown is a {@link DataSourceException} and provided execution exception strategy DOES NOT throw an exception
 	 * <br />
 	 * <br />
 	 * Default implementation use a {@link ObjectOutputStream} to dump bulkItems.
 	 * 
 	 * @param bulkDataPersister the persister used
 	 * @param bulkItems the information received
 	 * @param exception the exception generated by {@link BulkDataPersister#executeBulk(List)}
 	 * @throws DataSourceException if thrown propagated to mirror which triggers redo-log mechanism
 	 */
 	protected void handleUnrecoverableException(final BulkDataPersister bulkDataPersister, final List<BulkItem> bulkItems, final Exception exception) throws DataSourceException {
 		ObjectOutputStream objectOutputStream = null;
 		final String fileName = "unrecoverable-exception-"+bulkDataPersister.getClass().getName()+".txt";
 		try {
 			objectOutputStream = new ObjectOutputStream(new FileOutputStream(fileName, false));
 			objectOutputStream.writeObject(bulkItems);
 		} catch (IOException e) {
 			throw new DataSourceException("Got exception dealing with unrecoverable exception; propagating", exception);
 		} finally {
 			if (objectOutputStream != null) {
 				try {
 					objectOutputStream.close();
 				} catch (IOException e) {
 					if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.SEVERE)) {
 						CompositeBulkDataPersister.LOGGER.log(Level.SEVERE, "Exception closing <"+fileName+">", e);
 					}
 				}
 			}
 		}
 	}
 
 	/** 
 	 * {@inheritDoc}
 	 */
 	public void shutdown() throws DataSourceException {
 		//if one of provided BulkDataPersister is a ManagedDataSource safely shutdown it
 		for (final BulkDataPersister bulkDataPersister : this.bulkDataPersisterDelegates) {
 			if (bulkDataPersister instanceof ManagedDataSource) {
 				try {
 					ManagedDataSource.class.cast(bulkDataPersister).shutdown();
 				} catch (DataSourceException e) {
 					if (CompositeBulkDataPersister.LOGGER.isLoggable(Level.SEVERE)) {
 						CompositeBulkDataPersister.LOGGER.log(Level.SEVERE, "Exception trying to shutdown <"+bulkDataPersister+">", e);
 					}
 				}
 			}
 		}
 	}
 	
 	public void setExecutionFailureStrategy(final ExecutionFailureStrategy executionFailureStrategy) {
 		this.executionFailureStrategy = executionFailureStrategy;
 	}
 
 }