public class Flow implements Serializable{

    String id;
    String name;
    String description;

    Map<String,StreamDef> streams = new HashMap<String, StreamDef>();

    public String getId() {
        return id;
    }

    public String getName() {
        return name;
    }

    public String getDescription() {
        return description;
    }

    public Collection<StreamDef> getStreams() {
        return streams.values();
    }

    public StreamDef getStream(String name) {
        return streams.get(name);
    }

    public void setId(String id) {
        this.id = id;
    }

    public void setName(String name) {
        this.name = name;
    }

    public void setDescription(String description) {
        this.description = description;
    }

    public void setStreams(List<StreamDef> streams) {
        for(StreamDef def : streams)
            this.streams.put(def.getName(), def);
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;

        Flow flow = (Flow) o;

        if (description != null ? !description.equals(flow.description) : flow.description != null) return false;
        if (id != null ? !id.equals(flow.id) : flow.id != null) return false;
        if (name != null ? !name.equals(flow.name) : flow.name != null) return false;
        if (streams != null ? !streams.equals(flow.streams) : flow.streams != null) return false;

        return true;
    }

    @Override
    public int hashCode() {
        int result = id != null ? id.hashCode() : 0;
        result = 31 * result + (name != null ? name.hashCode() : 0);
        result = 31 * result + (description != null ? description.hashCode() : 0);
        result = 31 * result + (streams != null ? streams.hashCode() : 0);
        return result;
    }

    @Override
    public String toString() {
        return "Flow{" +
                "id='" + id + '\'' +
                ", name='" + name + '\'' +
                ", description='" + description + '\'' +
                ", streams=" + streams +
                '}';
    }
} public class FlowInfo {

  private String flowId;
  private Event event;
  private int idx;
  private String streamName;
  private String previousStream;
  private String partition;

  public FlowInfo(Tuple tuple) {
    flowId = tuple.getStringByField(FLOW_ID);
    event = (Event) tuple.getValueByField(EVENT);
    idx = tuple.getIntegerByField(FLOW_OP_IDX);
    idx++;
    streamName = tuple.getStringByField(STREAM_NAME);
    previousStream = tuple.getStringByField(LAST_STREAM);

    if(tuple.contains(PARTITION))
      partition = tuple.getStringByField(PARTITION);
  }

  public FlowInfo(String flowId, String stream, int idx) {
    this.flowId = flowId;
    this.streamName = stream;
    this.idx = idx;
  }

  public String getFlowId() {
    return flowId;
  }

  public Event getEvent() {
    return event;
  }

  public int getIdx() {
    return idx;
  }

  public String getStreamName() {
    return streamName;
  }

  public String getPreviousStream() {
    return previousStream;
  }

  public String getPartition() {
    return partition;
  }
} public enum Policy implements Serializable {

    COUNT, TIME, TIME_DELTA_LT
} public class StreamDef implements Serializable {

    private String name;
    private List<FlowOp> flowOps;
    private boolean stdInput;
    private boolean stdOutput;
    private String[] outputs;

    public StreamDef(String name, List<FlowOp> flowOps, boolean stdInput, boolean stdOutput, String[] outputs) {
      this.name = name;
      this.flowOps = flowOps;
      this.stdInput = stdInput;
      this.stdOutput = stdOutput;
      this.outputs = outputs;
    }

    public String getName() {
        return name;
    }

    public List<FlowOp> getFlowOps() {
        return flowOps;
    }

    public boolean isStdInput() {
      return stdInput;
    }

    public boolean isStdOutput() {
      return stdOutput;
    }

    public String[] getOutputs() {
      return outputs;
    }

    @Override
    public String toString() {
        return "StreamDef{" +
                "name='" + name + '\'' +
                ", flowOps=" + flowOps +
                '}';
    }

    @Override
    public boolean equals(Object o) {
      if (this == o) {
        return true;
      }
      if (o == null || getClass() != o.getClass()) {
        return false;
      }

      StreamDef streamDef = (StreamDef) o;

      if (stdInput != streamDef.stdInput) {
        return false;
      }
      if (stdOutput != streamDef.stdOutput) {
        return false;
      }
      if (flowOps != null ? !flowOps.equals(streamDef.flowOps) : streamDef.flowOps != null) {
        return false;
      }
      if (name != null ? !name.equals(streamDef.name) : streamDef.name != null) {
        return false;
      }
      if (!Arrays.equals(outputs, streamDef.outputs)) {
        return false;
      }

      return true;
    }

    @Override
    public int hashCode() {
      int result = name != null ? name.hashCode() : 0;
      result = 31 * result + (flowOps != null ? flowOps.hashCode() : 0);
      result = 31 * result + (stdInput ? 1 : 0);
      result = 31 * result + (stdOutput ? 1 : 0);
      result = 31 * result + (outputs != null ? Arrays.hashCode(outputs) : 0);
      return result;
    }
} public class AggregatedEvent {

  private Event event;
  private String previousStream;

  public AggregatedEvent(Event event, String previousStream) {
    this.event = event;
    this.previousStream = previousStream;
  }

  public AggregatedEvent(Event event) {
    this.event = event;
  }

  public Event getEvent() {
    return event;
  }

  public String getPreviousStream() {
    return previousStream;
  }

  @Override
  public boolean equals(Object o) {
    if (this == o) return true;
    if (o == null || getClass() != o.getClass()) return false;

    AggregatedEvent that = (AggregatedEvent) o;

    if (event != null ? !event.equals(that.event) : that.event != null) return false;
    if (previousStream != null ? !previousStream.equals(that.previousStream) : that.previousStream != null)
      return false;

    return true;
  }

  @Override
  public int hashCode() {
    int result = event != null ? event.hashCode() : 0;
    result = 31 * result + (previousStream != null ? previousStream.hashCode() : 0);
    return result;
  }

  @Override
  public String toString() {
    return "AggregatedEvent{" +
            "event=" + event +
            ", previousStream='" + previousStream + '\'' +
            '}';
  }
} public class AggregateOp implements FlowOp, RequiresPartitioning {

    public static final String AGGREGATE = "aggregate";

    private final Class<? extends Aggregator> aggregatorClass;
    private final Policy triggerPolicy;
    private final Policy evictionPolicy;
    private final long triggerThreshold;
    private final long evictionThreshold;
    private final boolean clearOnTrigger;

    private long windowEvictMillis;

    private Map<String,String> config;

    public AggregateOp(Class<? extends Aggregator> aggregatorClass, Policy triggerPolicy, long triggerThreshold,
                       Policy evictionPolicy, long evictionThreshold, Map<String,String> config, boolean clearOnTrigger,
                       long windowEvictMillis) {
        this.aggregatorClass = aggregatorClass;
        this.triggerPolicy = triggerPolicy;
        this.evictionPolicy = evictionPolicy;
        this.triggerThreshold = triggerThreshold;
        this.evictionThreshold = evictionThreshold;
        this.config = config;
        this.clearOnTrigger = clearOnTrigger;
        this.windowEvictMillis = windowEvictMillis;
    }

    public boolean isClearOnTrigger() {
      return clearOnTrigger;
    }

    public Class<? extends Aggregator> getAggregatorClass() {
        return aggregatorClass;
    }

    public Policy getTriggerPolicy() {
        return triggerPolicy;
    }

    public Policy getEvictionPolicy() {
        return evictionPolicy;
    }

    public long getTriggerThreshold() {
        return triggerThreshold;
    }

    public long getEvictionThreshold() {
        return evictionThreshold;
    }

    public Map<String, String> getConfig() {
      return config;
    }

    public long getWindowEvictMillis() {
      return windowEvictMillis;
    }

    @Override
    public String getComponentName() {
        return AGGREGATE;
    }
} public interface FlowOp extends Serializable {

    String getComponentName();
} public class PartitionOp implements FlowOp {

  public static final String PARTITION = "partition";
  private List<String> fields;

  public PartitionOp(List<String> fields) {
      this.fields = fields;
  }

  public List<String> getFields() {
      return fields;
  }

  @Override
  public String getComponentName() {
      return PARTITION;
  }

  @Override
  public String toString() {
    return "PartitionOp{" +
            "fields=" + fields +
            '}';
  }
} public interface Aggregator extends Serializable {

    public static final String GROUP_BY = "groupBy";
    public static final String GROUP_BY_DELIM = "\u0000";

    void configure(Map<String, String> configuration);

    void added(WindowItem item);

    void evicted(WindowItem item);

    List<AggregatedEvent> aggregate();
} public class AggregatorWindow extends Window {

    private Aggregator aggregator;

    public AggregatorWindow(Aggregator aggregator, String groupedIndex, long size) {
      events = new AggregatorLimitingDeque(size, aggregator);
      this.groupedIndex = groupedIndex;
      this.aggregator = aggregator;
    }

    public AggregatorWindow(Aggregator aggregator, String groupedIndex) {
        super(groupedIndex);
        this.aggregator = aggregator;
    }

    @Override
    public WindowItem add(Event event, String previousStream) {
        WindowItem item = super.add(event, previousStream);
        aggregator.added(item);
        return item;
    }

    @Override
    public WindowItem expire() {
        WindowItem item = super.expire();
        aggregator.evicted(item);
        return item;
    }

    @Override
    public void clear() {
      while(size() > 0)
        expire();
    }

  /**
   * Used for age-based expiration
   */
  public void timeEvict(long thresholdInSeconds) {
    while(events != null && events.peek() != null &&
            (System.currentTimeMillis() - events.peek().getTimestamp()) >= (thresholdInSeconds * 1000)) {
      WindowItem item = events.poll();
      aggregator.evicted(item);
    }
  }
  public Collection<AggregatedEvent> getAggregate() {
        return aggregator.aggregate();
    }
} public static void declareOutputStreams(OutputFieldsDeclarer declarer, Fields fields) {
    declarer.declareStream(PARTITION, fields);
    declarer.declareStream(FILTER, fields);
    declarer.declareStream(SELECT, fields);
    declarer.declareStream(AGGREGATE, fields);
    declarer.declareStream(SWITCH, fields);
    declarer.declareStream(SORT, fields);
    declarer.declareStream(JOIN, fields);
    declarer.declareStream(SPLIT, fields);
    declarer.declareStream(EACH, fields);
    declarer.declareStream(OUTPUT, fields);

    declarer.declareStream(BROADCAST_STREAM + PARTITION, fields);
    declarer.declareStream(BROADCAST_STREAM + FILTER, fields);
    declarer.declareStream(BROADCAST_STREAM + SELECT, fields);
    declarer.declareStream(BROADCAST_STREAM + AGGREGATE, fields);
    declarer.declareStream(BROADCAST_STREAM + SWITCH, fields);
    declarer.declareStream(BROADCAST_STREAM + SORT, fields);
    declarer.declareStream(BROADCAST_STREAM + JOIN, fields);
    declarer.declareStream(BROADCAST_STREAM + EACH, fields);
    declarer.declareStream(BROADCAST_STREAM + SPLIT, fields);
} public static Fields fields = new Fields(FLOW_ID, EVENT, FLOW_OP_IDX, STREAM_NAME, LAST_STREAM); public static final String FLOW_LOADER_STREAM = "flowLoaderStream"; public static boolean exportsToOtherStreams(Flow flow, String streamName, String nextStream) {
  return nextStream.equals(OUTPUT) && flow.getStream(streamName).getOutputs() != null;
} public static <T extends FlowOp>T getFlowOpFromStream(Flow flow, String stream, int idx) {
  return (T) flow.getStream(stream).getFlowOps().get(idx);
} public static boolean hasNextOutput(Flow flow, String streamName, String nextStream) {
  return (nextStream.equals(OUTPUT) && flow.getStream(streamName).isStdOutput()) || !nextStream.equals(OUTPUT);
}
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
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.Tuple;
import backtype.storm.tuple.Values;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import org.calrissian.flowmix.api.Flow;
import org.calrissian.flowmix.core.model.FlowInfo;
import org.calrissian.flowmix.api.Policy;
import org.calrissian.flowmix.core.model.StreamDef;
import org.calrissian.flowmix.core.model.event.AggregatedEvent;
import org.calrissian.flowmix.core.model.op.AggregateOp;
import org.calrissian.flowmix.core.model.op.FlowOp;
import org.calrissian.flowmix.core.model.op.PartitionOp;
import org.calrissian.flowmix.api.Aggregator;
import org.calrissian.flowmix.core.support.window.AggregatorWindow;
import static org.apache.commons.lang.StringUtils.join;
import static org.calrissian.flowmix.api.builder.FlowmixBuilder.declareOutputStreams;
import static org.calrissian.flowmix.api.builder.FlowmixBuilder.fields;
import static org.calrissian.flowmix.core.Constants.FLOW_LOADER_STREAM;
import static org.calrissian.flowmix.api.Aggregator.GROUP_BY;
import static org.calrissian.flowmix.api.Aggregator.GROUP_BY_DELIM;
import static org.calrissian.flowmix.core.support.Utils.exportsToOtherStreams;
import static org.calrissian.flowmix.core.support.Utils.getFlowOpFromStream;
import static org.calrissian.flowmix.core.support.Utils.hasNextOutput;
/*
 * Copyright (C) 2014 The Calrissian Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.calrissian.flowmix.core.storm.bolt;




public class AggregatorBolt extends BaseRichBolt {

    Map<String,Flow> flows;
    Map<String, Cache<String, AggregatorWindow>> windows;
    OutputCollector collector;

    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.collector = outputCollector;
        flows = new HashMap<String, Flow>();
        windows = new HashMap<String, Cache<String, AggregatorWindow>>();
    }

    @Override
    public void execute(Tuple tuple) {

        if(FLOW_LOADER_STREAM.equals(tuple.getSourceStreamId())) {
            for(Flow flow : (Collection<Flow>)tuple.getValue(0))
                flows.put(flow.getId(), flow);
        } else if("tick".equals(tuple.getSourceStreamId())) {

            /**
             * Don't bother evaluating if wwe don't even have any flows
             */
            if(flows.size() > 0) {

                for(Flow flow : flows.values()) {

                    for(StreamDef curStream : flow.getStreams()) {
                        int idx = 0;
                        for(FlowOp curFlowOp : curStream.getFlowOps()) {

                            if(curFlowOp instanceof AggregateOp) {
                                AggregateOp op = (AggregateOp)curFlowOp;

                                /**
                                 * If we need to trigger any time-based policies, let's do that here
                                 */
                                if(op.getTriggerPolicy() == Policy.TIME || op.getEvictionPolicy() == Policy.TIME) {
                                    Cache<String, AggregatorWindow> windowCache = windows.get(flow.getId() + "\0" + curStream.getName() + "\0" + idx);
                                    if(windowCache != null) {
                                        for(AggregatorWindow window : windowCache.asMap().values()) {
                                            if(op.getEvictionPolicy() == Policy.TIME)
                                                window.timeEvict(op.getEvictionThreshold());

                                            if(op.getTriggerPolicy() == Policy.TIME)
                                                window.incrTriggerTicks();

                                            if(window.getTriggerTicks() == op.getTriggerThreshold())
                                                emitAggregate(flow, op, curStream.getName(), idx, window);
                                        }
                                    }
                                }
                            }
                            idx++;
                        }

                    }
                }
            }

        } else if(!"tick".equals(tuple.getSourceStreamId())){

            FlowInfo flowInfo = new FlowInfo(tuple);