// Path: src/main/java/org/calrissian/flowmix/api/Aggregator.java
// 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();
// }
// 
// Path: src/main/java/org/calrissian/flowmix/core/model/event/AggregatedEvent.java
// 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 + '\'' +
//             '}';
//   }
// }
// 
// Path: src/main/java/org/calrissian/flowmix/core/support/deque/AggregatorLimitingDeque.java
// public class AggregatorLimitingDeque extends LimitingDeque<WindowItem> {
// 
//   Aggregator aggregator;
// 
//   public AggregatorLimitingDeque(long maxSize, Aggregator aggregator) {
//     super(maxSize);
//     this.aggregator = aggregator;
//   }
// 
//   @Override
//   public boolean offerLast(WindowItem windowItem) {
//     if(size() == getMaxSize())
//       aggregator.evicted(getFirst());
// 
//     return super.offerLast(windowItem);
//   }
// }


import java.util.Collection;
import org.calrissian.flowmix.api.Aggregator;
import org.calrissian.flowmix.core.model.event.AggregatedEvent;
import org.calrissian.flowmix.core.support.deque.AggregatorLimitingDeque;
import org.calrissian.mango.domain.event.Event;
    @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);
    }
  }
// Path: src/main/java/org/calrissian/flowmix/api/Aggregator.java
// 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();
// }
// 
// Path: src/main/java/org/calrissian/flowmix/core/model/event/AggregatedEvent.java
// 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 + '\'' +
//             '}';
//   }
// }
// 
// Path: src/main/java/org/calrissian/flowmix/core/support/deque/AggregatorLimitingDeque.java
// public class AggregatorLimitingDeque extends LimitingDeque<WindowItem> {
// 
//   Aggregator aggregator;
// 
//   public AggregatorLimitingDeque(long maxSize, Aggregator aggregator) {
//     super(maxSize);
//     this.aggregator = aggregator;
//   }
// 
//   @Override
//   public boolean offerLast(WindowItem windowItem) {
//     if(size() == getMaxSize())
//       aggregator.evicted(getFirst());
// 
//     return super.offerLast(windowItem);
//   }
// }


// Path: src/main/java/org/calrissian/flowmix/core/support/window/AggregatorWindow.java
import java.util.Collection;
import org.calrissian.flowmix.api.Aggregator;
import org.calrissian.flowmix.core.model.event.AggregatedEvent;
import org.calrissian.flowmix.core.support.deque.AggregatorLimitingDeque;
import org.calrissian.mango.domain.event.Event;

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