@@ -0,0 +1,2124 @@
+package quarks.analytics.sensors;%0A%0Aimport java.util.Objects;%0A%0Aimport quarks.function.Predicate;%0A%0A/**%0A * A generic %22valve%22 %7B@link Predicate%7D.%0A * %3Cp%3E%0A * A valve predicate accepts tuples when its state is %7B@link State#OPEN%7D,%0A * otherwise it rejects tuples.%0A * %3C/p%3E%3Cp%3E%0A * A valve is typically used to dynamically control whether or not%0A * some downstream tuple processing is enabled.  A decision to change the%0A * state of the valve may be a result of local analytics or an external%0A * command.%0A * %3Cbr%3E%0A * E.g., a Valve might be used to control whether or not logging%0A * of tuples is enabled.%0A * %3Cpre%3E%7B@code%0A * TStream%3CJsonObject%3E stream = ...;%0A * %0A * Valve%3CJsonObject%3E valve = new Valve%3C%3E(Valve.State.CLOSED);%0A * stream.filter(valve).sink(someTupleLoggingConsumer);%0A *                                 %0A * // from some analytic or device command handler...%0A *     valve.setState(Valve.State.OPEN);%0A * %7D%3C/pre%3E%0A * %3C/p%3E%0A *%0A * @param %3CT%3E tuple type%0A * @see Filters#valve(quarks.topology.TStream, State)%0A */%0Apublic class Valve%3CT%3E implements Predicate%3CT%3E %7B%0A    private static final long serialVersionUID = 1L;%0A    private transient State state = State.OPEN;%0A    %0A    /**%0A     * The valve state.%0A     */%0A    public enum State %7B %0A        /** accept tuples */ OPEN, /** reject tuples */ CLOSED %7D;%0A%0A    /**%0A     * Create a new Valve Predicate%0A     * %3Cp%3E%0A     * Same as %7B@code Valve(State.OPEN)%7D%0A     */%0A    public Valve() %7B%0A        this(State.OPEN);%0A    %7D%0A    %0A    /**%0A     * Create a new Valve Predicate%0A     * %3Cp%3E%0A     * @param state the initial %7B@link State%7D%0A     */%0A    public Valve(State state) %7B%0A        setState(state);%0A    %7D%0A    %0A    /**%0A     * Set the valve state%0A     * @param state the state of the valve%0A     */%0A    public synchronized void setState(State state) %7B%0A        Objects.requireNonNull(state, %22state%22);%0A        this.state = state;%0A    %7D%0A%0A    @Override%0A    public boolean test(T value) %7B%0A        return state == State.OPEN;%0A    %7D%0A%0A    /**%0A     * Returns a String for development/debug support.  Content subject to change.%0A     */%0A    @Override%0A    public String toString() %7B%0A        return %22state=%22+state;%0A    %7D%0A    %0A%7D

add Valve Predicate