/**
 * Put your copyright and license info here.
 */
package com.example.myapexapp;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Map;

import org.apache.commons.lang.mutable.MutableInt;
import org.apache.hadoop.conf.Configuration;

import com.datatorrent.api.annotation.ApplicationAnnotation;
import com.datatorrent.contrib.kafka.KafkaSinglePortStringInputOperator;
import com.datatorrent.api.StreamingApplication;
import com.datatorrent.api.DAG;
import com.datatorrent.api.DefaultInputPort;
import com.datatorrent.api.DefaultOutputPort;
import com.datatorrent.lib.db.jdbc.AbstractJdbcTransactionableOutputOperator;
import com.datatorrent.lib.db.jdbc.JdbcTransactionalStore;
import com.datatorrent.lib.io.ConsoleOutputOperator;
import com.datatorrent.lib.io.IdempotentStorageManager;
import com.datatorrent.lib.util.BaseUniqueKeyCounter;
import com.datatorrent.lib.util.KeyValPair;

@ApplicationAnnotation(name="ExactlyOnceExampleApplication")
public class Application implements StreamingApplication
{

  @Override
  public void populateDAG(DAG dag, Configuration conf)
  {
    KafkaSinglePortStringInputOperator kafkaInput = dag.addOperator("kafkaInput", new KafkaSinglePortStringInputOperator());
    kafkaInput.setIdempotentStorageManager(new IdempotentStorageManager.FSIdempotentStorageManager());
    UniqueCounter<String> count = dag.addOperator("count", new UniqueCounter<String>());
    CountStoreOperator store = dag.addOperator("store", new CountStoreOperator());
    store.setStore(new JdbcTransactionalStore());
    ConsoleOutputOperator cons = dag.addOperator("console", new ConsoleOutputOperator());
    dag.addStream("words", kafkaInput.outputPort, count.data);
    dag.addStream("counts", count.count, store.input, cons.input);
  }

  public static class CountStoreOperator extends AbstractJdbcTransactionableOutputOperator<KeyValPair<String, Integer>>
  {
    public static final String SQL =
        "MERGE INTO words USING (VALUES ?, ?) I (word, wcount)"
        + " ON (words.word=I.word)"
        + " WHEN MATCHED THEN UPDATE SET words.wcount = words.wcount + I.wcount"
        + " WHEN NOT MATCHED THEN INSERT (word, wcount) VALUES (I.word, I.wcount)";

    @Override
    protected String getUpdateCommand()
    {
      return SQL;
    }

    @Override
    protected void setStatementParameters(PreparedStatement statement, KeyValPair<String, Integer> tuple) throws SQLException
    {
      statement.setString(1, tuple.getKey());
      statement.setInt(2, tuple.getValue());
    }
  }

  public static class UniqueCounter<K> extends BaseUniqueKeyCounter<K>
  {
    /**
     * The input port which receives incoming tuples.
     */
    public final transient DefaultInputPort<K> data = new DefaultInputPort<K>()
    {
      /**
       * Reference counts tuples
       */
      @Override
      public void process(K tuple)
      {
        processTuple(tuple);
      }

    };

    public final transient DefaultOutputPort<KeyValPair<K, Integer>> count = new DefaultOutputPort<KeyValPair<K, Integer>>()
    {
      @Override
      public Unifier<KeyValPair<K, Integer>> getUnifier()
      {
        throw new UnsupportedOperationException("not partitionable");
      }
    };

    @Override
    public void endWindow()
    {
      for (Map.Entry<K, MutableInt> e: map.entrySet()) {
        count.emit(new KeyValPair<>(e.getKey(), e.getValue().toInteger()));
      }
      map.clear();
    }

  }

}

@@ -561,32 +561,79 @@
 aultOutputPort;%0A
+import com.datatorrent.lib.algo.UniqueCounter;%0A
 import com.datat
@@ -1398,24 +1398,20 @@
 eCounter
-%3CString%3E
+Flat
  count =
@@ -1457,16 +1457,12 @@
 nter
-%3CString%3E
+Flat
 ());
@@ -1787,16 +1787,17 @@
 nt.count
+s
 , store.
@@ -2616,19 +2616,20 @@
 eCounter
-%3CK%3E
+Flat
  extends
@@ -2633,29 +2633,22 @@
 nds 
-Base
 Unique
-Key
 Counter%3C
 K%3E%0A 
@@ -2647,329 +2647,20 @@
 ter%3C
-K%3E%0A  %7B%0A    /**%0A     * The input port which receives incoming tuples.%0A     */%0A    public final transient DefaultInputPort%3CK%3E data = new DefaultInputPort%3CK%3E()%0A 
+String%3E%0A
   
- 
 %7B%0A
-      /**%0A       * Reference counts tuples%0A       */%0A      @Override%0A      public void process(K tuple)%0A      %7B%0A        processTuple(tuple);%0A      %7D%0A%0A    %7D;%0A%0A
     
@@ -2703,33 +2703,38 @@
 Port%3CKeyValPair%3C
-K
+String
 , Integer%3E%3E coun
@@ -2734,16 +2734,17 @@
 %3E%3E count
+s
  = new D
@@ -2764,205 +2764,11 @@
 ort%3C
-KeyValPair%3CK, Integer%3E%3E()%0A    %7B%0A      @Override%0A      public Unifier%3CKeyValPair%3CK, Integer%3E%3E getUnifier()%0A      %7B%0A        throw new UnsupportedOperationException(%22not partitionable%22);%0A      %7D%0A    %7D
+%3E()
 ;%0A%0A 
@@ -2835,17 +2835,22 @@
 p.Entry%3C
-K
+String
 , Mutabl
@@ -2889,16 +2889,17 @@
    count
+s
 .emit(ne
@@ -2986,16 +2986,15 @@
 ;%0A    %7D%0A
-%0A
   %7D%0A%0A%7D%0A

Use algo/UniqueCounter to reduce code