/*
    * Copyright James Leigh (c) 2007.
    *
    * Licensed under the Aduna BSD-style license.
    */
   package org.openrdf.repository.flushable;
   
   import info.aduna.iteration.CloseableIteration;
   import info.aduna.iteration.FilterIteration;
  
  import java.util.Iterator;
  
  import org.openrdf.model.Resource;
  import org.openrdf.model.Statement;
  import org.openrdf.model.URI;
  import org.openrdf.model.Value;
  import org.openrdf.repository.Repository;
  import org.openrdf.repository.RepositoryConnection;
  import org.openrdf.repository.RepositoryException;
  import org.openrdf.repository.RepositoryResult;
  import org.openrdf.repository.base.RepositoryConnectionWrapper;
  import org.openrdf.repository.sail.SailRepository;
  import org.openrdf.repository.util.RDFInserter;
  import org.openrdf.rio.RDFHandler;
  import org.openrdf.rio.RDFHandlerException;
  import org.openrdf.sail.memory.MemoryStore;
  
  /**
   * When not in autoFlush mode, each statement added and removed is
   * tracked. When calling one of the read methods listed below the statement
   * results will be corrected to include or exclude the modified statements on
   * this connection. This behaviour mimics transaction isolation, but does not
   * include unflushed statements in query processing.
   * 
   * @author James Leigh
   * @see #getStatements(Resource, URI, Value, boolean, Resource[])
   * @see #hasStatement(Statement, boolean, Resource[])
   * @see #hasStatement(Resource, URI, Value, boolean, Resource[])
   * @see #isEmpty()
   * @see #export(RDFHandler, Resource[])
   * @see #exportStatements(Resource, URI, Value, boolean, RDFHandler, Resource[])
   * 
   */
  public class FlushableConnection extends RepositoryConnectionWrapper {
  
  	private boolean autoFlush = true;
  
  	private RepositoryConnection added;
  
  	RepositoryConnection removed;
  
  	public FlushableConnection(Repository repository)
  			throws RepositoryException {
  		this(repository, repository.getConnection());
  	}
  
  	public FlushableConnection(Repository repository, RepositoryConnection conn)
  			throws RepositoryException {
  		super(repository, conn);
  		if (!autoFlush) {
  			initUnFlushedStatements();
  		}
  	}
  
  	public boolean isAutoFlush() throws RepositoryException {
  		return autoFlush;
  	}
  
  	public void setAutoFlush(boolean autoFlush) throws RepositoryException {
  		if (this.autoFlush == autoFlush)
  			return;
  		if (autoFlush) {
  			flush();
  			destroyUnflushedStatements();
  		} else {
  			initUnFlushedStatements();
  		}
  		this.autoFlush = autoFlush;
  	}
  
  	public void flush() throws RepositoryException {
  		if (autoFlush)
  			return;
  		RepositoryResult<Statement> stmts;
  		if (!added.isEmpty()) {
  			stmts = added.getStatements(null, null, null, false);
  			try {
  				getDelegate().add(stmts);
  			} finally {
  				stmts.close();
  			}
  			added.clear();
  		}
  		if (!removed.isEmpty()) {
  			stmts = removed.getStatements(null, null, null, false);
  			try {
  				while (stmts.hasNext()) {
  					getDelegate().remove(stmts.next());
  				}
 			} finally {
 				stmts.close();
 			}
 			removed.clear();
 		}
 	}
 
 	public void clear() throws RepositoryException {
 		if (autoFlush)
 			return;
 		added.clear();
 		removed.clear();
 	}
 
 	@Override
 	protected boolean isDelegatingAdd() throws RepositoryException {
 		return autoFlush;
 	}
 
 	@Override
 	protected boolean isDelegatingRead() throws RepositoryException {
 		if (autoFlush)
 			return true;
 		return added.isEmpty() && removed.isEmpty();
 	}
 
 	@Override
 	protected boolean isDelegatingRemove() throws RepositoryException {
 		return autoFlush;
 	}
 
 	@Override
 	public void close() throws RepositoryException {
 		if (!autoFlush) {
 			clear();
 			setAutoFlush(true);
 		}
 		super.close();
 	}
 
 	@Override
 	public void setAutoCommit(boolean autoCommit) throws RepositoryException {
 		if (autoCommit && !autoFlush) {
 			flush();
 		}
 		super.setAutoCommit(autoCommit);
 	}
 
 	@Override
 	public void commit() throws RepositoryException {
 		if (!autoFlush) {
 			flush();
 		}
 		super.commit();
 	}
 
 	@Override
 	public void rollback() throws RepositoryException {
 		if (!autoFlush) {
 			clear();
 		}
 		super.rollback();
 	}
 
 	@Override
 	public synchronized RepositoryResult<Statement> getStatements(
 			Resource subj, URI pred, Value obj, final boolean includeInferred,
 			Resource... contexts) throws RepositoryException {
 		if (isDelegatingRead())
 			return super.getStatements(subj, pred, obj, includeInferred,
 					contexts);
 		final Iterator<Statement> include;
 		final boolean excluded = removed.hasStatement(subj, pred, obj,
 				includeInferred, contexts);
 		final RepositoryResult<Statement> result;
 		include = added.getStatements(subj, pred, obj, includeInferred,
 				contexts).asList().iterator();
 		result = getDelegate().getStatements(subj, pred, obj, includeInferred,
 				contexts);
 		if (!include.hasNext() && !excluded)
 			return result;
 		final CloseableIteration<Statement, RepositoryException> filtered;
 		if (excluded) {
 			filtered = new FilterIteration<Statement, RepositoryException>(
 					result) {
 				@Override
 				protected boolean accept(Statement stmt)
 						throws RepositoryException {
 					return !removed.hasStatement(stmt, includeInferred);
 				}
 			};
 		} else {
 			filtered = result;
 		}
 		CloseableIteration<Statement, RepositoryException> iter;
 		iter = new CloseableIteration<Statement, RepositoryException>() {
 			private Statement stmt;
 
 			public void close() throws RepositoryException {
 				filtered.close();
 			}
 
 			public boolean hasNext() throws RepositoryException {
 				return include.hasNext() || filtered.hasNext();
 			}
 
 			public Statement next() throws RepositoryException {
 				if (include.hasNext())
 					return stmt = include.next();
 				return stmt = filtered.next();
 			}
 
 			public void remove() throws RepositoryException {
 				FlushableConnection.this.remove(stmt);
 			}
 		};
 		return new RepositoryResult<Statement>(iter);
 	}
 
 	@Override
 	public boolean isEmpty() throws RepositoryException {
 		if (isDelegatingRead())
 			return super.isEmpty();
 		if (!added.isEmpty())
 			return false;
 		if (super.isEmpty())
 			return true;
 		if (removed.isEmpty())
 			return false;
 		if (super.size() > removed.size())
 			return false;
 		return super.isEmpty();
 	}
 
 	@Override
 	public long size(Resource... contexts) throws RepositoryException {
 		if (isDelegatingRead())
 			return super.size();
 		long size = super.size(contexts);
 		long rsize = removed.size(contexts);
 		long asize = added.size(contexts);
 		return size - rsize + asize;
 	}
 
 	@Override
 	protected synchronized void addWithoutCommit(Resource subject,
 			URI predicate, Value object, Resource... contexts)
 			throws RepositoryException {
 		if (!removed.hasStatement(subject, predicate, object, false, contexts)) {
 			added.add(subject, predicate, object, contexts);
 		}
 		removed.remove(subject, predicate, object, contexts);
 	}
 
 	@Override
 	protected synchronized void removeWithoutCommit(Resource subject,
 			URI predicate, Value object, Resource... contexts)
 			throws RepositoryException {
 		RDFHandler handler = new RDFInserter(removed);
 		try {
 			getDelegate().exportStatements(subject, predicate, object, true,
 					handler, contexts);
 		} catch (RDFHandlerException e) {
 			if (e.getCause() instanceof RepositoryException)
 				throw (RepositoryException) e.getCause();
 			throw new AssertionError(e);
 		}
 		added.remove(subject, predicate, object, contexts);
 	}
 
 	private void initUnFlushedStatements() throws RepositoryException {
 		Repository addRepo = new SailRepository(new MemoryStore());
 		Repository removeRepo = new SailRepository(new MemoryStore());
 		addRepo.initialize();
 		removeRepo.initialize();
 		added = addRepo.getConnection();
 		removed = removeRepo.getConnection();
 	}
 
 	private void destroyUnflushedStatements() throws RepositoryException {
 		Repository add = added.getRepository();
 		Repository remove = removed.getRepository();
 		added.close();
 		removed.close();
 		add.shutDown();
 		remove.shutDown();
 	}
 }