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);
      else if(eventsComponent instanceof IRichBolt)
        builder.setBolt(EVENT, (IRichBolt) eventsComponent, eventLoaderParallelism == -1 ? parallelismHint : eventLoaderParallelism);
      else
        throw new RuntimeException("The component for events is not valid. Must be IRichSpout or IRichBolt");


      if(flowLoaderSpout instanceof IRichSpout)
        builder.setSpout(FLOW_LOADER_STREAM, (IRichSpout) flowLoaderSpout, 1);
      else if(flowLoaderSpout instanceof IRichBolt)
        builder.setBolt(FLOW_LOADER_STREAM, (IRichBolt) flowLoaderSpout, 1);
      else
        throw new RuntimeException("The component for rules is not valid. Must be IRichSpout or IRichBolt");

      builder.setSpout("tick", new TickSpout(1000), 1);
      builder.setBolt(INITIALIZER, new FlowInitializerBolt(), parallelismHint)  // kicks off a flow determining where to start
              .localOrShuffleGrouping(EVENT)
              .allGrouping(FLOW_LOADER_STREAM, FLOW_LOADER_STREAM);

      declarebolt(builder, FILTER, new FilterBolt(), parallelismHint, true);
      declarebolt(builder, SELECT, new SelectorBolt(), parallelismHint, true);
      declarebolt(builder, PARTITION, new PartitionBolt(), parallelismHint, true);
      declarebolt(builder, SWITCH, new SwitchBolt(), parallelismHint, true);
      declarebolt(builder, AGGREGATE, new AggregatorBolt(), parallelismHint, true);
      declarebolt(builder, JOIN, new JoinBolt(), parallelismHint, true);
      declarebolt(builder, EACH, new EachBolt(), parallelismHint, true);
      declarebolt(builder, SORT, new SortBolt(), parallelismHint, true);
      declarebolt(builder, SPLIT, new SplitBolt(), parallelismHint, true);
      declarebolt(builder, OUTPUT, outputBolt, parallelismHint, false);

      return builder;
  }

  private static void declarebolt(TopologyBuilder builder, String boltName, IRichBolt bolt, int parallelism, boolean control) {
      BoltDeclarer declarer = builder.setBolt(boltName, bolt, parallelism)
          .allGrouping(FLOW_LOADER_STREAM, FLOW_LOADER_STREAM)
          .allGrouping("tick", "tick")
          .localOrShuffleGrouping(INITIALIZER, boltName)
          .localOrShuffleGrouping(FILTER, boltName)
          .fieldsGrouping(PARTITION, boltName, new Fields(FLOW_ID, PARTITION))    // guaranteed partitions will always group the same flow for flows that have joins with default partitions.
          .localOrShuffleGrouping(AGGREGATE, boltName)
          .localOrShuffleGrouping(SELECT, boltName)
          .localOrShuffleGrouping(EACH, boltName)
          .localOrShuffleGrouping(SORT, boltName)
          .localOrShuffleGrouping(SWITCH, boltName)
          .localOrShuffleGrouping(SPLIT, boltName)
          .localOrShuffleGrouping(JOIN, boltName);


          if(control) {
            // control stream is all-grouped
            declarer.allGrouping(INITIALIZER, BROADCAST_STREAM + boltName)
                    .allGrouping(FILTER, BROADCAST_STREAM + boltName)
                    .allGrouping(PARTITION, BROADCAST_STREAM + boltName)
                    .allGrouping(AGGREGATE, BROADCAST_STREAM + boltName)
                    .allGrouping(SELECT, BROADCAST_STREAM + boltName)
                    .allGrouping(EACH, BROADCAST_STREAM + boltName)
                    .allGrouping(SORT, BROADCAST_STREAM + boltName)
                    .allGrouping(SWITCH, BROADCAST_STREAM + boltName)
                    .allGrouping(SPLIT, BROADCAST_STREAM + boltName)
                    .allGrouping(JOIN, BROADCAST_STREAM + boltName);

          }
  }

  public static Fields fields = new Fields(FLOW_ID, EVENT, FLOW_OP_IDX, STREAM_NAME, LAST_STREAM);
  public static Fields partitionFields = new Fields(FLOW_ID, EVENT, FLOW_OP_IDX, STREAM_NAME, PARTITION, LAST_STREAM);

  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 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 class SelectOp implements FlowOp {

    public static final String SELECT = "select";
    List<String> fields;

    public SelectOp(List<String> fields) {
        this.fields = fields;
    }
    @Override
    public String getComponentName() {
        return SELECT;
    }

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

    private static final TypeRegistry<String> registry = LEXI_TYPES;

    private Utils() {}

  public static String buildKeyIndexForEvent(Event event, List<String> groupBy) {
    StringBuffer stringBuffer = new StringBuffer();

    if(groupBy == null || groupBy.size() == 0)
      return stringBuffer.toString();  // default partition when no groupBy fields are specified.

    for(String groupField : groupBy) {
      Collection<Tuple> tuples = event.getAll(groupField);
      SortedSet<String> values = new TreeSet<String>();

      if(tuples == null) {
        values.add("");
      } else {
        for(Tuple tuple : tuples)
          values.add(registry.encode(tuple.getValue()));
      }
      stringBuffer.append(groupField + join(values, "") + "|");
    }
    try {
      return stringBuffer.toString();
    } catch (Exception e) {
      return null;
    }
  }


  public static String buildKeyIndexForEvent(String flowId, Event event, List<String> groupBy) {
      return flowId + buildKeyIndexForEvent(event, groupBy);
  }


  public static String getNextStreamFromFlowInfo(Flow flow, String streamName, int idx) {
    return idx+1 < flow.getStream(streamName).getFlowOps().size() ?
        flow.getStream(streamName).getFlowOps().get(idx + 1).getComponentName() : OUTPUT;
  }

  public static boolean hasNextOutput(Flow flow, String streamName, String nextStream) {
    return (nextStream.equals(OUTPUT) && flow.getStream(streamName).isStdOutput()) || !nextStream.equals(OUTPUT);
  }

  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 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 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 org.calrissian.flowmix.api.builder.FlowmixBuilder;
import org.calrissian.flowmix.api.Flow;
import org.calrissian.flowmix.core.model.FlowInfo;
import org.calrissian.flowmix.core.model.op.SelectOp;
import org.calrissian.flowmix.core.support.Utils;
import org.calrissian.mango.domain.event.BaseEvent;
import org.calrissian.mango.domain.event.Event;
import static org.calrissian.flowmix.api.builder.FlowmixBuilder.fields;
import static org.calrissian.flowmix.core.Constants.FLOW_LOADER_STREAM;
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 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());