package com.hammurapi.common.store.tests;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import com.hammurapi.common.SimpleObservableConverter;
import com.hammurapi.common.concurrent.LocalTrackingExecutorService;
import com.hammurapi.common.extract.tests.Account;
import com.hammurapi.common.extract.tests.AccountNumberExtractor;
import com.hammurapi.common.extract.tests.ObservableAccount;
import com.hammurapi.common.extract.tests.Transaction;
import com.hammurapi.extract.AbstractExtractor;
import com.hammurapi.extract.AbstractPredicate;
import com.hammurapi.extract.Constant;
import com.hammurapi.extract.Extractor;
import com.hammurapi.extract.ExtractorFactory;
import com.hammurapi.extract.LessEqual;
import com.hammurapi.extract.Predicate;
import com.hammurapi.store.Index;
import com.hammurapi.store.MultiIndex;
import com.hammurapi.store.Store;
import com.hammurapi.store.Store.Handle;
import com.hammurapi.store.Store.ViewType;
import com.hammurapi.store.StoreException;
import com.hammurapi.store.local.LocalStore;
import com.hammurapi.store.local.LocalStoreBase.Config;
import com.hammurapi.store.local.LocalStoreImpl;

public class LocalStoreTests {		
	
	private static final String HAMMURAPI_GROUP = "Hammurapi Group";
	private ExecutorService executorService;

	@Before
	public void setUp() throws Exception {
		executorService = Executors.newFixedThreadPool(10);
	}

	@After
	public void tearDown() throws Exception {
		executorService.shutdown();
	}
	
	@Test
	public void testPutGet() {
		Config<Account, Integer, LocalStore<Account, Integer>> config = new Config<Account, Integer, LocalStore<Account, Integer>>();
		config.setPrimaryKeyExtractor(new AccountNumberExtractor());
		config.setExecutorService(executorService);
		LocalStore<Account, Integer> store = new LocalStoreImpl<Account,Integer>(config);
		
		Account account = new Account(HAMMURAPI_GROUP, 123, null);
		store.put(account);
		assertSame(account, store.getByPrimaryKey(123));				
	}
	
	@Test
	public void testDuplicate() {
		Config<Account, Integer, LocalStore<Account, Integer>> config = new Config<Account, Integer, LocalStore<Account, Integer>>();
		config.setPrimaryKeyExtractor(new AccountNumberExtractor());
		config.setExecutorService(executorService);		
		LocalStore<Account, Integer> store = new LocalStoreImpl<Account,Integer>(config);
		
		Account account = new Account(HAMMURAPI_GROUP, 123, null);
		store.put(account);
		
		Account account2 = new Account(HAMMURAPI_GROUP, 123, null);
		store.put(account2);
		Account result = store.getByPrimaryKey(123);
		assertNotSame(account, result);
		assertSame(account2, result);		
		
	}
	
	@Test
	public void testGetAll() {
		Config<Account, Integer, LocalStore<Account, Integer>> config = new Config<Account, Integer, LocalStore<Account, Integer>>();
		config.setPrimaryKeyExtractor(new AccountNumberExtractor());
		config.setExecutorService(executorService);		
		LocalStore<Account, Integer> store = new LocalStoreImpl<Account,Integer>(config);
		
		store.put(new Account(HAMMURAPI_GROUP, 123, null));		
		store.put(new Account(HAMMURAPI_GROUP, 456, null));

		int cnt = 0;
		for (Account a: store) {
			++cnt;
		}
		
		assertEquals(cnt, 2);		
	}
	
	private Extractor<Account, BigDecimal, LocalStore<Account, Integer>> balanceExtractor = new AbstractExtractor<Account, BigDecimal, LocalStore<Account, Integer>>(0, null, false, 0) {

		@Override
		protected BigDecimal extractInternal(
				LocalStore<Account, Integer> context,
				Map<LocalStore<Account, Integer>, Map<Extractor<Account, ? super BigDecimal, LocalStore<Account, Integer>>, ? super BigDecimal>> cache,
				Account... obj) {
			return obj[0].getBalance();
		}
		
	};
	
	@Test
	public void testGet() {
		Config<Account, Integer, LocalStore<Account, Integer>> config = new Config<Account, Integer, LocalStore<Account, Integer>>();
		config.setPrimaryKeyExtractor(new AccountNumberExtractor());
		config.setExecutorService(executorService);		
		LocalStore<Account, Integer> store = new LocalStoreImpl<Account,Integer>(config);
		
		store.put(new Account(HAMMURAPI_GROUP, 123, new BigDecimal(1580.00)));		
		store.put(new Account(HAMMURAPI_GROUP, 456, new BigDecimal(330.67)));		
		
		int cnt = 0;
		Constant<Account, BigDecimal, LocalStore<Account, Integer>> constant = new Constant<Account, BigDecimal, LocalStore<Account, Integer>>(new BigDecimal(330.67));
		Predicate<Account, LocalStore<Account, Integer>> balancePredicate = new LessEqual<Account, BigDecimal, LocalStore<Account, Integer>>(-1, null, balanceExtractor, constant);
		for (Account a: store.get(balancePredicate)) {
			++cnt;
			assertEquals(constant.getValue(), a.getBalance());
		}
		
		assertEquals(cnt, 1);		
	}
	
	@Test
	public void testValidator() {
		Config<Account, Integer, LocalStore<Account, Integer>> config = new Config<Account, Integer, LocalStore<Account, Integer>>();
		config.setPrimaryKeyExtractor(new AccountNumberExtractor());
		config.setExecutorService(executorService);		
		LocalStore<Account, Integer> store = new LocalStoreImpl<Account,Integer>(config);
		
		final boolean[] ok = {true};
		
		Predicate<Account, LocalStore<Account, Integer>> validator = new AbstractPredicate<Account, LocalStore<Account, Integer>>(0, null, false, 0) {

			@Override
			protected Boolean extractInternal(
					LocalStore<Account, Integer> context,
					Map<LocalStore<Account, Integer>, Map<Extractor<Account, ? super Boolean, LocalStore<Account, Integer>>, ? super Boolean>> cache,
					Account... obj) {
				return ok[0];
			}
		};		
		
		store.put(new Account(HAMMURAPI_GROUP, 123, new BigDecimal(1580.00)), validator);		
		store.put(new Account(HAMMURAPI_GROUP, 456, new BigDecimal(330.67)));		

		// Verify two accounts in the store.
		int cnt =  0;
		for (Account a: store.getAll()) {
			++cnt;
		}
		assertEquals(2, cnt);
		
		// Invalidate the first one
		ok[0] = false;
		
		cnt =  0;
		for (Account a: store.getAll()) {
			++cnt;
		}
		assertEquals(1, cnt);
		
	}
	
	@Test
	public void testCaching() {
		Config<Account, Integer, LocalStore<Account, Integer>> config = new Config<Account, Integer, LocalStore<Account, Integer>>();
		config.setPrimaryKeyExtractor(new AccountNumberExtractor());
		config.setExecutorService(executorService);		
		LocalStore<Account, Integer> store = new LocalStoreImpl<Account,Integer>(config);
		
		Account account = new Account(HAMMURAPI_GROUP, 123, new BigDecimal(1580.00));
		Handle<Account, Integer, LocalStore<Account, Integer>> handle = store.put(account);
		
		Extractor<Account,BigDecimal, LocalStore<Account, Integer>> balanceExtractor = ExtractorFactory.INSTANCE.createExtractor(
				"java",
				"account.getBalance()", 
				new String[] {"account"}, 
				new Class[] {Account.class},
				BigDecimal.class, 
				Store.class, 
				this.getClass().getClassLoader());

		assertNotNull("Extractor is null, check classpath for java extractors", balanceExtractor);
		BigDecimal balance = handle.extract(balanceExtractor);
		assertEquals(new BigDecimal(1580.00), balance);
		assertEquals(1, account.getNumberOfBalanceInquiries());
		balance = handle.extract(balanceExtractor);
		assertEquals(new BigDecimal(1580.00), balance);
		assertEquals(1, account.getNumberOfBalanceInquiries());
	}
	
	@Test
	public void testObservable() {
		Config<Account, Integer, LocalStore<Account, Integer>> config = new Config<Account, Integer, LocalStore<Account, Integer>>();
		config.setPrimaryKeyExtractor(new AccountNumberExtractor());
		config.setExecutorService(executorService);		
		config.setObservableConverter(new SimpleObservableConverter<Account>());
		LocalStore<Account, Integer> store = new LocalStoreImpl<Account,Integer>(config);
		
		ObservableAccount account = new ObservableAccount(HAMMURAPI_GROUP, 123, new BigDecimal(1580.00));
		Handle<Account, Integer, LocalStore<Account, Integer>> handle = store.put(account);
		assertEquals(1, account.countObservers());		
		
		Extractor<Account,BigDecimal, LocalStore<Account, Integer>> balanceExtractor = ExtractorFactory.INSTANCE.createExtractor(
				"java",
				"account.getBalance()", 
				new String[] {"account"}, 
				new Class[] {Account.class},
				BigDecimal.class, 
				Store.class, 
				this.getClass().getClassLoader());
		
		assertNotNull("Extractor is null, check classpath for java extractors", balanceExtractor);
		BigDecimal balance = handle.extract(balanceExtractor);
		assertEquals(new BigDecimal(1580.00), balance);
		assertEquals(1, account.getNumberOfBalanceInquiries());
		balance = handle.extract(balanceExtractor);
		assertEquals(new BigDecimal(1580.00), balance);
		assertEquals(1, account.getNumberOfBalanceInquiries());
		
		// Update balance, retrieve
		account.setBalance(new BigDecimal(33.00));
		BigDecimal newBalance = handle.extract(balanceExtractor);
		assertEquals(new BigDecimal(33.00), newBalance);
		assertEquals(2, account.getNumberOfBalanceInquiries());
		
		store.remove(account);
		assertEquals(0, account.countObservers());
	}
	
	@Test
	public void testNoCaching() {
		Config<Account, Integer, LocalStore<Account, Integer>> config = new Config<Account, Integer, LocalStore<Account, Integer>>();
		config.setPrimaryKeyExtractor(new AccountNumberExtractor());
		config.setExecutorService(executorService);
		config.setCacheExtracted(false);
		LocalStore<Account, Integer> store = new LocalStoreImpl<Account,Integer>(config);
		
		Account account = new Account(HAMMURAPI_GROUP, 123, new BigDecimal(1580.00));
		Handle<Account, Integer, LocalStore<Account, Integer>> handle = store.put(account);
		
		Extractor<Account,BigDecimal, LocalStore<Account, Integer>> balanceExtractor = ExtractorFactory.INSTANCE.createExtractor(
				"jxpath",
				"$account/balance", 
				new String[] {"account"}, 
				new Class[] {Account.class},
				BigDecimal.class, 
				store.getClass(), 
				this.getClass().getClassLoader());

		assertNotNull("Extractor is null, check classpath for java extractors", balanceExtractor);
		BigDecimal balance = handle.extract(balanceExtractor);
		assertEquals(new BigDecimal(1580.00), balance);
		assertEquals(1, account.getNumberOfBalanceInquiries());
		balance = handle.extract(balanceExtractor);
		assertEquals(new BigDecimal(1580.00), balance);
		assertEquals(2, account.getNumberOfBalanceInquiries());
	}
	
	@Test
	public void testGetWithExtractAndSort() {
		Config<Account, Integer, LocalStore<Account, Integer>> config = new Config<Account, Integer, LocalStore<Account, Integer>>();
		config.setPrimaryKeyExtractor(new AccountNumberExtractor());
		config.setExecutorService(executorService);		
		LocalStore<Account, Integer> store = new LocalStoreImpl<Account,Integer>(config);
		
		store.put(new Account(HAMMURAPI_GROUP, 789, new BigDecimal(23000.33)));		
		store.put(new Account(HAMMURAPI_GROUP, 888, new BigDecimal(31580.00)));		
		store.put(new Account(HAMMURAPI_GROUP, 123, new BigDecimal(1580.00)));		
		store.put(new Account(HAMMURAPI_GROUP, 456, new BigDecimal(330.67)));
		
		// Extract by account number, order by balance.
		Predicate<Account, LocalStore<Account, Integer>> predicate = ExtractorFactory.INSTANCE.createPredicate(
				"javascript",
				"account.getNumber()!=123", 
				new String[] {"account"}, 
				new Class[] {Account.class},
				store.getClass(), 
				this.getClass().getClassLoader());
		
		assertNotNull("Predicate is null, check language name", predicate);
		
		Extractor<Account,BigDecimal, LocalStore<Account, Integer>> balanceExtractor = ExtractorFactory.INSTANCE.createExtractor(
				"jxpath",
				"$account/balance", 
				new String[] {"account"}, 
				new Class[] {Account.class},
				BigDecimal.class, 
				store.getClass(), 
				this.getClass().getClassLoader());
		
		assertNotNull(balanceExtractor);

		List<BigDecimal> results = new ArrayList<BigDecimal>();
		for (BigDecimal balance: store.get(predicate, balanceExtractor, true, null)) {
			results.add(balance);
		}

		assertEquals(3, results.size());
		assertEquals(new BigDecimal(330.67), results.get(0));
		assertEquals(new BigDecimal(23000.33), results.get(1));		
		assertEquals(new BigDecimal(31580.00), results.get(2));				
	}
	
	private class TestSet {
		
		private BigDecimal minTransactionAmount;
		private BigDecimal maxTransactionAmount;
		
		private BigDecimal minAccountBalance;
		private BigDecimal maxAccountBalance; 
		
		private int numberOfTransactions;
		
		private Collection<Account> accounts = new LinkedList<Account>();
		
		public TestSet(int numberOfAccunts, int numberOfTransactionsPerAccount) {
			Random random = new Random(System.currentTimeMillis());
			for (int i=0; i<numberOfAccunts; ++i) {
				BigDecimal accountBalance = new BigDecimal(random.nextDouble()*1000000);
				Account account = new Account(HAMMURAPI_GROUP, random.nextInt(1000000), accountBalance);
				accounts.add(account);
				if (minAccountBalance==null || minAccountBalance.compareTo(accountBalance)>0) {
					minAccountBalance = accountBalance;
				}
				if (maxAccountBalance==null || maxAccountBalance.compareTo(accountBalance)<0) {
					maxAccountBalance = accountBalance;
				}
				for (int j=0, l=random.nextInt(numberOfTransactionsPerAccount*2); j<l; ++j) {
					Transaction t = new Transaction(false);
					BigDecimal transactionAmount = new BigDecimal(random.nextDouble()*1000);
					t.setAmount(transactionAmount);
					t.setDate(new Date(System.currentTimeMillis()+ random.nextInt(10000000)));
					account.getTransactionsImmediately().add(t);
					if (minTransactionAmount==null || minTransactionAmount.compareTo(transactionAmount)>0) {
						minTransactionAmount = transactionAmount;
					}
					if (maxTransactionAmount==null || maxTransactionAmount.compareTo(transactionAmount)<0) {
						maxTransactionAmount = transactionAmount;
					}
					
					++numberOfTransactions;
				}
			}
		}

		Collection<Account> getAccounts() {
			return accounts;
		}

		public BigDecimal getMinTransactionAmount() {
			return minTransactionAmount;
		}

		public BigDecimal getMaxTransactionAmount() {
			return maxTransactionAmount;
		}

		public BigDecimal getMinAccountBalance() {
			return minAccountBalance;
		}

		public BigDecimal getMaxAccountBalance() {
			return maxAccountBalance;
		}

		public int getNumberOfTransactions() {
			return numberOfTransactions;
		}
		
		
	}
	
	@Test
	public void testGetMulti() {
		Config<Account, Integer, LocalStore<Account, Integer>> config = new Config<Account, Integer, LocalStore<Account, Integer>>();
		config.setPrimaryKeyExtractor(new AccountNumberExtractor());
		config.setExecutorService(executorService);		
		LocalStore<Account, Integer> store = new LocalStoreImpl<Account,Integer>(config);
		
		TestSet testSet = new TestSet(20, 100);
		for (Account a: testSet.getAccounts()) {
			store.put(a);			
		}
		
		// Select transactions from accounts with more than average balance ordered transaction amount.
		final BigDecimal averageBalance = testSet.getMinAccountBalance().add(testSet.getMaxAccountBalance()).divide(new BigDecimal(2));
		
		Predicate<Account, LocalStore<Account, Integer>> selector = new AbstractPredicate<Account, LocalStore<Account, Integer>>(0, null, false, 0) {

			@Override
			protected Boolean extractInternal(
					LocalStore<Account, Integer> context,
					Map<LocalStore<Account, Integer>, Map<Extractor<Account, ? super Boolean, LocalStore<Account, Integer>>, ? super Boolean>> cache,
					Account... obj) {
				return obj[0].getBalance().compareTo(averageBalance)>0;
			}
		};
						
		Extractor<Account, Iterable<Transaction>, LocalStore<Account, Integer>>  extractor = new AbstractExtractor<Account, Iterable<Transaction>, LocalStore<Account, Integer>>(0, null, false, 0) {

			@Override
			protected Iterable<Transaction> extractInternal(
					LocalStore<Account, Integer> context,
					Map<LocalStore<Account, Integer>, Map<Extractor<Account, ? super Iterable<Transaction>, LocalStore<Account, Integer>>, ? super Iterable<Transaction>>> cache,
					Account... obj) {
				return obj[0].getTransactions();
			}
		};
		
		Comparator<Transaction> comparator = new Comparator<Transaction>() {
			
			@Override
			public int compare(Transaction o1, Transaction o2) {
				return o1.getAmount().compareTo(o2.getAmount());
			}
			
		};
		
		long start = System.currentTimeMillis();
		
		// Verify that transactions are ordered and that the total number of transactions
		// is more than 0 and less than total.
		BigDecimal lastAmount = null;
		int totalTransactions = 0;
		for (Transaction t: store.getMultiple(selector, extractor, null, true, comparator)) {
			if (lastAmount!=null) {
				assertTrue("Transaction amounts are not ordered", lastAmount.compareTo(t.getAmount())<=0);
			}
			lastAmount = t.getAmount();
			++totalTransactions;
		}
		assertTrue("All transactions included", totalTransactions<testSet.getNumberOfTransactions());
		assertTrue("No transactions are included", totalTransactions>0);
		
		long period = System.currentTimeMillis() - start;
		
		// Verify parallelism
		// We invoke getBalance() - 100 ms, getTransactions() - 150 ms.
		long singleThreadedTime = testSet.getAccounts().size()*250;
		// We have 10 threads, so total time should be at least 5 times less than single threaded time.
		assertTrue("Execution took too long, single-threaded?", period<singleThreadedTime/5);		
	}
	
	@Test
	public void testQuery() {
		Config<Account, Integer, LocalStore<Account, Integer>> config = new Config<Account, Integer, LocalStore<Account, Integer>>();
		config.setPrimaryKeyExtractor(new AccountNumberExtractor());
		config.setExecutorService(executorService);		
		LocalStore<Account, Integer> store = new LocalStoreImpl<Account,Integer>(config);
		
		TestSet testSet = new TestSet(20, 100);
		for (Account a: testSet.getAccounts()) {
			store.put(a);			
		}
		
		// Select transactions from accounts with more than average balance ordered transaction amount.
		final BigDecimal averageBalance = testSet.getMinAccountBalance().add(testSet.getMaxAccountBalance()).divide(new BigDecimal(2));
		
		Predicate<Account, LocalStore<Account, Integer>> selector = new AbstractPredicate<Account, LocalStore<Account, Integer>>(0, null, false, 0) {

			@Override
			protected Boolean extractInternal(
					LocalStore<Account, Integer> context,
					Map<LocalStore<Account, Integer>, Map<Extractor<Account, ? super Boolean, LocalStore<Account, Integer>>, ? super Boolean>> cache,
					Account... obj) {
				return obj[0].getBalance().compareTo(averageBalance)>0;
			}
		};
						
		long start = System.currentTimeMillis();
		
		final AtomicInteger totalTransactions = new AtomicInteger();
		final AtomicInteger totalAccounts = new AtomicInteger();
		
		Store.QueryTask<Account, Integer, LocalStore<Account, Integer>> qTask = new Store.QueryTask<Account, Integer, LocalStore<Account, Integer>>() {
			
			@Override
			public void execute(LocalStore<Account, Integer> store, Account obj, Integer primaryKey) {
				totalTransactions.addAndGet(obj.getTransactions().size());
				totalAccounts.incrementAndGet();
			}
		};
		
		store.query(selector, qTask);
		assertTrue("All transactions included", totalTransactions.get()<testSet.getNumberOfTransactions());
		assertTrue("No transactions are included", totalTransactions.get()>0);
		
		long period = System.currentTimeMillis() - start;
		
		// Verify parallelism
		// We invoke getBalance() - 100 ms, getTransactions() - 150 ms and for each account query.execute()/account.getTransactions() - 150 ms.
		long singleThreadedTime = testSet.getAccounts().size()*250+totalAccounts.get()*150;
		// We have 10 threads, so total time should be at least 5 times less than single threaded time.
		assertTrue("Execution took too long, single-threaded?", period<singleThreadedTime/5);		
	}
	
	@Test
	public void testMultiQuery() {
		Config<Account, Integer, LocalStore<Account, Integer>> config = new Config<Account, Integer, LocalStore<Account, Integer>>();
		config.setPrimaryKeyExtractor(new AccountNumberExtractor());
		config.setExecutorService(executorService);		
		LocalStore<Account, Integer> store = new LocalStoreImpl<Account,Integer>(config);
		
		TestSet testSet = new TestSet(20, 100);
		for (Account a: testSet.getAccounts()) {
			store.put(a);			
		}
		
		// Select transactions from accounts with more than average balance ordered transaction amount.
		final BigDecimal averageBalance = testSet.getMinAccountBalance().add(testSet.getMaxAccountBalance()).divide(new BigDecimal(2));
		
		Predicate<Account, LocalStore<Account, Integer>> selector = new AbstractPredicate<Account, LocalStore<Account, Integer>>(0, null, false, 0) {

			@Override
			protected Boolean extractInternal(
					LocalStore<Account, Integer> context,
					Map<LocalStore<Account, Integer>, Map<Extractor<Account, ? super Boolean, LocalStore<Account, Integer>>, ? super Boolean>> cache,
					Account... obj) {
				return obj[0].getBalance().compareTo(averageBalance)>0;
			}
		};
						
		Extractor<Account, Iterable<Transaction>, LocalStore<Account, Integer>>  extractor = new AbstractExtractor<Account, Iterable<Transaction>, LocalStore<Account, Integer>>(0, null, false, 0) {

			@Override
			protected Iterable<Transaction> extractInternal(
					LocalStore<Account, Integer> context,
					Map<LocalStore<Account, Integer>, Map<Extractor<Account, ? super Iterable<Transaction>, LocalStore<Account, Integer>>, ? super Iterable<Transaction>>> cache,
					Account... obj) {
				return obj[0].getTransactions();
			}
		};
		
		long start = System.currentTimeMillis();
		
		final AtomicInteger totalTransactions = new AtomicInteger();
		
		Store.QueryTask<Transaction, Integer, LocalStore<Account, Integer>> qTask = new Store.QueryTask<Transaction, Integer, LocalStore<Account, Integer>>() {
			
			@Override
			public void execute(LocalStore<Account, Integer> store, Transaction obj, Integer primaryKey) {
				try {
					Thread.sleep(150);
				} catch (InterruptedException e) {
					throw new StoreException(e);
				}
				totalTransactions.incrementAndGet();				
			}
		};

		store.queryMultiple(selector, extractor, null, qTask);
		
		assertTrue("All transactions included", totalTransactions.get()<testSet.getNumberOfTransactions());
		assertTrue("No transactions are included", totalTransactions.get()>0);
		
		long period = System.currentTimeMillis() - start;
		
		// Verify parallelism
		// We invoke getBalance() - 100 ms, getTransactions() - 150 ms and for each transaction query.execute() - 100 ms.
		long singleThreadedTime = testSet.getAccounts().size()*250+totalTransactions.get()*150;
		// We have 10 threads, so total time should be at least 5 times less than single threaded time.
		assertTrue("Execution took too long, single-threaded?", period<singleThreadedTime/5);		
	}
	
	// Test update 
	
	@Test
	public void testUpdate() {
		Config<Account, Integer, LocalStore<Account, Integer>> config = new Config<Account, Integer, LocalStore<Account, Integer>>();
		config.setPrimaryKeyExtractor(new AccountNumberExtractor());
		config.setExecutorService(executorService);		
		LocalStore<Account, Integer> store = new LocalStoreImpl<Account,Integer>(config);
		
		TestSet testSet = new TestSet(20, 100);
		for (Account a: testSet.getAccounts()) {
			store.put(a);			
		}
		
		// Select transactions from accounts with more than average balance ordered transaction amount.
		final BigDecimal averageBalance = testSet.getMinAccountBalance().add(testSet.getMaxAccountBalance()).divide(new BigDecimal(2));
		
		Predicate<Account, LocalStore<Account, Integer>> selector = new AbstractPredicate<Account, LocalStore<Account, Integer>>(0, null, false, 0) {

			@Override
			protected Boolean extractInternal(
					LocalStore<Account, Integer> context,
					Map<LocalStore<Account, Integer>, Map<Extractor<Account, ? super Boolean, LocalStore<Account, Integer>>, ? super Boolean>> cache,
					Account... obj) {
				return obj[0].getBalance().compareTo(averageBalance)>0;
			}
		};
						
		long start = System.currentTimeMillis();
		
		final AtomicInteger totalTransactions = new AtomicInteger();
		final AtomicInteger totalAccounts = new AtomicInteger();
		
		Store.UpdateTask<Account, Integer, LocalStore<Account, Integer>> qTask = new Store.UpdateTask<Account, Integer, LocalStore<Account, Integer>>() {
			
			@Override
			public Runnable execute(LocalStore<Account, Integer> store, Handle<Account, Integer, LocalStore<Account, Integer>> handle) {
				Account account = handle.get();
				totalTransactions.addAndGet(account.getTransactions().size());
				totalAccounts.incrementAndGet();
				
				// We create a new account with the same number, no transactions and increased balance.
				Account newAccount = new Account(HAMMURAPI_GROUP, account.getNumber(), account.getBalance().add(new BigDecimal(333)));
				handle.update(newAccount);
				return null;
			}
		};
		
		store.update(selector, qTask);
		assertTrue("All transactions included", totalTransactions.get()<testSet.getNumberOfTransactions());
		assertTrue("No transactions are included", totalTransactions.get()>0);
		
		long period = System.currentTimeMillis() - start;
		
		// Verify parallelism
		// We invoke getBalance() - 100 ms, getTransactions() - 150 ms and for each account query.execute()/account.getTransactions() - 150 ms.
		long singleThreadedTime = testSet.getAccounts().size()*250+totalAccounts.get()*150;
		// We have 10 threads, so total time should be at least 5 times less than single threaded time.
		assertTrue("Execution took too long, single-threaded?", period<singleThreadedTime/5);		
	}
	
	
	@Test
	public void testView() {
		Config<Account, Integer, LocalStore<Account, Integer>> config = new Config<Account, Integer, LocalStore<Account, Integer>>();
		config.setPrimaryKeyExtractor(new AccountNumberExtractor());
		config.setExecutorService(executorService);		
		LocalStore<Account, Integer> store = new LocalStoreImpl<Account,Integer>(config);
		
		store.put(new Account(HAMMURAPI_GROUP, 123, null));		
		store.put(new Account(HAMMURAPI_GROUP, 456, null));
		store.put(new Account("Pavel Vlasov", 789, null));
		
		Predicate<Account, LocalStore<Account, Integer>> viewPredicate = new AbstractPredicate<Account, LocalStore<Account, Integer>>(0, null, false, 0) {

			@Override
			protected Boolean extractInternal(
					LocalStore<Account, Integer> context,
					Map<LocalStore<Account, Integer>, Map<Extractor<Account, ? super Boolean, LocalStore<Account, Integer>>, ? super Boolean>> cache,
					Account... obj) {
				return HAMMURAPI_GROUP.equals(obj[0].getCustomer());
			}
			 
		};
		
		LocalStore<Account, Integer> view = store.createView(viewPredicate, ViewType.LIVE);

		int cnt = 0;
		for (Account a: view) {
			++cnt;
			assertEquals(a.getCustomer(), HAMMURAPI_GROUP);
		}
		
		assertEquals(cnt, 2);		
	}
	
	@Test(expected=StoreException.class)
	public void testViewPut() {
		Config<Account, Integer, LocalStore<Account, Integer>> config = new Config<Account, Integer, LocalStore<Account, Integer>>();
		config.setPrimaryKeyExtractor(new AccountNumberExtractor());
		config.setExecutorService(executorService);		
		LocalStore<Account, Integer> store = new LocalStoreImpl<Account,Integer>(config);
		
		Predicate<Account, LocalStore<Account, Integer>> viewPredicate = new AbstractPredicate<Account, LocalStore<Account, Integer>>(0, null, false, 0) {

			@Override
			protected Boolean extractInternal(
					LocalStore<Account, Integer> context,
					Map<LocalStore<Account, Integer>, Map<Extractor<Account, ? super Boolean, LocalStore<Account, Integer>>, ? super Boolean>> cache,
					Account... obj) {
				return HAMMURAPI_GROUP.equals(obj[0].getCustomer());
			}
			 
		};
		
		LocalStore<Account, Integer> view = store.createView(viewPredicate, ViewType.LIVE);
		view.put(new Account("Pavel Vlasov", 789, null));
	}
	
	@Test(expected=StoreException.class)
	public void testUniqueIndex() {
		Config<Account, Integer, LocalStore<Account, Integer>> config = new Config<Account, Integer, LocalStore<Account, Integer>>();
		config.setPrimaryKeyExtractor(new AccountNumberExtractor());
		config.setExecutorService(executorService);		
		LocalStore<Account, Integer> store = new LocalStoreImpl<Account,Integer>(config);
		
		store.put(new Account(HAMMURAPI_GROUP, 123, null));
		
		Extractor<Account, String, LocalStore<Account, Integer>> customerExtractor = new AbstractExtractor<Account, String, LocalStore<Account, Integer>>(0, null, false, 0) {

			@Override
			protected String extractInternal(
					LocalStore<Account, Integer> context,
					Map<LocalStore<Account, Integer>, Map<Extractor<Account, ? super String, LocalStore<Account, Integer>>, ? super String>> cache,
					Account... obj) {
				return obj[0].getCustomer();
			}
			
		};
		
		store.addIndex(null, customerExtractor, Index.Type.UNIQUE, false, null);
		
		store.put(new Account(HAMMURAPI_GROUP, 456, null));
	}
	
	@Test
	public void testIndex() {
		Config<Account, Integer, LocalStore<Account, Integer>> config = new Config<Account, Integer, LocalStore<Account, Integer>>();
		config.setPrimaryKeyExtractor(new AccountNumberExtractor());
		config.setExecutorService(executorService);		
		LocalStore<Account, Integer> store = new LocalStoreImpl<Account,Integer>(config);
		
		store.put(new Account(HAMMURAPI_GROUP, 123, new BigDecimal(100)));		
		store.put(new Account(HAMMURAPI_GROUP, 456, new BigDecimal(10)));
		store.put(new Account("Pavel Vlasov", 789, new BigDecimal(1)));
		
		Predicate<Account, LocalStore<Account, Integer>> indexPredicate = new AbstractPredicate<Account, LocalStore<Account, Integer>>(0, null, false, 0) {

			@Override
			protected Boolean extractInternal(
					LocalStore<Account, Integer> context,
					Map<LocalStore<Account, Integer>, Map<Extractor<Account, ? super Boolean, LocalStore<Account, Integer>>, ? super Boolean>> cache,
					Account... obj) {
				return HAMMURAPI_GROUP.equals(obj[0].getCustomer());
			}
			 
		};
				
		MultiIndex<Account, Account, Integer, BigDecimal, LocalStore<Account, Integer>> index = (MultiIndex<Account, Account, Integer, BigDecimal, LocalStore<Account, Integer>>) store.addIndex(indexPredicate, balanceExtractor, Index.Type.ASYNCHRONOUS, true, null);
				
		// Using index to access data
		int cnt = 0;
		BigDecimal prevBal = null;
		for (Account a: index) {
			++cnt;
			assertEquals(a.getCustomer(), HAMMURAPI_GROUP);
			if (prevBal!=null) {
				assertTrue(a.getBalance().compareTo(prevBal)>0);
			}
			prevBal = a.getBalance();
		}
						
		assertEquals(cnt, 2);
		
		cnt = 0;
		for (Account a: index.find(new BigDecimal(10))) {
			assertEquals(new BigDecimal(10), a.getBalance());
			++cnt;
		}
		
		assertEquals(1, cnt);
		
		cnt = 0;
		for (Account a: index.find(new BigDecimal(12345))) {
			assertEquals(new BigDecimal(12345), a.getBalance());
			++cnt;
		}
		
		assertEquals(0, cnt);
		
		// Doing get, use of index is verified by checking that results are ordered.
		cnt = 0;
		prevBal = null;
		for (Account a: store.get(indexPredicate)) {
			++cnt;
			assertEquals(a.getCustomer(), HAMMURAPI_GROUP);
			if (prevBal!=null) {
				assertTrue(a.getBalance().compareTo(prevBal)>0);
			}
			prevBal = a.getBalance();
		}
		
		assertEquals(cnt, 2);	
		
	}
	
	/**
	 * Tests that no tasks are executed after join() returns.
	 * @throws InterruptedException
	 */
	@Test
	public void testLocalTrackingServiceJoin() throws InterruptedException {
		
		final AtomicBoolean afterJoin = new AtomicBoolean(false);
		final AtomicInteger failures = new AtomicInteger();		
		final AtomicInteger completed = new AtomicInteger();
		final Random random = new Random(System.currentTimeMillis());
		final AtomicInteger createdTasks = new AtomicInteger();
		
		final LocalTrackingExecutorService ltes = new LocalTrackingExecutorService(executorService, false, null);
		final int numberOfTasks = 5;
		
		class TestTask implements Runnable {
			
			private int depth;

			public TestTask(long delay, int depth) {
				createdTasks.incrementAndGet();
				this.delay = delay;
				this.depth = depth;
			}
			
			long delay;
			
			@Override
			public void run() {
				try {
					if (depth>0) {
						for (int i=0; i<numberOfTasks; ++i) {
							ltes.submit(new TestTask(random.nextInt(100), depth-1));
						}
					}
					Thread.sleep(delay);
					if (afterJoin.get()) {
						failures.incrementAndGet();
					}
				} catch (Exception e) {
					e.printStackTrace();
				} finally {
					completed.incrementAndGet();
				}
			}
		};
		
		
		int depth = 3;
		for (int i=0; i<numberOfTasks; ++i) {
			ltes.submit(new TestTask(random.nextInt(100), depth));
		}
		
		ltes.join();
		
		assertEquals(createdTasks.get(), completed.get());
		assertEquals(0, failures.get());				

	}
}
