code: package resa.evaluation.topology.fp;
  
  import backtype.storm.serialization.SerializableSerializer;
  import backtype.storm.task.OutputCollector;
  import backtype.storm.task.TopologyContext;
  import backtype.storm.topology.OutputFieldsDeclarer;
  import backtype.storm.topology.base.BaseRichBolt;
  import backtype.storm.tuple.Fields;
  import backtype.storm.tuple.Tuple;
  import com.esotericsoftware.kryo.DefaultSerializer;
  import com.esotericsoftware.kryo.Kryo;
  import com.esotericsoftware.kryo.KryoSerializable;
  import com.esotericsoftware.kryo.io.Input;
  import com.esotericsoftware.kryo.io.Output;
  import com.esotericsoftware.kryo.serializers.DefaultSerializers;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import resa.util.ConfigUtil;
  
  import java.io.Serializable;
  import java.util.*;
  
  /**
   * Created by ding on 14-6-5.
   */
  public class Detector extends BaseRichBolt implements Constant {
  
      private static final Logger LOG = LoggerFactory.getLogger(Detector.class);
  
      @DefaultSerializer(SerializableSerializer.class)
      public static class Entry implements Serializable {
          int count = 0;
          boolean detectedBySelf;
          int refCount = 0;
          boolean flagMFPattern = false;
          long timestamp;
  
          public Entry(long timestamp) {
              this.timestamp = timestamp;
          }
  
          public long setTimestamp(long timestamp) {
              long tmp = this.timestamp;
              this.timestamp = timestamp;
              return tmp;
          }
  
          public void setDetectedBySelf(boolean detectedBySelf) {
              this.detectedBySelf = detectedBySelf;
          }
  
          public boolean isDetectedBySelf() {
              return detectedBySelf;
          }
  
          public void setMFPattern(boolean flag) {
              this.flagMFPattern = flag;
          }
  
          public boolean isMFPattern() {
              return this.flagMFPattern;
          }
  
          int getCount() {
              return count;
          }
  
          int incCountAndGet() {
              return ++count;
          }
  
          int decCountAndGet() {
              return --count;
          }
  
          int getRefCount() {
              return refCount;
          }
  
          int incRefCountAndGet() {
              return ++refCount;
          }
  
          int decRefCountAndGet() {
              return --refCount;
          }
  
          boolean hasReference() {
              return this.refCount > 0;
          }
  
          boolean unused() {
              return count <= 0 && refCount <= 0;
          }
  
          String reportCnt() {
              return String.format(" cnt: %d, refCnt: %d", this.count, this.refCount);
          }
  
      }
  
      private PatternDB patterns;
      private int threshold;
      private OutputCollector collector;
      private List<Integer> targetTasks;
  
      @DefaultSerializer(DefaultSerializers.KryoSerializableSerializer.class)
      public static class PatternDB extends LinkedHashMap<WordList, Entry> implements KryoSerializable {
          private long maxKeep;
  
          public PatternDB(long maxKeep) {
              super(65536, 0.75f, true);
              this.maxKeep = maxKeep;
          }
  
          public PatternDB() {
              this(Long.MAX_VALUE);
          }
  
          @Override
          protected boolean removeEldestEntry(Map.Entry eldest) {
              return System.currentTimeMillis() - ((Entry) eldest.getValue()).timestamp > maxKeep;
          }
  
          public void removeExpired(long now) {
              for (Iterator<Map.Entry<WordList, Entry>> iter = entrySet().iterator(); iter.hasNext(); ) {
                  Map.Entry<WordList, Entry> e = iter.next();
                  if (now - e.getValue().timestamp > maxKeep) {
                      iter.remove();
                  } else {
                      return;
                  }
              }
          }
  
          @Override
          public void write(Kryo kryo, Output output) {
              output.writeLong(maxKeep);
              output.writeInt(size());
              output.writeLong(System.currentTimeMillis());
              forEach((k, v) -> {
                  kryo.writeClassAndObject(output, k);
                  kryo.writeClassAndObject(output, v);
              });
              LOG.info("write out {} patterns", size());
          }
  
          @Override
          public void read(Kryo kryo, Input input) {
              maxKeep = Long.MAX_VALUE;
              long maxKeepTmp = input.readLong();
              int size = input.readInt();
              long last = input.readLong();
              // rest timestamp
              for (int i = 0; i < size; i++) {
                  WordList p = (WordList) kryo.readClassAndObject(input);
                  Entry entry = (Entry) kryo.readClassAndObject(input);
                  put(p, entry);
              }
              long toAdd = System.currentTimeMillis() - last + 10000;
              forEach((k, v) -> v.setTimestamp(v.timestamp + toAdd));
              maxKeep = maxKeepTmp;
              LOG.info("read in {} patterns", size);
          }
      }
  
      @Override
      public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
          String patternData = "pattern";
          this.patterns = (PatternDB) context.getTaskData(patternData);
          if (this.patterns == null) {
              long maxKeepInterval = ConfigUtil.getInt(stormConf, MAX_KEEP_PROP, 60000);
              context.setTaskData(patternData, (this.patterns = new PatternDB(maxKeepInterval)));
          }
          this.collector = collector;
          this.threshold = ConfigUtil.getInt(stormConf, THRESHOLD_PROP, 20);
          targetTasks = context.getComponentTasks(context.getThisComponentId());
          Collections.sort(targetTasks);
          LOG.info("In Detector, threshold: " + threshold);
      }
  
      /////////////////// State Transition Graph, Implementation III/////////////////////////////
      /// States: ( Cnt > Threshold ? , hasReference? | isMaxFreqPattern? ) Stable state      ///
      ///         [   ,   |  ] Temp State -->  trigger update                                 ///
      /// Event and Output: Cnt (+/-), pattern count inc/dec (DefaultStream)                  ///
      ///                   RefCnt (+/-), reference count update from (Feedback Stream)       ///
      ///                   Output to next bolt: *T*/*F*                                      ///
      ///                                                                                     ///
      ///     +--(Cnt-)---(F, F | F)<--(direct update, *F*)--[F, F | T]                       ///
      ///     |              |   ^                                ^                           ///
      ///     |     (RefCnt-)|   | (RefCnt+)                (Cnt-)|                           ///
      ///     |              V   |                                |                           ///
      ///     |           (F, T | F )                        (T, F | T)<--------------+       ///
      ///     |              |   ^                                |                   |       ///
      ///     |        (Cnt-)|   | (Cnt+)                         |(RefCnt+)          |       ///
      ///     |              V   |                                V                   |       ///
      ///     |           (T, T | F)<--(direct update, *F*)--[T, T | T]               |       ///
      ///     |              |   ^                                                    |       ///
      ///     |     (RefCnt-)|   | (RefCnt+)                                          |       ///
      ///     |              V   |                                                    |       ///
      ///     +---------->[T, F | F ]----------->(direct update, *T*)-----------------+       ///
      ///                                                                                     ///
      ///////////////////////////////////////////////////////////////////////////////////////////
  
      @Override
      public void execute(Tuple input) {
          //doneTODO:
          //WordList pattern = (WordList) input.getValueByField(PATTERN_FIELD);
          final long now = System.currentTimeMillis();
          ArrayList<WordList> wordListArrayList = (ArrayList<WordList>) input.getValueByField(PATTERN_FIELD);
  
          wordListArrayList.forEach((pattern) -> {
  
              Entry entry = patterns.computeIfAbsent(pattern, (k) -> new Entry(now));
  
              if (!input.getSourceStreamId().equals(FEEDBACK_STREAM)) {
                  ///Pattern Count Stream, only affect pattern count
                  ///We only change IncRef and DecRef at two events: [++count == threshold] and [--count == threshold-1]
                  if (input.getBooleanByField(IS_ADD_FIELD)) {
                      entry.incCountAndGet();
                      LOG.debug(
                              "In DetectorNew(default), cntInc: " + pattern + "," + entry.reportCnt());
                      if (entry.getCount() == threshold) {
                          incRefToSubPatternExcludeSelf(pattern.getWords(), collector, input);
                          LOG.debug(
                                  "In DetectorNew(default), cntInc: " + pattern + ",satisfy thresh and incRef");
                      }
                  } else {
                      entry.decCountAndGet();
                      LOG.debug(
                              "In DetectorNew(default), cntDec: " + pattern + "," + entry.reportCnt());
                      if (entry.getCount() == threshold - 1) {
                          decRefToSubPatternExcludeSelf(pattern.getWords(), collector, input);
                          LOG.debug(
                                  "In DetectorNew(default), cntDec: " + pattern + ",dissatisfy thresh and decRef");
                      }
                  }
  
                  ///We separate the action of update refCount and update states
                  if (!entry.isMFPattern()) {///entry.isMFPattern == false
                      if (entry.getCount() >= threshold && !entry.hasReference()) {
                          ///State (F, F | F) -> (T, F | T),
                          ///State (T, F | F) -> (T, F | T),
                          ///[output pattern, T]
                          entry.setMFPattern(true);
                          collector.emit(REPORT_STREAM, input, Arrays.asList(pattern, true));
                          LOG.debug(
                                  "In DetectorNew(default), set isMFP" + pattern + "," + entry.reportCnt());
                      }
                  } else {///entry.isMFPattern == true
                      if (entry.hasReference() || (!entry.hasReference() && entry.getCount() < threshold)) {
                          ///State (T, T | T) -> (F, T | F)
                          ///State (T, T | T) -> (T, T | F)
                          ///State (T, F | T) -> (F, F | F)
                          ///[output pattern, F]
                          entry.setMFPattern(false);
                          collector.emit(REPORT_STREAM, input, Arrays.asList(pattern, false));
  
                          LOG.debug(
                                  "In DetectorNew(default), cancel isMFP" + pattern + "," + entry.reportCnt());
                      }
                  }
              } else {
                  ///Feedback_STREAM, only affect refCount, also check states.
                  ///State (F, F | F) <--> (F, T | F)
                  ///State (T, F | F) <--> (T, T | F)
                  ///State (T, F | T) <--> (T, T | T)
                  if (input.getBooleanByField(IS_ADD_FIELD)) {
                      entry.incRefCountAndGet();
                      LOG.debug(
                              "In DetectorNew(FB), incReferenceCnt: " + pattern + ", " + entry.reportCnt());
                  } else {
                      entry.decRefCountAndGet();
                      LOG.debug(
                              "In DetectorNew(FB), DecReferenceCnt: " + pattern + ", " + entry.reportCnt());
                  }
  
                  ///TODO: please double check if we need to anchor tuple here?
                  ///update states
                  if (entry.hasReference() && entry.isMFPattern()) {
                      ///State [*, T | T] --> (*, T | F), update states and output F
                      entry.setMFPattern(false);
                      collector.emit(REPORT_STREAM, input, Arrays.asList(pattern, false));
                      LOG.debug(
                              "In DetectorNew(FB), cancel isMFP: " + pattern + ", " + entry.reportCnt());
                  } else if (!entry.hasReference() && !entry.isMFPattern() && entry.getCount() >= threshold) {
                      ///State [T, F | F] --> (T, F | T) update states and output T
                      entry.setMFPattern(true);
                      collector.emit(REPORT_STREAM, input, Arrays.asList(pattern, true));
                      LOG.debug(
                              "In DetectorNew(FB), set isMFP" + pattern + "," + entry.reportCnt());
                  }
              }
              if (entry.unused()) {
                  patterns.remove(pattern);
              } else {
                  entry.setTimestamp(now);
              }
          });
          patterns.removeExpired(now);
          sleep(wordListArrayList.size());
          collector.ack(input);
      }
  
      private static void sleep(long t) {
          long t1 = System.currentTimeMillis();
          do {
              for (int i = 0; i < 10; i++) {
                  Math.atan(Math.sqrt(Math.random() * Integer.MAX_VALUE));
              }
          } while (System.currentTimeMillis() - t1 < t);
      }
  
      private void incRefToSubPatternExcludeSelf(int[] wordIds, OutputCollector collector, Tuple input) {
          adjRefToSubPatternExcludeSelf(wordIds, collector, input, true);
      }
  
      private void decRefToSubPatternExcludeSelf(int[] wordIds, OutputCollector collector, Tuple input) {
          adjRefToSubPatternExcludeSelf(wordIds, collector, input, false);
      }
  
      private void adjRefToSubPatternExcludeSelf(int[] wordIds, OutputCollector collector, Tuple input, boolean adj) {
          int n = wordIds.length;
          int[] buffer = new int[n];
          ArrayList<WordList>[] wordListForTargetTask = new ArrayList[targetTasks.size()];
          ///Note that here we exclude itself as one of the sub-patterns
          ///for (int i = 1; i < (1 << n); i++) {
          for (int i = 1; i < (1 << n) - 1; i++) {
              int k = 0;
              for (int j = 0; j < n; j++) {
                  if ((i & (1 << j)) > 0) {
                      buffer[k++] = wordIds[j];
                  }
              }
              //doneTODO:
              //collector.emit(FEEDBACK_STREAM, input, Arrays.asList(new WordList(Arrays.copyOf(buffer, k)), adj));
              WordList wl = new WordList(Arrays.copyOf(buffer, k));
              int targetIndex = WordList.getPartition(targetTasks.size(), wl);
              if (wordListForTargetTask[targetIndex] == null) {
                  wordListForTargetTask[targetIndex] = new ArrayList<>();
              }
              wordListForTargetTask[targetIndex].add(wl);
          }
          for (int i = 0; i < wordListForTargetTask.length; i++) {
              if (wordListForTargetTask[i] != null && wordListForTargetTask[i].size() > 0) {
                  collector.emitDirect(
                          targetTasks.get(i),
                          FEEDBACK_STREAM,
                          input, Arrays.asList(wordListForTargetTask[i], adj));
              }
          }
      }
  
      @Override
      public void declareOutputFields(OutputFieldsDeclarer declarer) {
          declarer.declareStream(REPORT_STREAM, new Fields(PATTERN_FIELD, IS_ADD_MFP));
          //declarer.declareStream(FEEDBACK_STREAM, new Fields(PATTERN_FIELD, IS_ADD_FIELD));
          //doneTODO: add true for direct grouping
          declarer.declareStream(FEEDBACK_STREAM, true, new Fields(PATTERN_FIELD, IS_ADD_FIELD));
      }
  }
  