package marceline.storm.trident.clojure;

import backtype.storm.task.IMetricsContext;
import backtype.storm.utils.Utils;
import marceline.storm.trident.state.map.IInstrumentedMap;
import clojure.lang.IFn;
import clojure.lang.RT;

import java.util.List;
import java.util.ArrayList;
import java.util.Map;

public class ClojureInstrumentedMap implements IInstrumentedMap<Object> {
  List<Object> _params;
  List<String> _fnSpec;
  IInstrumentedMap _instrumentedMap;

  public ClojureInstrumentedMap(List fnSpec, List<Object> params) {
    _params = params;
    _fnSpec = fnSpec;

    try {
      IFn hof = Utils.loadClojureFn(_fnSpec.get(0), _fnSpec.get(1));
      _instrumentedMap = (IInstrumentedMap) hof.applyTo(RT.seq(_params));
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }

  @Override
  public List<Object> multiGet(List<List<Object>> keys) {
    return _instrumentedMap.multiGet(keys);
  }

  @Override
  public void multiPut(List<List<Object>> keys, List<Object> vals) {
    _instrumentedMap.multiPut(keys, vals);
  }

  @Override
  public void instrument(Map conf, IMetricsContext metrics) {
    _instrumentedMap.instrument(conf, metrics);
  }
}

@@ -79,16 +79,62 @@
 ontext;%0A
+import backtype.storm.metric.api.CountMetric;%0A
 import b
@@ -503,16 +503,65 @@
 edMap;%0A%0A
+  CountMetric _mreads;%0A  CountMetric _mwrites;%0A%0A%0A
   public
@@ -1005,16 +1005,25 @@
 Get(keys
+, _mreads
 );%0A  %7D%0A%0A
@@ -1143,16 +1143,26 @@
 ys, vals
+, _mwrites
 );%0A  %7D%0A%0A
@@ -1231,23 +1231,332 @@
  metrics
-) %7B%0A   
+, String mapStateMetricName, int bucketSize) %7B%0A    String metricBaseName = %22hambo/%22 + mapStateMetricName;%0A    _mreads = context.registerMetric(metricBaseName + %22/read-count%22, new CountMetric(), bucketSize);%0A    _mwrites = context.registerMetric(metricBaseName + %22/write-count%22, new CountMetric(), bucketSize);%0A    //
  _instru

define a couple of stack metrics