package org.infinispan.cdi.test.distexec;

import static org.infinispan.cdi.test.testutil.Deployments.baseDeployment;

import java.util.Iterator;
import java.util.Map;
import java.util.StringTokenizer;

import javax.inject.Inject;

import org.infinispan.Cache;
import org.infinispan.cdi.Input;
import org.infinispan.cdi.test.DefaultTestEmbeddedCacheManagerProducer;
import org.infinispan.distexec.mapreduce.BaseWordCountMapReduceTest;
import org.infinispan.distexec.mapreduce.Collector;
import org.infinispan.distexec.mapreduce.MapReduceTask;
import org.infinispan.distexec.mapreduce.Mapper;
import org.infinispan.distexec.mapreduce.Reducer;
import org.infinispan.distexec.mapreduce.SimpleTwoNodesMapReduceTest;
import org.infinispan.test.MultipleCacheManagersTest;
import org.jboss.arquillian.container.test.api.Deployment;
import org.jboss.shrinkwrap.api.Archive;
import org.testng.Assert;
import org.testng.annotations.Test;

/**
 * BaseTest for MapReduceTask
 * 
 * @author Vladimir Blagojevic
 */
@Test(enabled = true, groups = "functional", testName = "cdi.test.distexec.WordCountMapReduceCDITest")
public class WordCountMapReduceCDITest extends MultipleCacheManagersArquillianTest {

   BaseWordCountMapReduceTest delegate;

   public WordCountMapReduceCDITest() {
      delegate = new SimpleTwoNodesMapReduceTest();
   }
   
   @Override
   MultipleCacheManagersTest getDelegate() {
      return delegate;
   }

   @Deployment
   public static Archive<?> deployment() {
      return baseDeployment().addClass(WordCountMapReduceCDITest.class)
            .addClass(DefaultTestEmbeddedCacheManagerProducer.class);
   }

   public void testinvokeMapReduceOnSubsetOfKeys() throws Exception {
      MapReduceTask<String, String, String, Integer> task = delegate.invokeMapReduce(new String[] {
               "1", "2", "3" }, new WordCountMapper(), new WordCountReducer());
      Map<String, Integer> mapReduce = task.execute();
      Integer count = mapReduce.get("Infinispan");
      assert count == 1;
      count = mapReduce.get("Boston");
      assert count == 1;
   }
   
   public void testinvokeMapReduceWithInputCacheOnSubsetOfKeys() throws Exception {
      MapReduceTask<String, String, String, Integer> task = delegate.invokeMapReduce(new String[] {
               "1", "2", "3" }, new WordCountImpliedInputCacheMapper(), new WordCountReducer());
      Map<String, Integer> mapReduce = task.execute();
      Integer count = mapReduce.get("Infinispan");
      assert count == 1;
      count = mapReduce.get("Boston");
      assert count == 1;
   }

   private static class WordCountMapper implements Mapper<String, String, String, Integer> {
      /** The serialVersionUID */
      private static final long serialVersionUID = -5943370243108735560L;

      @Inject
      private Cache<String, String> cache;

      @Override
      public void map(String key, String value, Collector<String, Integer> collector) {
         Assert.assertNotNull(cache, "Cache not injected into " + this);
         StringTokenizer tokens = new StringTokenizer(value);
         while (tokens.hasMoreElements()) {
            String s = (String) tokens.nextElement();
            collector.emit(s, 1);
         }
      }
   }
   
   private static class WordCountImpliedInputCacheMapper implements Mapper<String, String, String, Integer> {
    
      /** The serialVersionUID */
      private static final long serialVersionUID = 7525403183805551028L;
      
      @Inject
      @Input
      private Cache<String, String> cache;

      @Override
      public void map(String key, String value, Collector<String, Integer> collector) {
         Assert.assertNotNull(cache, "Cache not injected into " + this);
         //the right cache injected         
         Assert.assertTrue(cache.getName().equals("mapreducecache")); 
         StringTokenizer tokens = new StringTokenizer(value);
         while (tokens.hasMoreElements()) {
            String s = (String) tokens.nextElement();
            collector.emit(s, 1);
         }
      }
   }

   private static class WordCountReducer implements Reducer<String, Integer> {
      /** The serialVersionUID */
      private static final long serialVersionUID = 1901016598354633256L;

      @Inject
      private Cache<String, String> cache;

      @Override
      public Integer reduce(String key, Iterator<Integer> iter) {
         Assert.assertNotNull(cache, "Cache not injected into " + this);
         int sum = 0;
         while (iter.hasNext()) {
            Integer i = iter.next();
            sum += i;
         }
         return sum;
      }
   }
}
