package org.jboss.cache.search.blackbox;

import org.apache.lucene.analysis.standard.StandardAnalyzer;
import org.apache.lucene.queryParser.ParseException;
import org.apache.lucene.queryParser.QueryParser;
import org.apache.lucene.search.Query;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jboss.cache.Cache;
import org.jboss.cache.DefaultCacheFactory;
import org.jboss.cache.Fqn;
import org.jboss.cache.config.Configuration;
import org.jboss.cache.search.CacheQuery;
import org.jboss.cache.search.SearchableCache;
import org.jboss.cache.search.SearchableCacheFactory;
import org.jboss.cache.search.helper.IndexCleanUp;
import org.jboss.cache.search.test.Person;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

import java.util.List;

/**
 * @author Navin Surtani (<a href="mailto:nsurtani@redhat.com">nsurtani@redhat.com</a>)
 */
@Test(groups = "functional")
public class ClusteredCacheTest
{
   Cache cache1, cache2;
   SearchableCache searchableCache1, searchableCache2;
   Person person1;
   Person person2;
   Person person3;
   Person person4;
   QueryParser queryParser;
   Query luceneQuery;
   CacheQuery cacheQuery;
   List found;
   String key1 = "Navin";
   String key2 = "BigGoat";
   String key3 = "MiniGoat";
   private static final Log log = LogFactory.getLog(Person.class);



   @BeforeMethod
   public void setUp() throws CloneNotSupportedException
   {
      Configuration cacheCfg = new Configuration();
      cacheCfg.setCacheMode(Configuration.CacheMode.REPL_SYNC);
      cacheCfg.setFetchInMemoryState(false);
      cache1 = new DefaultCacheFactory().createCache(cacheCfg.clone());
      searchableCache1 = new SearchableCacheFactory().createSearchableCache(cache1, Person.class);

      cache2 = new DefaultCacheFactory().createCache(cacheCfg.clone());
      searchableCache2 = new SearchableCacheFactory().createSearchableCache(cache2, Person.class);

      // wait until both caches are started and can see each other on the network.
      long giveUpTime = System.currentTimeMillis() + (60 * 1000); // give up after 1 minute of waiting
      boolean wait = true;
      while (wait && System.currentTimeMillis() < giveUpTime)
      {
         wait = !((cache1.getMembers().size() == 2) && (cache2.getMembers().size() == 2));
         // spin-lock
         try
         {
            Thread.sleep(100);
         }
         catch (InterruptedException e)
         {
            // do nothing
         }
      }

      if (wait)
      {
         // we still haven't got the cache views we want!!
         throw new RuntimeException("Unable to initialise cluster!");
      }


      person1 = new Person();
      person1.setName("Navin Surtani");
      person1.setBlurb("Likes playing WoW");

      person2 = new Person();
      person2.setName("BigGoat");
      person2.setBlurb("Eats grass");

      person3 = new Person();
      person3.setName("MiniGoat");
      person3.setBlurb("Eats cheese");

      //Put the 3 created objects in the searchableCache1.

      searchableCache1.put(Fqn.fromString("/a/b/c"), key1, person1);
      searchableCache1.put(Fqn.fromString("/a/b/d"), key2, person2);
      searchableCache1.put(Fqn.fromString("/a/b/c"), key3, person3);


   }

   @AfterMethod
   public void tearDown()
   {
      if (searchableCache1 != null) searchableCache1.stop();
      if (searchableCache2 != null) searchableCache2.stop();

      IndexCleanUp.cleanUpIndexes();

   }

   public void testSimple() throws ParseException
   {
      queryParser = new QueryParser("blurb", new StandardAnalyzer());
      luceneQuery = queryParser.parse("playing");
      cacheQuery = searchableCache2.createQuery(luceneQuery);

      found = cacheQuery.list();

      assert found.size() == 1;

      if(found.get(0) == null)
      {
         if(log.isTraceEnabled()) log.warn("found.get(0) is null");
         Person p1 = (Person) searchableCache2.get("/a/b/c", key1);
                         http://wiki.jboss.org/wiki/JBossCacheSearchable
         if(p1 == null)
         {
            if(log.isTraceEnabled()) log.warn("Person p1 is null in sc2 and cannot actually see the data of person1 in sc1");
         }

         else
         {
            if(log.isTraceEnabled()) log.trace("p1 name is  " + p1.getName());

         }
      }


      assert found.get(0).equals(person1);

   }

   public void testModified() throws ParseException
   {
      queryParser = new QueryParser("blurb", new StandardAnalyzer());
      luceneQuery = queryParser.parse("playing");
      cacheQuery = searchableCache2.createQuery(luceneQuery);

      found = cacheQuery.list();

      assert found.size() == 1;
      assert found.get(0).equals(person1);

      person1.setBlurb("Likes pizza");
      searchableCache1.put(Fqn.fromString("/a/b/c/"), "Navin", person1);


      queryParser = new QueryParser("blurb", new StandardAnalyzer());
      luceneQuery = queryParser.parse("pizza");
      cacheQuery = searchableCache2.createQuery(luceneQuery);

      found = cacheQuery.list();

      assert found.size() == 1;
      assert found.get(0).equals(person1);
   }

   public void testAdded() throws ParseException
   {
      queryParser = new QueryParser("blurb", new StandardAnalyzer());

      luceneQuery = queryParser.parse("eats");
      cacheQuery = searchableCache2.createQuery(luceneQuery);
      found = cacheQuery.list();

      System.out.println("found.size() is " + found.size());

      assert found.size() == 2 : "Size of list should be 2";
      assert found.contains(person2);
      assert found.contains(person3);
      assert !found.contains(person4) : "This should not contain object person4";

      person4 = new Person();
      person4.setName("Mighty Goat");
      person4.setBlurb("Also eats grass");

      searchableCache1.put(Fqn.fromString("/r/a/m/"), "Ram", person4);

      luceneQuery = queryParser.parse("eats");
      cacheQuery = searchableCache2.createQuery(luceneQuery);
      found = cacheQuery.list();

      assert found.size() == 3 : "Size of list should be 3";
      assert found.contains(person2);
      assert found.contains(person3);
      assert found.contains(person4) : "This should now contain object person4";
   }

   public void testRemoved() throws ParseException
   {
      queryParser = new QueryParser("blurb", new StandardAnalyzer());
      luceneQuery = queryParser.parse("eats");
      cacheQuery = searchableCache2.createQuery(luceneQuery);
      found = cacheQuery.list();

      assert found.size() == 2;
      assert found.contains(person2);
      assert found.contains(person3) : "This should still contain object person3";

      searchableCache1.remove(Fqn.fromString("/a/b/c/"), key3);

      queryParser = new QueryParser("blurb", new StandardAnalyzer());
      luceneQuery = queryParser.parse("eats");
      cacheQuery = searchableCache2.createQuery(luceneQuery);
      found = cacheQuery.list();

      assert found.size() == 1;
      assert found.contains(person2);
      assert !found.contains(person3) : "The search should not return person3";


   }


}


