001/*
002 * (C) Copyright 2017 Nuxeo SA (http://nuxeo.com/) and others.
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License");
005 * you may not use this file except in compliance with the License.
006 * You may obtain a copy of the License at
007 *
008 *     http://www.apache.org/licenses/LICENSE-2.0
009 *
010 * Unless required by applicable law or agreed to in writing, software
011 * distributed under the License is distributed on an "AS IS" BASIS,
012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013 * See the License for the specific language governing permissions and
014 * limitations under the License.
015 *
016 * Contributors:
017 *     bdelbosc
018 */
019package org.nuxeo.lib.stream.computation.internals;
020
021import org.apache.commons.logging.Log;
022import org.apache.commons.logging.LogFactory;
023import org.nuxeo.lib.stream.computation.Watermark;
024
025/**
026 * Keep track of minimum and maximum watermark level. On checkpoint move the low watermark to the previous maximum mark.
027 *
028 * @since 9.3
029 */
030public class WatermarkMonotonicInterval {
031    private static final Log log = LogFactory.getLog(WatermarkMonotonicInterval.class);
032
033    protected volatile Watermark low = Watermark.LOWEST;
034
035    protected Watermark lowest = Watermark.LOWEST;
036
037    protected Watermark high = Watermark.LOWEST;
038
039    public WatermarkMonotonicInterval() {
040    }
041
042    /**
043     * Take in account the watermark.<br/>
044     * Not thread safe.
045     */
046    public long mark(long watermarkValue) {
047        return mark(Watermark.ofValue(watermarkValue));
048    }
049
050    /**
051     * Take in account the watermark.<br/>
052     * Not thread safe.
053     */
054    public long mark(Watermark watermark) {
055        if (Watermark.LOWEST.equals(low)) {
056            low = high = watermark;
057        } else if (watermark.compareTo(low) < 0) {
058            if (watermark.compareTo(lowest) < 0) {
059                // low watermark must increase to be monotonic
060                if (log.isTraceEnabled()) {
061                    log.trace("receive too low watermark, rejected " + watermark + " lowest: " + lowest);
062                }
063                low = lowest;
064            } else {
065                low = watermark;
066            }
067        }
068        if (watermark.compareTo(high) > 0) {
069            high = watermark;
070        }
071        return low.getValue();
072    }
073
074    /**
075     * Move the low watermark to the highest mark. Returns the low watermark that should be monotonic (the value
076     * returned here never decrease).<br/>
077     * Not thread safe.
078     */
079    public long checkpoint() {
080        low = Watermark.completedOf(high);
081        lowest = low;
082        // System.out.println(low);
083        return low.getValue();
084    }
085
086    public boolean isDone(long timestamp) {
087        return low.isDone(timestamp);
088    }
089
090    /**
091     * Returns the low mark. The value can decrease but not under the last checkpoint value.<br/>
092     * Thread safe usage.
093     */
094    public Watermark getLow() {
095        return low;
096    }
097
098    public Watermark getHigh() {
099        return high;
100    }
101
102    @Override
103    public String toString() {
104        return "WatermarkInterval{" + "low=" + low + ", high=" + high + '}';
105    }
106
107}