public abstract class Trigger<T, W extends Window> implements Serializable {

    /**<br/>
     * Called for every element that gets added to a pane. The result of this will determine<br/>
     * whether the pane is evaluated to emit results.<br/>
     *<br/>
     * @param element The element that arrived.<br/>
     * @param timestamp The timestamp of the element that arrived.<br/>
     * @param window The window to which the element is being added.<br/>
     * @param ctx A context object that can be used to register timer callbacks.<br/>
     */<br/>
    public abstract TriggerResult onElement(T element, long timestamp, W window, TriggerContext ctx) throws Exception;

    /**<br/>
     * Called when a processing-time timer that was set using the trigger context fires.<br/>
     *<br/>
     * @param time The timestamp at which the timer fired.<br/>
     * @param window The window for which the timer fired.<br/>
     * @param ctx A context object that can be used to register timer callbacks.<br/>
     */<br/>
    public abstract TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception;

    /**<br/>
     * Called when an event-time timer that was set using the trigger context fires.<br/>
     *<br/>
     * @param time The timestamp at which the timer fired.<br/>
     * @param window The window for which the timer fired.<br/>
     * @param ctx A context object that can be used to register timer callbacks.<br/>
     */<br/>
    public abstract TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception;

    /**<br/>
     * Called when several windows have been merged into one window by the<br/>
     * {@link org.apache.flink.streaming.api.windowing.assigners.WindowAssigner}.<br/>
     *<br/>
     * @param window The new window that results from the merge.<br/>
     * @param ctx A context object that can be used to register timer callbacks and access state.<br/>
     */<br/>
    public TriggerResult onMerge(W window, OnMergeContext ctx) throws Exception {<br/>
        throw new RuntimeException("This trigger does not support merging.");<br/>
    }

/**<br/>
 * A {@link Trigger} that fires once the current system time passes the end of the window<br/>
 * to which a pane belongs.<br/>
 */<br/>
public class ProcessingTimeTrigger implements Trigger<Object, TimeWindow> {<br/>
    private static final long serialVersionUID = 1L;

    private ProcessingTimeTrigger() {}

    @Override<br/>
    public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) {<br/>
        ctx.registerProcessingTimeTimer(window.maxTimestamp()); //对于processingTime，element的trigger时间是current+window，所以这里需要注册定时器去触发<br/>
        return TriggerResult.CONTINUE;<br/>
    }

    @Override<br/>
    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {<br/>
        return TriggerResult.CONTINUE;<br/>
    }

    @Override<br/>
    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) {//触发后调用<br/>
        return TriggerResult.FIRE_AND_PURGE;<br/>
    }

    @Override<br/>
    public String toString() {<br/>
        return "ProcessingTimeTrigger()";<br/>
    }

    /**<br/>
    * Creates a new trigger that fires once system time passes the end of the window.<br/>
    */<br/>
    public static ProcessingTimeTrigger create() {<br/>
        return new ProcessingTimeTrigger();<br/>
    }<br/>
}

public class CountTrigger<W extends Window> extends Trigger<Object, W> {

    private final long maxCount;

    private final ReducingStateDescriptor<Long> stateDesc =<br/>
            new ReducingStateDescriptor<>("count", new Sum(), LongSerializer.INSTANCE);

    private CountTrigger(long maxCount) {<br/>
        this.maxCount = maxCount;<br/>
    }

    @Override<br/>
    public TriggerResult onElement(Object element, long timestamp, W window, TriggerContext ctx) throws Exception {<br/>
        ReducingState<Long> count = ctx.getPartitionedState(stateDesc); //从backend取出conunt state<br/>
        count.add(1L); //加1<br/>
        if (count.get() >= maxCount) {<br/>
            count.clear();<br/>
            return TriggerResult.FIRE;<br/>
        }<br/>
        return TriggerResult.CONTINUE;<br/>
    }

    @Override<br/>
    public TriggerResult onEventTime(long time, W window, TriggerContext ctx) {<br/>
        return TriggerResult.CONTINUE;<br/>
    }

    @Override<br/>
    public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception {<br/>
        return TriggerResult.CONTINUE;<br/>
    }

    @Override<br/>
    public TriggerResult onMerge(W window, OnMergeContext ctx) throws Exception {<br/>
        ctx.mergePartitionedState(stateDesc); //先调用merge，底层backend里面的window进行merge<br/>
        ReducingState<Long> count = ctx.getPartitionedState(stateDesc); //merge后再取出state，count，进行判断<br/>
        if (count.get() >= maxCount) {<br/>
            return TriggerResult.FIRE;<br/>
        }<br/>
        return TriggerResult.CONTINUE;<br/>
    }

enum TriggerResult {<br/>
    CONTINUE(false, false), FIRE_AND_PURGE(true, true), FIRE(true, false), PURGE(false, true);

    private final boolean fire;<br/>
    private final boolean purge;<br/>
}

/**<br/>
 * Merges two {@code TriggerResults}. This specifies what should happen if we have<br/>
 * two results from a Trigger, for example as a result from<br/>
 * {@link Trigger#onElement(Object, long, Window, Trigger.TriggerContext)} and<br/>
 * {@link Trigger#onEventTime(long, Window, Trigger.TriggerContext)}.<br/>
 *<br/>
 * <p><br/>
 * For example, if one result says {@code CONTINUE} while the other says {@code FIRE}<br/>
 * then {@code FIRE} is the combined result;<br/>
 */<br/>
public static TriggerResult merge(TriggerResult a, TriggerResult b) {<br/>
    if (a.purge || b.purge) {<br/>
        if (a.fire || b.fire) {<br/>
            return FIRE_AND_PURGE;<br/>
        } else {<br/>
            return PURGE;<br/>
        }<br/>
    } else if (a.fire || b.fire) {<br/>
        return FIRE;<br/>
    } else {<br/>
        return CONTINUE;<br/>
    }<br/>
}

/**<br/>
     * A context object that is given to {@link Trigger} methods to allow them to register timer<br/>
     * callbacks and deal with state.<br/>
     */<br/>
    public interface TriggerContext {

        long getCurrentProcessingTime();<br/>
        long getCurrentWatermark();

        /**<br/>
         * Register a system time callback. When the current system time passes the specified<br/>
         * time {@link Trigger#onProcessingTime(long, Window, TriggerContext)} is called with the time specified here.<br/>
         *<br/>
         * @param time The time at which to invoke {@link Trigger#onProcessingTime(long, Window, TriggerContext)}<br/>
         */<br/>
        void registerProcessingTimeTimer(long time);<br/>
        void registerEventTimeTimer(long time);

        void deleteProcessingTimeTimer(long time);<br/>
        void deleteEventTimeTimer(long time);

        <S extends State> S getPartitionedState(StateDescriptor<S, ?> stateDescriptor);<br/>
    }

public interface OnMergeContext extends TriggerContext {<br/>
    <S extends MergingState<?, ?>> void mergePartitionedState(StateDescriptor<S, ?> stateDescriptor);<br/>
}

/**<br/>
 * {@code Context} is a utility for handling {@code Trigger} invocations. It can be reused<br/>
 * by setting the {@code key} and {@code window} fields. No internal state must be kept in<br/>
 * the {@code Context}<br/>
 */<br/>
public class Context implements Trigger.OnMergeContext {<br/>
    protected K key; //Context对应的window上下文<br/>
    protected W window;

    protected Collection<W> mergedWindows; //onMerge中被赋值

    @SuppressWarnings("unchecked")<br/>
    public <S extends State> S getPartitionedState(StateDescriptor<S, ?> stateDescriptor) {<br/>
        try {<br/>
            return WindowOperator.this.getPartitionedState(window, windowSerializer, stateDescriptor); //从backend里面读出改window的状态，即window buffer<br/>
        } catch (Exception e) {<br/>
            throw new RuntimeException("Could not retrieve state", e);<br/>
        }<br/>
    }

    @Override<br/>
    public <S extends MergingState<?, ?>> void mergePartitionedState(StateDescriptor<S, ?> stateDescriptor) {<br/>
        if (mergedWindows != null && mergedWindows.size() > 0) {<br/>
            try {<br/>
                WindowOperator.this.getStateBackend().mergePartitionedStates(window, //在backend层面把mergedWindows merge到window中<br/>
                        mergedWindows,<br/>
                        windowSerializer,<br/>
                        stateDescriptor);<br/>
            } catch (Exception e) {<br/>
                throw new RuntimeException("Error while merging state.", e);<br/>
            }<br/>
        }<br/>
    }

    @Override<br/>
    public void registerProcessingTimeTimer(long time) {<br/>
        Timer<K, W> timer = new Timer<>(time, key, window);<br/>
        // make sure we only put one timer per key into the queue<br/>
        if (processingTimeTimers.add(timer)) {<br/>
            processingTimeTimersQueue.add(timer);<br/>
            //If this is the first timer added for this timestamp register a TriggerTask<br/>
            if (processingTimeTimerTimestamps.add(time, 1) == 0) { //如果这个window是第一次注册的话<br/>
                ScheduledFuture<?> scheduledFuture = WindowOperator.this.registerTimer(time, WindowOperator.this); //对于processTime必须注册定时器主动触发<br/>
                processingTimeTimerFutures.put(time, scheduledFuture);<br/>
            }<br/>
        }<br/>
    }

    @Override<br/>
    public void registerEventTimeTimer(long time) {<br/>
        Timer<K, W> timer = new Timer<>(time, key, window);<br/>
        if (watermarkTimers.add(timer)) {<br/>
            watermarkTimersQueue.add(timer);<br/>
        }<br/>
    }

    //封装一遍trigger的接口，并把self作为context传入trigger的接口中<br/>
    public TriggerResult onElement(StreamRecord<IN> element) throws Exception {<br/>
        return trigger.onElement(element.getValue(), element.getTimestamp(), window, this);<br/>
    }

    public TriggerResult onProcessingTime(long time) throws Exception {<br/>
        return trigger.onProcessingTime(time, window, this);<br/>
    }

    public TriggerResult onEventTime(long time) throws Exception {<br/>
        return trigger.onEventTime(time, window, this);<br/>
    }

    public TriggerResult onMerge(Collection<W> mergedWindows) throws Exception {<br/>
        this.mergedWindows = mergedWindows;<br/>
        return trigger.onMerge(window, this);<br/>
    }

}

/**<br/>
 * An {@code Evictor} can remove elements from a pane before it is being processed and after<br/>
 * window evaluation was triggered by a<br/>
 * {@link org.apache.flink.streaming.api.windowing.triggers.Trigger}.<br/>
 *<br/>
 * <p><br/>
 * A pane is the bucket of elements that have the same key (assigned by the<br/>
 * {@link org.apache.flink.api.java.functions.KeySelector}) and same {@link Window}. An element can<br/>
 * be in multiple panes of it was assigned to multiple windows by the<br/>
 * {@link org.apache.flink.streaming.api.windowing.assigners.WindowAssigner}. These panes all<br/>
 * have their own instance of the {@code Evictor}.<br/>
 *<br/>
 * @param <T> The type of elements that this {@code Evictor} can evict.<br/>
 * @param <W> The type of {@link Window Windows} on which this {@code Evictor} can operate.<br/>
 */<br/>
public interface Evictor<T, W extends Window> extends Serializable {

    /**<br/>
     * Computes how many elements should be removed from the pane. The result specifies how<br/>
     * many elements should be removed from the beginning.<br/>
     *<br/>
     * @param elements The elements currently in the pane.<br/>
     * @param size The current number of elements in the pane.<br/>
     * @param window The {@link Window}<br/>
     */<br/>
    int evict(Iterable<StreamRecord<T>> elements, int size, W window);<br/>
}

/**<br/>
 * An {@link Evictor} that keeps only a certain amount of elements.<br/>
 *<br/>
 * @param <W> The type of {@link Window Windows} on which this {@code Evictor} can operate.<br/>
 */<br/>
public class CountEvictor<W extends Window> implements Evictor<Object, W> {<br/>
    private static final long serialVersionUID = 1L;

    private final long maxCount;

    private CountEvictor(long count) {<br/>
        this.maxCount = count;<br/>
    }

    @Override<br/>
    public int evict(Iterable<StreamRecord<Object>> elements, int size, W window) {<br/>
        if (size > maxCount) {<br/>
            return (int) (size - maxCount);<br/>
        } else {<br/>
            return 0;<br/>
        }<br/>
    }

    /**<br/>
    * Creates a {@code CountEvictor} that keeps the given number of elements.<br/>
    *<br/>
    * @param maxCount The number of elements to keep in the pane.<br/>
    */<br/>
    public static <W extends Window> CountEvictor<W> of(long maxCount) {<br/>
        return new CountEvictor<>(maxCount);<br/>
    }<br/>
}

/**<br/>
 * An {@link Evictor} that keeps elements for a certain amount of time. Elements older<br/>
 * than {@code current_time - keep_time} are evicted.<br/>
 *<br/>
 * @param <W> The type of {@link Window Windows} on which this {@code Evictor} can operate.<br/>
 */<br/>
public class TimeEvictor<W extends Window> implements Evictor<Object, W> {<br/>
    private static final long serialVersionUID = 1L;

    private final long windowSize;

    public TimeEvictor(long windowSize) {<br/>
        this.windowSize = windowSize;<br/>
    }

    @Override<br/>
    public int evict(Iterable<StreamRecord<Object>> elements, int size, W window) {<br/>
        int toEvict = 0;<br/>
        long currentTime = Iterables.getLast(elements).getTimestamp();<br/>
        long evictCutoff = currentTime - windowSize;<br/>
        for (StreamRecord<Object> record: elements) {<br/>
            if (record.getTimestamp() > evictCutoff) {<br/>
                break;<br/>
            }<br/>
            toEvict++;<br/>
        }<br/>
        return toEvict;<br/>
    }<br/>
}