public abstract class EventsLoaderBaseBolt extends BaseRichBolt {

  @Override
  public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    outputFieldsDeclarer.declare(new Fields("event"));
  }


} public abstract class FlowLoaderBaseBolt extends BaseRichBolt {

  @Override
  public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    outputFieldsDeclarer.declareStream(FLOW_LOADER_STREAM, new Fields("flows"));
  }


  protected void emitFlows(SpoutOutputCollector collector, Collection<Flow> flows) {
    collector.emit(FLOW_LOADER_STREAM, new Values(flows));
  }

} public abstract class EventsLoaderBaseSpout extends BaseRichSpout {

  @Override
  public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    outputFieldsDeclarer.declare(new Fields("event"));
  }

} public abstract class FlowLoaderBaseSpout extends BaseRichSpout {

  @Override
  public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    outputFieldsDeclarer.declareStream(FLOW_LOADER_STREAM, new Fields("flows"));
  }


  protected void emitFlows(SpoutOutputCollector collector, Collection<Flow> flows) {
    collector.emit(FLOW_LOADER_STREAM, new Values(flows));
  }


} 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);

            Flow flow = flows.get(flowInfo.getFlowId());

            if(flow != null) {

                AggregateOp op =  getFlowOpFromStream(flow, flowInfo.getStreamName(), flowInfo.getIdx());
                Cache<String, AggregatorWindow> windowCache = windows.get(flowInfo.getFlowId() + "\0" + flowInfo.getStreamName() + "\0" + flowInfo.getIdx());

                AggregatorWindow window = null;
                if(windowCache != null) {
                    window = windowCache.getIfPresent(flowInfo.getPartition());
                    if(window != null) { // if we have a window already constructed, proces it

                        /**
                         * If we need to evict any buffered items, let's do that here
                         */
                        if(op.getEvictionPolicy() == Policy.TIME)
                            window.timeEvict(op.getEvictionThreshold());
                    } else {
                        window = buildWindow(op, flowInfo.getStreamName(), flowInfo.getIdx(), flowInfo.getPartition(), flowInfo.getFlowId(), windowCache);
                    }
                } else {
                    windowCache = CacheBuilder.newBuilder().expireAfterWrite(op.getWindowEvictMillis(), TimeUnit.MILLISECONDS).build();
                    window = buildWindow(op, flowInfo.getStreamName(), flowInfo.getIdx(), flowInfo.getPartition(), flowInfo.getFlowId(), windowCache);
                }

                window.add(flowInfo.getEvent(), flowInfo.getPreviousStream());
                windowCache.put(flowInfo.getPartition(), window); // window eviction is on writes, so we need to write to the window to reset our expiration.

                /**
                 * Perform count-based trigger if necessary
                 */
                if(op.getTriggerPolicy() == Policy.COUNT) {
                    window.incrTriggerTicks();

                    if(window.getTriggerTicks() == op.getTriggerThreshold())
                        emitAggregate(flow, op, flowInfo.getStreamName(), flowInfo.getIdx(), window);
                }
            }
        }

      collector.ack(tuple);
    }

    private AggregatorWindow buildWindow(AggregateOp op, String stream, int idx, String hash, String flowId, Cache<String, AggregatorWindow> windowCache) {
      try {
        Aggregator agg = op.getAggregatorClass().newInstance();
        AggregatorWindow window = op.getEvictionPolicy() == Policy.TIME || op.getEvictionPolicy() == Policy.TIME_DELTA_LT ?
                new AggregatorWindow(agg, hash) : new AggregatorWindow(agg, hash, op.getEvictionThreshold());

        Map<String,String> aggConfig = op.getConfig();
        if(flows.get(flowId).getStream(stream).getFlowOps().get(idx-1) instanceof PartitionOp) {
          PartitionOp partitionOp = (PartitionOp)flows.get(flowId).getStream(stream).getFlowOps().get(idx-1);
          aggConfig.put(GROUP_BY, join(partitionOp.getFields(), GROUP_BY_DELIM));
        }

        agg.configure(aggConfig);

        windowCache.put(hash, window);
          windows.put(flowId + "\0" + stream + "\0" + idx, windowCache);
          return window;
      } catch (Exception e) {
          throw new RuntimeException(e);
      }
    }

    private void emitAggregate(Flow flow, AggregateOp op, String stream, int idx, AggregatorWindow window) {
        Collection<AggregatedEvent> eventsToEmit = window.getAggregate();
        String nextStream = idx+1 < flow.getStream(stream).getFlowOps().size() ? flow.getStream(stream).getFlowOps().get(idx+1).getComponentName() : "output";

        if(hasNextOutput(flow, stream, nextStream)) {
          for(AggregatedEvent event : eventsToEmit) {
            String previousStream = event.getPreviousStream() != null ? event.getPreviousStream() : stream;
            collector.emit(nextStream, new Values(flow.getId(), event.getEvent(), idx, stream, previousStream));  // Note: If aggregated event isn't keeping the previous stream, it's possible it could be lost
          }
        }

        // send to any other streams that are configured (aside from output)
        if(exportsToOtherStreams(flow, stream, nextStream)) {
          for(String output : flow.getStream(stream).getOutputs()) {
            for(AggregatedEvent event : eventsToEmit) {
              String outputComponent = flow.getStream(output).getFlowOps().get(0).getComponentName();
              collector.emit(outputComponent, new Values(flow.getId(), event.getEvent(), -1, output, stream));
            }
          }
        }

        if(op.isClearOnTrigger())
          window.clear();

        window.resetTriggerTicks();
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        declareOutputStreams(outputFieldsDeclarer, fields);
    }
} public class EachBolt extends BaseRichBolt {

    Map<String,Flow> flows;
    OutputCollector collector;

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

    @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())){

            FlowInfo flowInfo = new FlowInfo(tuple);

            Flow flow = flows.get(flowInfo.getFlowId());

            if(flow != null) {
                EachOp functionOp =  getFlowOpFromStream(flow, flowInfo.getStreamName(), flowInfo.getIdx());

                String nextStream = getNextStreamFromFlowInfo(flow, flowInfo.getStreamName(), flowInfo.getIdx());

                List<Event> events = functionOp.getFunction().execute(flowInfo.getEvent());

                if(hasNextOutput(flow, flowInfo.getStreamName(), nextStream)) {
                  if (events != null) {
                    for (Event newEvent : events)
                      collector.emit(nextStream, tuple, new Values(flowInfo.getFlowId(), newEvent, flowInfo.getIdx(), flowInfo.getStreamName(), flowInfo.getPreviousStream()));
                  }
                }

                // send directly to any non std output streams that may be configured
                if (exportsToOtherStreams(flow, flowInfo.getStreamName(), nextStream)) {
                  for (String output : flow.getStream(flowInfo.getStreamName()).getOutputs()) {
                    if (events != null) {
                      for (Event newEvent : events) {
                        String outputStream = flow.getStream(output).getFlowOps().get(0).getComponentName();
                        collector.emit(outputStream, tuple, new Values(flowInfo.getFlowId(), newEvent, -1, output, flowInfo.getStreamName()));
                      }
                    }
                  }
                }
              }
        }

      collector.ack(tuple);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        FlowmixBuilder.declareOutputStreams(outputFieldsDeclarer, fields);
    }
} public class FilterBolt extends BaseRichBolt {

    Map<String,Flow> flows;
    OutputCollector collector;

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

    @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())){

            FlowInfo flowInfo = new FlowInfo(tuple);

            Flow flow = flows.get(flowInfo.getFlowId());

            if(flow != null) {

                FilterOp filterOp = getFlowOpFromStream(flow, flowInfo.getStreamName(), flowInfo.getIdx());

                if(filterOp.getFilter().accept(flowInfo.getEvent())) {
                  String nextStream = getNextStreamFromFlowInfo(flow, flowInfo.getStreamName(), flowInfo.getIdx());

                  if(hasNextOutput(flow, flowInfo.getStreamName(), nextStream))
                    collector.emit(nextStream, tuple, new Values(flow.getId(), flowInfo.getEvent(), flowInfo.getIdx(), flowInfo.getStreamName(), flowInfo.getPreviousStream()));

                  // send directly to any non std output streams
                  if(exportsToOtherStreams(flow, flowInfo.getStreamName(), nextStream)) {
                    for (String output : flow.getStream(flowInfo.getStreamName()).getOutputs()) {
                      String outputStream = flow.getStream(output).getFlowOps().get(0).getComponentName();
                      collector.emit(outputStream, tuple, new Values(flowInfo.getFlowId(), flowInfo.getEvent(), -1, output, flowInfo.getStreamName()));
                    }
                  }
                }
            }
        }

      collector.ack(tuple);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        declareOutputStreams(outputFieldsDeclarer, fields);
    }
} public class FlowInitializerBolt extends BaseRichBolt {

    Map<String,Flow> flows;
    OutputCollector collector;

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

    @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())){

            if(flows.size() > 0) {
                for(Flow flow : flows.values()) {
                    Collection<Event> events = (Collection<Event>) tuple.getValue(0);
                    for(Event event : events) {

                        for(StreamDef stream : flow.getStreams()) {
                            String streamid = stream.getFlowOps().get(0).getComponentName();
                            String streamName = stream.getName();

                            if(stream.isStdInput())
                              collector.emit(streamid, tuple, new Values(flow.getId(), event, -1, streamName, streamName));
                        }
                    }
                }
            }

            collector.ack(tuple);
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        FlowmixBuilder.declareOutputStreams(outputFieldsDeclarer, fields);
    }
} public class JoinBolt extends BaseRichBolt {

    Map<String, Flow> rulesMap;
    Map<String, Cache<String, Window>> windows;

    OutputCollector collector;

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

    @Override
    public void execute(Tuple tuple) {

        /**
         * Update rules if necessary
         */
        if(FLOW_LOADER_STREAM.equals(tuple.getSourceStreamId())) {

            Collection<Flow> rules = (Collection<Flow>) tuple.getValue(0);
            Set<String> rulesToRemove = new HashSet<String>();

            // find deleted rules and remove them
            for(Flow rule : rulesMap.values()) {
                if(!rules.contains(rule))
                    rulesToRemove.add(rule.getId());
            }

            /**
             * Remove any deleted rules
             */
            for(String ruleId : rulesToRemove) {
                rulesMap.remove(ruleId);
                windows.remove(ruleId);
            }

            for(Flow rule : rules) {
                /**
                 * If a rule has been updated, let's drop the window windows and start out fresh.
                 */
                if(rulesMap.get(rule.getId()) != null && !rulesMap.get(rule.getId()).equals(rule) ||
                        !rulesMap.containsKey(rule.getId())) {
                    rulesMap.put(rule.getId(), rule);
                    windows.remove(rule.getId());
                }
            }

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

            /**
             * Don't bother evaluating if we don't even have any rules
             */
            if(rulesMap.size() > 0) {

                for(Flow rule : rulesMap.values()) {

                    for(StreamDef stream : rule.getStreams()) {

                        int idx = 0;
                        for(FlowOp curOp : stream.getFlowOps()) {

                            if(curOp instanceof JoinOp) {

                                JoinOp op = (JoinOp) curOp;
                                /**
                                 * If we need to trigger any time-based policies, let's do that here.
                                 */
                                if(op.getEvictionPolicy() == Policy.TIME) {

                                    Cache<String, Window> buffersForRule = windows.get(rule.getId() + "\0" + stream.getName() + "\0" + idx);
                                    if(buffersForRule != null)
                                        for (Window buffer : buffersForRule.asMap().values())
                                            buffer.timeEvict(op.getEvictionThreshold());
                                }
                            }
                            idx++;
                        }

                    }

                }
            }

        } else {

            /**
             * Short circuit if we don't have any rules.
             */
            if (rulesMap.size() > 0) {

                FlowInfo flowInfo = new FlowInfo(tuple);

                Flow flow = rulesMap.get(flowInfo.getFlowId());

                JoinOp op =  getFlowOpFromStream(flow, flowInfo.getStreamName(), flowInfo.getIdx());

                // do processing on lhs
                if(flowInfo.getPreviousStream().equals(op.getLeftStream())) {

                    Cache<String, Window> buffersForRule = windows.get(flow.getId() + "\0" + flowInfo.getStreamName() + "\0" + flowInfo.getIdx());
                    Window buffer;
                    if (buffersForRule != null) {
                        buffer = buffersForRule.getIfPresent(flowInfo.getPartition());

                        if (buffer != null) {    // if we have a buffer already, process it
                            /**
                             * If we need to evict any buffered items, let's do it here
                             */
                            if(op.getEvictionPolicy() == Policy.TIME)
                                buffer.timeEvict(op.getEvictionThreshold());
                        }
                    } else {
                        buffersForRule = CacheBuilder.newBuilder().expireAfterAccess(60, TimeUnit.MINUTES).build(); // just in case we get some rogue data, we don't wan ti to sit for too long.
                        buffer = op.getEvictionPolicy() == Policy.TIME ? new Window(flowInfo.getPartition()) :
                                new Window(flowInfo.getPartition(), op.getEvictionThreshold());
                        buffersForRule.put(flowInfo.getPartition(), buffer);
                        windows.put(flow.getId() + "\0" + flowInfo.getStreamName() + "\0" + flowInfo.getIdx(), buffersForRule);
                    }

                    buffer.add(flowInfo.getEvent(), flowInfo.getPreviousStream());

                } else if(flowInfo.getPreviousStream().equals(op.getRightStream())) {

                    Cache<String, Window> buffersForRule = windows.get(flow.getId() + "\0" + flowInfo.getStreamName() + "\0" + flowInfo.getIdx());
                    Window buffer;
                    if (buffersForRule != null) {
                        buffer = buffersForRule.getIfPresent(flowInfo.getPartition());

                        for(WindowItem bufferedEvent : buffer.getEvents()) {
                          Event joined = new BaseEvent(bufferedEvent.getEvent().getId(), bufferedEvent.getEvent().getTimestamp());

                          // the hashcode will filter duplicates
                          joined.putAll(concat(bufferedEvent.getEvent().getTuples()));
                          joined.putAll(concat(flowInfo.getEvent().getTuples()));
                          String nextStream = getNextStreamFromFlowInfo(flow, flowInfo.getStreamName(), flowInfo.getIdx());

                          if(hasNextOutput(flow, flowInfo.getStreamName(), nextStream))
                              collector.emit(nextStream, new Values(flow.getId(), joined, flowInfo.getIdx(), flowInfo.getStreamName(), bufferedEvent.getPreviousStream()));

                          // send to any other streams that are configured (aside from output)
                          if(exportsToOtherStreams(flow, flowInfo.getStreamName(), nextStream)) {
                            for(String output : flow.getStream(flowInfo.getStreamName()).getOutputs()) {
                              String outputComponent = flow.getStream(output).getFlowOps().get(0).getComponentName();
                              collector.emit(outputComponent, new Values(flow.getId(), joined, -1, output, flowInfo.getStreamName()));
                            }
                          }
                        }
                    }
                } else {
                    throw new RuntimeException("Received event for stream that does not match the join. Flowmix has been miswired.");
                }
            }

        }

        collector.ack(tuple);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        declareOutputStreams(outputFieldsDeclarer, fields);
    }
} public class PartitionBolt extends BaseRichBolt {

    Map<String,Flow> flows;
    OutputCollector collector;

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

    @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())) {

            FlowInfo flowInfo = new FlowInfo(tuple);

            Flow flow = flows.get(flowInfo.getFlowId());

            if(flow != null) {

                PartitionOp partitionOp = getFlowOpFromStream(flow, flowInfo.getStreamName(), flowInfo.getIdx());

                String nextStream = Utils.getNextStreamFromFlowInfo(flow, flowInfo.getStreamName(), flowInfo.getIdx());
                String hash = buildKeyIndexForEvent(flowInfo.getFlowId(), flowInfo.getEvent(), partitionOp.getFields());

                if(hasNextOutput(flow, flowInfo.getStreamName(), nextStream))
                  collector.emit(nextStream, tuple, new Values(flowInfo.getFlowId(), flowInfo.getEvent(), flowInfo.getIdx(), flowInfo.getStreamName(), hash, flowInfo.getPreviousStream()));

                // send directly to any other non std output streams that may be configured
                if(exportsToOtherStreams(flow, flowInfo.getStreamName(), nextStream)) {
                  for (String output : flow.getStream(flowInfo.getStreamName()).getOutputs()) {
                    String outputStream = flow.getStream(output).getFlowOps().get(0).getComponentName();
                    collector.emit(outputStream, tuple, new Values(flowInfo.getFlowId(), flowInfo.getEvent(), -1, output, hash, flowInfo.getStreamName()));
                  }
                }
            }
        }

      collector.ack(tuple);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        declareOutputStreams(outputFieldsDeclarer, partitionFields);
    }
} public class SelectorBolt extends BaseRichBolt {

    Map<String,Flow> flows;
    OutputCollector collector;


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

    @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())) {

          FlowInfo flowInfo = new FlowInfo(tuple);

          Flow flow = flows.get(flowInfo.getFlowId());

            if (flow != null) {

                SelectOp selectOp =  getFlowOpFromStream(flow, flowInfo.getStreamName(), flowInfo.getIdx());

                String nextStream = Utils.getNextStreamFromFlowInfo(flow, flowInfo.getStreamName(), flowInfo.getIdx());

                Event newEvent = new BaseEvent(flowInfo.getEvent().getId(), flowInfo.getEvent().getTimestamp());
                for(org.calrissian.mango.domain.Tuple eventTuple : flowInfo.getEvent().getTuples()) {
                    if(selectOp.getFields().contains(eventTuple.getKey()))
                      newEvent.put(eventTuple);
                }

                /**
                 * If no selected tuples existed, event will not be emitted
                 */
                if(hasNextOutput(flow, flowInfo.getStreamName(), nextStream)) {
                  if (newEvent.getTuples().size() > 0)
                    collector.emit(nextStream, tuple, new Values(flowInfo.getFlowId(), newEvent, flowInfo.getIdx(), flowInfo.getStreamName(), flowInfo.getPreviousStream()));
                }

                // send directly to any non std output streams
                if(exportsToOtherStreams(flow, flowInfo.getStreamName(), nextStream)) {
                  for (String output : flow.getStream(flowInfo.getStreamName()).getOutputs()) {
                    String outputStream = flow.getStream(output).getFlowOps().get(0).getComponentName();
                    collector.emit(outputStream, tuple, new Values(flowInfo.getFlowId(), flowInfo.getEvent(), -1, output, flowInfo.getStreamName()));
                  }
                }

            }
        }
      collector.ack(tuple);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        FlowmixBuilder.declareOutputStreams(outputFieldsDeclarer, fields);
    }
} public class SortBolt extends BaseRichBolt {

  Map<String, Flow> flowMap;
  Map<String, Cache<String,SortedWindow>> windows;

  OutputCollector collector;


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

  @Override
  public void execute(Tuple tuple) {

    /**
     * Update rules if necessary
     */
    if(FLOW_LOADER_STREAM.equals(tuple.getSourceStreamId())) {

      Collection<Flow> flows = (Collection<Flow>) tuple.getValue(0);
      Set<String> rulesToRemove = new HashSet<String>();

      // find deleted rules and remove them
      for(Flow flow : flowMap.values()) {
        if(!flows.contains(flow))
          rulesToRemove.add(flow.getId());
      }

      /**
       * Remove any deleted rules
       */
      for(String flowId : rulesToRemove) {
        flowMap.remove(flowId);
        windows.remove(flowId);
      }

      for(Flow flow : flows) {
        /**
         * If a rule has been updated, let's drop the window windows and start out fresh.
         */
        if(flowMap.get(flow.getId()) != null && !flowMap.get(flow.getId()).equals(flow) ||
                !flowMap.containsKey(flow.getId())) {
          flowMap.put(flow.getId(), flow);
          windows.remove(flow.getId());
        }
      }

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

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

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

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

              if(curFlowOp instanceof SortOp) {
                SortOp op = (SortOp)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, SortedWindow> windowCache = windows.get(flow.getId() + "\0" + curStream.getName() + "\0" + idx);
                  if(windowCache != null) {
                    for(SortedWindow 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()) {
                        FlowInfo flowInfo = new FlowInfo(flow.getId(), curStream.getName(), idx);
                        emitWindow(flowInfo, flow, op, window);
                      }
                    }
                  }

                }
              }
              idx++;
            }

          }
        }
      }

    } else {

      /**
       * Short circuit if we don't have any rules.
       */
      if (flowMap.size() > 0) {

        /**
         * If we've received an event for an flowmix rule, we need to act on it here. Purposefully, the groupBy
         * fields have been hashed so that we know the buffer exists on this current bolt for the given rule.
         *
         * The hashKey was added to the "fieldsGrouping" in an attempt to share pointers where possible. Different
         * rules with like fields groupings can store the items in their windows on the same node.
         */
        FlowInfo flowInfo = new FlowInfo(tuple);

        Flow flow = flowMap.get(flowInfo.getFlowId());

        SortOp op = (SortOp) flow.getStream(flowInfo.getStreamName()).getFlowOps().get(flowInfo.getIdx());

        Cache<String, SortedWindow> buffersForRule = windows.get(flow.getId() + "\0" + flowInfo.getStreamName() + "\0" + flowInfo.getIdx());
        SortedWindow buffer;
        if (buffersForRule != null) {
          buffer = buffersForRule.getIfPresent(flowInfo.getPartition());

          if (buffer != null) {    // if we have a buffer already, process it
            if(op.getEvictionPolicy() == Policy.TIME)
              buffer.timeEvict(op.getEvictionThreshold());

          } else {
            buffersForRule = CacheBuilder.newBuilder().expireAfterAccess(60, TimeUnit.MINUTES).build(); // just in case we get some rogue data, we don't wan ti to sit for too long.
            buffer = buildWindow(flowInfo.getPartition(), op);
            buffersForRule.put(flowInfo.getPartition(), buffer);
            windows.put(flow.getId() + "\0" + flowInfo.getStreamName() + "\0" + flowInfo.getIdx(), buffersForRule);
          }
        } else {
          buffersForRule = CacheBuilder.newBuilder().expireAfterAccess(60, TimeUnit.MINUTES).build(); // just in case we get some rogue data, we don't wan ti to sit for too long.
          buffer = buildWindow(flowInfo.getPartition(), op);
          buffersForRule.put(flowInfo.getPartition(), buffer);
          windows.put(flow.getId() + "\0" + flowInfo.getStreamName() + "\0" + flowInfo.getIdx(), buffersForRule);
        }

        if(op.getEvictionPolicy() == Policy.COUNT && op.getEvictionThreshold() == buffer.size())
          buffer.expire();

        buffer.add(flowInfo.getEvent(), flowInfo.getPreviousStream());

        /**
         * Perform count-based trigger if necessary
         */
        if (op.getTriggerPolicy() == Policy.COUNT) {
          buffer.incrTriggerTicks();

          if(buffer.getTriggerTicks() == op.getTriggerThreshold())
            emitWindow(flowInfo, flow, op, buffer);
        } else if(op.getTriggerPolicy() == Policy.TIME_DELTA_LT && buffer.timeRange() > -1 && buffer.timeRange() <= op.getTriggerThreshold() * 1000)
          emitWindow(flowInfo, flow, op, buffer);

//        /**
//         * If we aren't supposed to clear the window right now, then we need to emit
//         */
//        else if(!op.isClearOnTrigger()) {
//
//          if(op.getEvictionPolicy() != Policy.COUNT || (op.getEvictionPolicy() == Policy.COUNT && op.getTriggerThreshold() == windows.size()))
//            emitWindow(flow, streamName, op, buffer, idx);
//        }
      }

    }

    collector.ack(tuple);
  }

  private void emitWindow(FlowInfo flowInfo, Flow flow, SortOp op, Window window) {

    /**
     * If the window is set to be cleared, we need to emit everything. Otherwise, just emit the last item in the list.
     */
    Iterable<WindowItem> items = null;
    if(op.isClearOnTrigger())
      items = window.getEvents();
    else {
      if(op.isProgressive() && window.size() == op.getEvictionThreshold())    // we know if it's a progressive window, the eviction policy is count.
        items = singleton(window.expire());
      else
        items = window.getEvents();
    }

    if(items != null) {
      for(WindowItem item : items) {

        String nextStream = getNextStreamFromFlowInfo(flow, flowInfo.getStreamName(), flowInfo.getIdx());
        if(hasNextOutput(flow, flowInfo.getStreamName(), nextStream))
          collector.emit(nextStream, new Values(flow.getId(), item.getEvent(), flowInfo.getIdx(), flowInfo.getStreamName(), item.getPreviousStream()));

        // send directly to any non std output streams
        if(exportsToOtherStreams(flow, flowInfo.getStreamName(), nextStream)) {
          for (String output : flow.getStream(flowInfo.getStreamName()).getOutputs()) {
            String outputStream = flow.getStream(output).getFlowOps().get(0).getComponentName();
            collector.emit(outputStream, new Values(flow.getId(), item.getEvent(), -1, output, flowInfo.getStreamName()));
          }
        }
      }

      if(op.isClearOnTrigger())
        window.clear();
    }

    window.resetTriggerTicks();
  }

  private SortedWindow buildWindow(String hash, SortOp op) {
    Comparator<WindowItem> comparator = new EventSortByComparator(op.getSortBy());
    return op.getEvictionPolicy() != Policy.COUNT ? new SortedWindow(hash, comparator, op.isClearOnTrigger()) :
            new SortedWindow(hash, comparator, op.getEvictionThreshold(), op.isClearOnTrigger());
  }

  @Override
  public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    declareOutputStreams(outputFieldsDeclarer, fields);
  }

} public class SplitBolt extends BaseRichBolt {

  Map<String,Flow> flows;
  OutputCollector collector;

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

  @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())) {

      FlowInfo flowInfo = new FlowInfo(tuple);

      Flow flow = flows.get(flowInfo.getFlowId());

      if (flow != null) {
        SplitOp splitOp = getFlowOpFromStream(flow, flowInfo.getStreamName(), flowInfo.getIdx());

        String nextStream = getNextStreamFromFlowInfo(flow, flowInfo.getStreamName(), flowInfo.getIdx());

        // first check the default path
        Filter filter = splitOp.getDefaultPath();

        if(filter != null && filter.accept(flowInfo.getEvent())) {
          if (hasNextOutput(flow, flowInfo.getStreamName(), nextStream))
              collector.emit(nextStream, tuple, new Values(flowInfo.getFlowId(), flowInfo.getEvent(), flowInfo.getIdx(), flowInfo.getStreamName(), flowInfo.getPreviousStream()));

          // send directly to any non std output streams
          if (exportsToOtherStreams(flow, flowInfo.getStreamName(), nextStream)) {
            for (String output : flow.getStream(flowInfo.getStreamName()).getOutputs()) {
              String outputStream = flow.getStream(output).getFlowOps().get(0).getComponentName();
              collector.emit(outputStream, tuple, new Values(flowInfo.getFlowId(), flowInfo.getEvent(), -1, output, flowInfo.getStreamName()));
            }
          }
        }

        // then check all other paths
        if(splitOp.getPaths() != null) {
          for(Pair<Filter, String> pathPair : splitOp.getPaths()) {
            if(pathPair.getOne().accept(flowInfo.getEvent())) {
              String outputStream = flow.getStream(pathPair.getTwo()).getFlowOps().get(0).getComponentName();
              collector.emit(outputStream, tuple, new Values(flowInfo.getFlowId(), flowInfo.getEvent(), -1, pathPair.getTwo(), flowInfo.getStreamName()));
            }
          }
        }
      }
    }
    collector.ack(tuple);
  }

  @Override
  public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    FlowmixBuilder.declareOutputStreams(outputFieldsDeclarer, fields);
  }
} public class SwitchBolt extends BaseRichBolt {

    Map<String, Flow> flowMap;
    Map<String, Cache<String, SwitchWindow>> windows;

    OutputCollector collector;


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

    @Override
    public void execute(Tuple tuple) {

        collector.ack(tuple);

        /**
         * Update rules if necessary
         */
        if(FLOW_LOADER_STREAM.equals(tuple.getSourceStreamId())) {

            Collection<Flow> flows = (Collection<Flow>) tuple.getValue(0);
            Set<String> rulesToRemove = new HashSet<String>();

            // find deleted rules and remove them
            for(Flow flow : flowMap.values()) {
                if(!flows.contains(flow))
                    rulesToRemove.add(flow.getId());
            }

            /**
             * Remove any deleted rules
             */
            for(String flowId : rulesToRemove) {
                flowMap.remove(flowId);
                windows.remove(flowId);
            }

            for(Flow flow : flows) {
                /**
                 * If a rule has been updated, let's drop the window windows and start out fresh.
                 */
                if(flowMap.get(flow.getId()) != null && !flowMap.get(flow.getId()).equals(flow) ||
                        !flowMap.containsKey(flow.getId())) {
                    flowMap.put(flow.getId(), flow);
                    windows.remove(flow.getId());
                }
            }

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

            /**
             * Don't bother evaluating if we don't even have any rules
             */
            if(flowMap.size() > 0) {

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

                    for(StreamDef stream : flow.getStreams()) {

                        int idx = 0;
                        for(FlowOp curOp : stream.getFlowOps()) {
                            if(curOp instanceof SwitchOp) {
                                SwitchOp op = (SwitchOp)curOp;
                                /**
                                 * If we need to trigger any time-based policies, let's do that here.
                                 */
                                if(op.getOpenPolicy() == Policy.TIME || op.getClosePolicy() == Policy.TIME || op.getEvictionPolicy() == Policy.TIME) {
                                    Cache<String, SwitchWindow> buffersForRule = windows.get(flow.getId() + "\0" + stream.getName() + "\0" + idx);
                                    if(buffersForRule != null) {
                                        for (SwitchWindow buffer : buffersForRule.asMap().values()) {
                                            if(op.getOpenPolicy() == Policy.TIME && !buffer.isStopped()) {
                                                if (buffer.getTriggerTicks() == op.getOpenThreshold()) {
                                                    buffer.setStopped(true);
                                                    buffer.clear();
                                                } else {
                                                    buffer.incrTriggerTicks();
                                                }
                                            }

                                            boolean justOpened = false;
                                            if(op.getClosePolicy() == Policy.TIME && buffer.isStopped()) {
                                                if(buffer.getStopTicks() == op.getCloseThreshold()) {
                                                    buffer.setStopped(false);
                                                    buffer.resetStopTicks();
                                                } else {
                                                    buffer.incrementStopTicks();
                                                }
                                            }

                                            if(!justOpened)
                                            if(op.getEvictionPolicy() == Policy.TIME && !buffer.isStopped()) {
                                                if(buffer.getEvictionTicks() == op.getEvictionThreshold()) {
                                                    activateOpenPolicy(buffer, op);
                                                } else {
                                                    buffer.incrementEvictionTicks();
                                                }
                                            }
                                        }
                                    }
                                }
                            }

                            idx++;
                        }

                    }
                }
            }

        } else {

            /**
             * Short circuit if we don't have any rules.
             */
            if (flowMap.size() > 0) {

                /**
                 * If we've received an event for an flowmix rule, we need to act on it here. Purposefully, the groupBy
                 * fields have been hashed so that we know the buffer exists on this current bolt for the given rule.
                 *
                 * The hashKey was added to the "fieldsGrouping" in an attempt to share pointers where possible. Different
                 * rules with like fields groupings can store the items in their windows on the same node.
                 */
                FlowInfo flowInfo = new FlowInfo(tuple);

                Flow flow = flowMap.get(flowInfo.getFlowId());

                SwitchOp op = (SwitchOp) flow.getStream(flowInfo.getStreamName()).getFlowOps().get(flowInfo.getIdx());

                Cache<String, SwitchWindow> buffersForRule = windows.get(flow.getId() + "\0" + flowInfo.getStreamName() + "\0" + flowInfo.getIdx());
                SwitchWindow buffer;
                if (buffersForRule != null) {
                    buffer = buffersForRule.getIfPresent(flowInfo.getPartition());

                    if (buffer != null) {    // if we have a buffer already, process it

                        if(!buffer.isStopped()) {
                            /**
                             * If we need to evict any buffered items, let's do it here
                             */
                            if(op.getEvictionPolicy() == Policy.TIME)
                                buffer.timeEvict(op.getEvictionThreshold());
                        }
                    } else {
                        buffer = op.getEvictionPolicy() == Policy.TIME ? new SwitchWindow(flowInfo.getPartition()) :
                            new SwitchWindow(flowInfo.getPartition(), op.getEvictionThreshold());
                        buffersForRule.put(flowInfo.getPartition(), buffer);
                    }
                } else {
                    buffersForRule = CacheBuilder.newBuilder().expireAfterAccess(60, TimeUnit.MINUTES).build(); // just in case we get some rogue data, we don't wan ti to sit for too long.
                    buffer = op.getEvictionPolicy() == Policy.TIME ? new SwitchWindow(flowInfo.getPartition()) :
                            new SwitchWindow(flowInfo.getPartition(), op.getEvictionThreshold());
                    buffersForRule.put(flowInfo.getPartition(), buffer);
                    windows.put(flow.getId() + "\0" + flowInfo.getStreamName() + "\0" + flowInfo.getIdx(), buffersForRule);
                }

                if(buffer.isStopped()) {
                    if(op.getClosePolicy() == Policy.COUNT ) {
                        if(buffer.getStopTicks() == op.getCloseThreshold()) {
                            buffer.setStopped(false);
                            buffer.resetStopTicks();
                        } else {
                            buffer.incrementStopTicks();
                        }
                    }
                }

                /**
                 * Perform count-based trigger if necessary
                 */
                if(!buffer.isStopped()) {

                    if (op.getOpenPolicy() == Policy.COUNT || op.getEvictionPolicy() == Policy.COUNT) {
                        buffer.incrTriggerTicks();
                        activateOpenPolicy(buffer, op);
                    }
                }

                if(!buffer.isStopped()) {
                    buffer.add(flowInfo.getEvent(), flowInfo.getPreviousStream());
                    String nextStream = getNextStreamFromFlowInfo(flow, flowInfo.getStreamName(), flowInfo.getIdx());

                    if(hasNextOutput(flow, flowInfo.getStreamName(), nextStream))
                      collector.emit(nextStream, tuple, new Values(flow.getId(), flowInfo.getEvent(), flowInfo.getIdx(), flowInfo.getStreamName(), flowInfo.getPreviousStream()));

                    // send directly to any non std output streams
                    if(exportsToOtherStreams(flow, flowInfo.getStreamName(), nextStream)) {
                      for (String output : flow.getStream(flowInfo.getStreamName()).getOutputs()) {
                        String outputStream = flow.getStream(output).getFlowOps().get(0).getComponentName();
                        collector.emit(outputStream, tuple, new Values(flowInfo.getFlowId(), flowInfo.getEvent(), -1, output, flowInfo.getStreamName()));
                      }
                    }
                  }
            }

        }

      collector.ack(tuple);
    }

    private boolean isWindowFull(SwitchOp op, SwitchWindow window) {

       return (op.getEvictionPolicy() == Policy.COUNT && op.getEvictionThreshold() == window.size()) ||
              (op.getEvictionPolicy() == Policy.TIME && op.getEvictionThreshold() == window.getEvictionTicks());
    }

    private void activateOpenPolicy(SwitchWindow buffer, SwitchOp op) {

        if(buffer.getTriggerTicks() == op.getOpenThreshold()) {
            buffer.setStopped(true);
            buffer.resetTriggerTicks();
            buffer.clear();
            buffer.resetEvictionTicks();
        }

        if(op.getOpenPolicy() == Policy.TIME_DELTA_LT && buffer.timeRange() > -1 && buffer.timeRange() <= op.getOpenThreshold() * 1000) {
            if(isWindowFull(op, buffer)) {
                buffer.setStopped(true);
                buffer.clear();
                buffer.resetEvictionTicks();
            }
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        declareOutputStreams(outputFieldsDeclarer, fields);
    }

} public class TickSpout extends BaseRichSpout{

    private long sleepMillis;
    private SpoutOutputCollector collector;

    public TickSpout(long sleepMillis) {
        this.sleepMillis = sleepMillis;
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declareStream("tick", new Fields());
    }

    @Override
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        this.collector = spoutOutputCollector;
    }

    @Override
    public void nextTuple() {
        collector.emit("tick", new Values());
        try {
            Thread.sleep(sleepMillis);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
} public static final String BROADCAST_STREAM = "control"; public static final String EVENT = "event"; public static final String FLOW_ID = "flowId"; public static final String FLOW_LOADER_STREAM = "flowLoaderStream"; public static final String FLOW_OP_IDX = "flowOpIdx"; public static final String INITIALIZER = "initializer"; public static final String LAST_STREAM = "lastStream"; public static final String OUTPUT = "output"; public static final String STREAM_NAME = "streamName"; public static final String AGGREGATE = "aggregate"; public static final String EACH = "each"; public static final String FILTER = "filter"; public static final String JOIN = "join"; public static final String PARTITION = "partition"; public static final String SELECT = "select"; public static final String SORT = "sort"; public static final String SPLIT = "split"; public static final String SWITCH = "stopGate";
import backtype.storm.topology.BoltDeclarer;
import backtype.storm.topology.IComponent;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import org.calrissian.flowmix.api.storm.bolt.EventsLoaderBaseBolt;
import org.calrissian.flowmix.api.storm.bolt.FlowLoaderBaseBolt;
import org.calrissian.flowmix.api.storm.spout.EventsLoaderBaseSpout;
import org.calrissian.flowmix.api.storm.spout.FlowLoaderBaseSpout;
import org.calrissian.flowmix.core.storm.bolt.AggregatorBolt;
import org.calrissian.flowmix.core.storm.bolt.EachBolt;
import org.calrissian.flowmix.core.storm.bolt.FilterBolt;
import org.calrissian.flowmix.core.storm.bolt.FlowInitializerBolt;
import org.calrissian.flowmix.core.storm.bolt.JoinBolt;
import org.calrissian.flowmix.core.storm.bolt.PartitionBolt;
import org.calrissian.flowmix.core.storm.bolt.SelectorBolt;
import org.calrissian.flowmix.core.storm.bolt.SortBolt;
import org.calrissian.flowmix.core.storm.bolt.SplitBolt;
import org.calrissian.flowmix.core.storm.bolt.SwitchBolt;
import org.calrissian.flowmix.core.storm.spout.TickSpout;
import static org.calrissian.flowmix.core.Constants.BROADCAST_STREAM;
import static org.calrissian.flowmix.core.Constants.EVENT;
import static org.calrissian.flowmix.core.Constants.FLOW_ID;
import static org.calrissian.flowmix.core.Constants.FLOW_LOADER_STREAM;
import static org.calrissian.flowmix.core.Constants.FLOW_OP_IDX;
import static org.calrissian.flowmix.core.Constants.INITIALIZER;
import static org.calrissian.flowmix.core.Constants.LAST_STREAM;
import static org.calrissian.flowmix.core.Constants.OUTPUT;
import static org.calrissian.flowmix.core.Constants.STREAM_NAME;
import static org.calrissian.flowmix.core.model.op.AggregateOp.AGGREGATE;
import static org.calrissian.flowmix.core.model.op.EachOp.EACH;
import static org.calrissian.flowmix.core.model.op.FilterOp.FILTER;
import static org.calrissian.flowmix.core.model.op.JoinOp.JOIN;
import static org.calrissian.flowmix.core.model.op.PartitionOp.PARTITION;
import static org.calrissian.flowmix.core.model.op.SelectOp.SELECT;
import static org.calrissian.flowmix.core.model.op.SortOp.SORT;
import static org.calrissian.flowmix.core.model.op.SplitOp.SPLIT;
import static org.calrissian.flowmix.core.model.op.SwitchOp.SWITCH;
/*
 * 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.api.builder;



/**
 * Builds the base flowmix topology configuration. The topology builder is returned so that it can be further
 * customized. Most often, it will be useful to further provision a downstream bolt that will process the data
 * even after the output. The output stream and component id provisioned on the output of the builder are both
 * "output".
 */

public class FlowmixBuilder {

  private IComponent flowLoaderSpout;
  private IComponent eventsComponent;
  private IRichBolt outputBolt;
  private int parallelismHint = 1;
  private int eventLoaderParallelism = -1;

  /**
   * @param flowLoader A spout that feeds rules into flowmix. This just needs to emit a Collection<Flow> in each tuple
   *                  at index 0 with a field name of "flows".
   * @param eventsSpout A spout that provides the events to std input.
   * @param outputBolt  A bolt to accept the output events (with the field name "event")
   * @param parallelismHint The number of executors to run the parallel streams.
   */
  public FlowmixBuilder setFlowLoader(FlowLoaderBaseSpout flowLoader) {
    this.flowLoaderSpout = flowLoader;
    return this;
  }

  public FlowmixBuilder setFlowLoader(FlowLoaderBaseBolt flowLoader) {
    this.flowLoaderSpout = flowLoader;
    return this;
  }

  public FlowmixBuilder setEventsLoader(EventsLoaderBaseBolt eventsLoader) {
    this.eventsComponent = eventsLoader;
    return this;
  }

  public FlowmixBuilder setEventsLoader(EventsLoaderBaseSpout eventsLoader) {
    this.eventsComponent = eventsLoader;
    return this;
  }

  public FlowmixBuilder setEventLoaderParallelism(int eventLoaderParallelism) {
    this.eventLoaderParallelism = eventLoaderParallelism;
    return this;
  }

  public FlowmixBuilder setOutputBolt(IRichBolt outputBolt) {
    this.outputBolt = outputBolt;
    return this;
  }

  public FlowmixBuilder setParallelismHint(int parallelismHint) {
    this.parallelismHint = parallelismHint;
    return this;
  }

  private void validateOptions() {

    String errorPrefix = "Error constructing Flowmix: ";
    if(flowLoaderSpout == null)
      throw new RuntimeException(errorPrefix + "A flow loader component needs to be set.");
    else if(eventsComponent == null)
      throw new RuntimeException(errorPrefix + "An event loader component needs to be set.");
    else if(outputBolt == null)
      throw new RuntimeException(errorPrefix + "An output bolt needs to be set.");
  }

  /**
   * @return A topology builder than can further be customized.
   */
  public TopologyBuilder create() {

      TopologyBuilder builder = new TopologyBuilder();

      if(eventsComponent instanceof IRichSpout)
        builder.setSpout(EVENT, (IRichSpout) eventsComponent, eventLoaderParallelism == -1 ? parallelismHint : eventLoaderParallelism);