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 interface FlowOp extends Serializable {

    String getComponentName();
} public class SortOp implements FlowOp, RequiresPartitioning {

  public static final String SORT = "sort";

  private List<Pair<String, Order>> sortBy;
  private boolean clearOnTrigger = false;   // this determines whether or or not the dataset is sorted all the time
  private Policy evictionPolicy;
  private long evictionThreshold;
  private Policy triggerPolicy;
  private long triggerThreshold;
  private boolean progressive;

  public SortOp(List<Pair<String,Order>> sortBy, boolean clearOnTrigger, Policy evictionPolicy, long evictionThreshold, Policy triggerPolicy, long triggerThreshold, boolean progressive) {
    this.sortBy = sortBy;
    this.clearOnTrigger = clearOnTrigger;
    this.evictionPolicy = evictionPolicy;
    this.evictionThreshold = evictionThreshold;
    this.triggerPolicy = triggerPolicy;
    this.triggerThreshold = triggerThreshold;
    this.progressive = progressive;
  }


  public boolean isProgressive() {
    return progressive;
  }

  public List<Pair<String,Order>> getSortBy() {
    return sortBy;
  }

  public boolean isClearOnTrigger() {
    return clearOnTrigger;
  }

  public Policy getEvictionPolicy() {
    return evictionPolicy;
  }

  public long getEvictionThreshold() {
    return evictionThreshold;
  }

  public Policy getTriggerPolicy() {
    return triggerPolicy;
  }

  public long getTriggerThreshold() {
    return triggerThreshold;
  }


  @Override
  public boolean equals(Object o) {

    if (this == o) return true;
    if (o == null || getClass() != o.getClass()) return false;

    SortOp sortOp = (SortOp) o;

    if (clearOnTrigger != sortOp.clearOnTrigger) return false;
    if (evictionThreshold != sortOp.evictionThreshold) return false;
    if (triggerThreshold != sortOp.triggerThreshold) return false;
    if (evictionPolicy != sortOp.evictionPolicy) return false;
    if (sortBy != null ? !sortBy.equals(sortOp.sortBy) : sortOp.sortBy != null) return false;
    if (triggerPolicy != sortOp.triggerPolicy) return false;

    return true;
  }

  @Override
  public int hashCode() {
    int result = sortBy != null ? sortBy.hashCode() : 0;
    result = 31 * result + (clearOnTrigger ? 1 : 0);
    result = 31 * result + (evictionPolicy != null ? evictionPolicy.hashCode() : 0);
    result = 31 * result + (int) (evictionThreshold ^ (evictionThreshold >>> 32));
    result = 31 * result + (triggerPolicy != null ? triggerPolicy.hashCode() : 0);
    result = 31 * result + (int) (triggerThreshold ^ (triggerThreshold >>> 32));
    return result;
  }

  @Override
  public String toString() {
    return "SortOp{" +
            "sortBy=" + sortBy +
            ", clearOnTrigger=" + clearOnTrigger +
            ", evictionPolicy=" + evictionPolicy +
            ", evictionThreshold=" + evictionThreshold +
            ", triggerPolicy=" + triggerPolicy +
            ", triggerThreshold=" + triggerThreshold +
            '}';
  }

  @Override
  public String getComponentName() {
    return SORT;
  }
} public class EventSortByComparator implements Comparator<WindowItem> {

    private static ComparableComparator comparator = new ComparableComparator();
    private List<Pair<String,Order>> sortBy;

    public EventSortByComparator(List<Pair<String,Order>> sortBy) {
        checkNotNull(sortBy);
        checkArgument(sortBy.size() > 0);
        this.sortBy = sortBy;
    }

    @Override
    public int compare(WindowItem windowItem, WindowItem windowItem2) {

        for(Pair<String,Order> sortField : sortBy) {

            Object val1 = windowItem.getEvent().get(sortField.getOne()).getValue();
            Object val2 = windowItem2.getEvent().get(sortField.getOne()).getValue();

            int compare = comparator.compare(val1,val2);

            if(compare != 0)
                return sortField.getTwo() == Order.DESC ? compare * -1 : compare;
        }

        return 0; // if they are the same then they're the same...
    }
} public class SortedWindow extends Window {

  private Comparator<WindowItem> comparator;
  private boolean sortOnGet = false;

  public SortedWindow(String groupedIndex, Comparator<WindowItem> comparator, long size, boolean sortOnGet) {
    this.groupedIndex = groupedIndex;
    this.sortOnGet = sortOnGet;

    if(!sortOnGet)
      events = new LimitingPriorityDeque<WindowItem>(size, comparator);  // WARNING
    else
      events = new LimitingDeque<WindowItem>(size);

    this.comparator = comparator;
  }

  /**
   * Used for count-based expiration
   */
  public WindowItem expire() {
    return events.removeLast();
  }

  public SortedWindow(String groupedIndex, Comparator<WindowItem> comparator, boolean sortOnGet) {
    this.groupedIndex = groupedIndex;
    this.sortOnGet = sortOnGet;

    if(!sortOnGet)
      events = new PriorityBlockingDeque<WindowItem>(comparator);
    else
      events = new LinkedBlockingDeque<WindowItem>();

    this.comparator = comparator;
  }

  /**
   * Returns a sorted view of the events
   */
  @Override
  public Iterable<WindowItem> getEvents() {
    if(sortOnGet) {
      WindowItem[] items = events.toArray(new WindowItem[]{});
      sort(items, comparator);
      return asList(items);
    } else {
      return super.getEvents();
    }
  }
} public class Window {

  protected String groupedIndex;        // a unique key given to the groupBy field/value combinations in the window buffer

  protected Deque<WindowItem> events;     // using standard array list for proof of concept.
                                             // Circular buffer needs to be used after concept is proven
  protected int triggerTicks = 0;

  /**
   * A progressive window buffer which automatically evicts by count
   */
  public Window(String groupedIndex, long size) {
    this(groupedIndex);
    events = new LimitingDeque<WindowItem>(size);
  }

  public Window() {
  }

  public Window(String groupedIndex) {
    events = new LinkedBlockingDeque<WindowItem>();
    this.groupedIndex = groupedIndex;
  }

  public WindowItem add(Event event, String previousStream) {
        WindowItem item = new WindowItem(event, currentTimeMillis(), previousStream);
        events.add(item);
        return item;
    }

    /**
     * Used for age-based expiration
     */
    public void timeEvict(long thresholdInSeconds) {
        while(events != null && events.peek() != null &&
                (System.currentTimeMillis() - events.peek().getTimestamp()) >= (thresholdInSeconds * 1000))
            events.poll();
    }

    public void resetTriggerTicks() {
        triggerTicks = 0;
    }

    public int getTriggerTicks() {
        return triggerTicks;
    }

    public void incrTriggerTicks() {
        triggerTicks += 1;
    }

    public String getGroupedIndex() {
        return groupedIndex;
    }

    /**
     * Returns the difference(in millis) between the HEAD & TAIL timestamps.
     */
    public long timeRange() {
        if(events.size() <= 1)
            return -1;
        return events.getLast().getTimestamp() - events.getFirst().getTimestamp();
    }

    public Iterable<WindowItem> getEvents() {
        return events;
    }

    public int size() {
        return events.size();
    }

    /**
     * Used for count-based expiration
     */
    public WindowItem expire() {
        return events.removeLast();
    }


    public void clear() {
        events.clear();
    }

    @Override
    public String toString() {
        return "Window{" +
                "groupedIndex='" + groupedIndex + '\'' +
                ", size=" + events.size() +
                ", events=" + events +
                ", triggertTicks=" + triggerTicks +
                '}';
    }
} public class WindowItem {
    Event event;
    long timestamp;
    String previousStream;

    public WindowItem(Event event, long timestamp, String previousStream) {
        this.event = event;
        this.timestamp = timestamp;
        this.previousStream = previousStream;
    }

    public Event getEvent() {
        return event;
    }

    public long getTimestamp() {
        return timestamp;
    }

    public String getPreviousStream() {
      return previousStream;
    }

    @Override
    public String toString() {
        return "WindowItem{" +
                "event=" + event +
                ", timestamp=" + timestamp +
                '}';
    }
} 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 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);
}
import java.util.Collection;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
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.op.FlowOp;
import org.calrissian.flowmix.core.model.op.SortOp;
import org.calrissian.flowmix.core.support.EventSortByComparator;
import org.calrissian.flowmix.core.support.window.SortedWindow;
import org.calrissian.flowmix.core.support.window.Window;
import org.calrissian.flowmix.core.support.window.WindowItem;
import static java.util.Collections.singleton;
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.core.support.Utils.exportsToOtherStreams;
import static org.calrissian.flowmix.core.support.Utils.getNextStreamFromFlowInfo;
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;




/**
 * Sorts a window. This is similar to the Sort operator in InfoSphere Streams.
 *
 * As in infosphere streams:
 *
 * - Sliding windows only allow count-based trigger and count-based expiration
 * - Tumbling windows allows count, delta, and time based trigger
 *
 * TODO: Need to begin enforcing the different accepted trigger vs. eviction policies
 */
public class SortBolt extends BaseRichBolt {

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