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    /**
040     * Take in account the watermark.
041     *
042     * @implNote Not thread safe.
043     */
044    public long mark(long watermarkValue) {
045        return mark(Watermark.ofValue(watermarkValue));
046    }
047
048    /**
049     * Take in account the watermark.
050     *
051     * @implNote Not thread safe.
052     */
053    public long mark(Watermark watermark) {
054        if (Watermark.LOWEST.equals(low)) {
055            low = high = watermark;
056        } else if (watermark.compareTo(low) < 0) {
057            if (watermark.compareTo(lowest) < 0) {
058                // low watermark must increase to be monotonic
059                if (log.isTraceEnabled()) {
060                    log.trace("receive too low watermark, rejected " + watermark + " lowest: " + lowest);
061                }
062                low = lowest;
063            } else {
064                low = watermark;
065            }
066        }
067        if (watermark.compareTo(high) > 0) {
068            high = watermark;
069        }
070        return low.getValue();
071    }
072
073    /**
074     * Move the low watermark to the highest mark. Returns the low watermark that should be monotonic (the value
075     * returned here never decrease).
076     *
077     * @implNote Not thread safe.
078     */
079    public long checkpoint() {
080        low = Watermark.completedOf(high);
081        lowest = low;
082        return low.getValue();
083    }
084
085    public boolean isDone(long timestamp) {
086        return low.isDone(timestamp);
087    }
088
089    /**
090     * Returns the low mark. The value can decrease but not under the last checkpoint value.
091     *
092     * @implNote 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}