SortedMovingWindow extends AbstractSlidingWindow<T, List<T>> { @SuppressWarnings("unchecked") @Override public void endWindow() { super.endWindow(); tuplesInCurrentStreamWindow = new LinkedList<T>(); if (lastExpiredWindowState == null) { return; } for (T expiredTuple : lastExpiredWindowState) { PriorityQueue<T> sortedListForE = sortedListInSlidingWin.get(function.apply(expiredTuple)); for (Iterator<T> iterator = sortedListForE.iterator(); iterator.hasNext();) { T minElemInSortedList = iterator.next(); int k = 0; if (comparator == null) { if (expiredTuple instanceof Comparable) { k = ((Comparable<T>)expiredTuple).compareTo(minElemInSortedList); } else { errorOutput.emit(expiredTuple); throw new IllegalArgumentException("Operator \"" + ClassUtils.getShortClassName(this.getClass()) + "\" encounters an invalid tuple " + expiredTuple + "\nNeither the tuple is comparable Nor Comparator is specified!"); } } else { k = comparator.compare(expiredTuple, minElemInSortedList); } if (k < 0) { break; } else { outputPort.emit(minElemInSortedList); iterator.remove(); } } } }  @Override List<T> createWindowState(); @SuppressWarnings("unchecked") @Override void endWindow(); void setComparator(Comparator<T> comparator); void setFunction(Function<T, K> function); transient DefaultOutputPort<T> outputPort; @OutputPortFieldAnnotation(error = true)
  transient DefaultOutputPort<T> errorOutput; }
@Test public void testSortingSimpleNumberTuple() { SortedMovingWindow<Integer, Null> smw = new SortedMovingWindow<Integer, Null>(); CollectorTestSink<Object> testSink = new CollectorTestSink<Object>(); smw.outputPort.setSink(testSink); smw.setup(null); smw.setWindowSize(2); emitObjects(smw, new Integer[][]{{1,3}, {2,5}, {4}, {6}}); smw.beginWindow(4); smw.endWindow(); smw.beginWindow(5); smw.endWindow(); Assert.assertEquals(Lists.newArrayList(1, 2, 3, 4, 5, 6), testSink.collectedTuples); }