/*
 *  Copyright (c) 2012-2013 Malhar, Inc.
 *  All Rights Reserved.
 */
package com.malhartech.lib.util;

import java.util.HashMap;
import java.util.Map;
import java.util.TreeMap;
import org.apache.commons.lang.mutable.MutableDouble;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author David Yan <davidyan@malhar-inc.com>
 */
public class DimensionTimeBucketSumOperator extends DimensionTimeBucketOperator
{
  private static final Logger LOG = LoggerFactory.getLogger(DimensionTimeBucketSumOperator.class);
  private Map<String, Map<String, Number>> dataMap = new HashMap<String, Map<String, Number>>();

  @Override
  public void process(String timeBucket, String key, String field, Number value)
  {
    String finalKey = timeBucket + "|" + key;
    Map<String, Number> m = dataMap.get(finalKey);
    if (value == null) {
      return;
    }
    if (m == null) {
      m = new HashMap<String, Number>();
      m.put(field, new MutableDouble(value));
      dataMap.put(finalKey, m);
    }
    else {
      Number n = m.get(field);
      if (n == null) {
        m.put(field, new MutableDouble(value));
      } else {
        ((MutableDouble)n).add(value);
      }
    }
  }

  @Override
  public void endWindow()
  {
    if (!dataMap.isEmpty()) {
      out.emit(dataMap);
      LOG.info("Number of keyval pairs: {}", dataMap.size());
    }
  }

}

@@ -535,16 +535,26 @@
 private 
+transient 
 Map%3CStri
@@ -589,53 +589,8 @@
 aMap
- = new HashMap%3CString, Map%3CString, Number%3E%3E()
 ;%0A%0A 
@@ -1159,16 +1159,136 @@
  %7D%0A  %7D%0A%0A
+  @Override%0A  public void beginWindow(long windowId)%0A  %7B%0A    dataMap = new HashMap%3CString, Map%3CString, Number%3E%3E();%0A  %7D%0A%0A
   @Overr

make dataMap transient and refreshed to a new one for every window