| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| package org.apache.jackrabbit.oak.jcr.observation; |
| |
| import static com.google.common.base.Preconditions.checkNotNull; |
| import static com.google.common.base.Preconditions.checkState; |
| import static java.util.Collections.emptyMap; |
| import static org.apache.jackrabbit.api.stats.RepositoryStatistics.Type.OBSERVATION_EVENT_COUNTER; |
| import static org.apache.jackrabbit.api.stats.RepositoryStatistics.Type.OBSERVATION_EVENT_DURATION; |
| import static org.apache.jackrabbit.oak.plugins.observation.filter.VisibleFilter.VISIBLE_FILTER; |
| import static org.apache.jackrabbit.oak.spi.observation.ChangeSet.COMMIT_CONTEXT_OBSERVATION_CHANGESET; |
| import static org.apache.jackrabbit.oak.spi.whiteboard.WhiteboardUtils.registerMBean; |
| import static org.apache.jackrabbit.oak.spi.whiteboard.WhiteboardUtils.scheduleWithFixedDelay; |
| |
| import java.util.Map; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.concurrent.atomic.AtomicReference; |
| |
| import javax.jcr.observation.Event; |
| import javax.jcr.observation.EventIterator; |
| import javax.jcr.observation.EventListener; |
| |
| import org.apache.jackrabbit.api.jmx.EventListenerMBean; |
| import org.apache.jackrabbit.commons.observation.ListenerTracker; |
| import org.apache.jackrabbit.oak.api.ContentSession; |
| import org.apache.jackrabbit.oak.api.blob.BlobAccessProvider; |
| import org.apache.jackrabbit.oak.commons.PerfLogger; |
| import org.apache.jackrabbit.oak.namepath.NamePathMapper; |
| import org.apache.jackrabbit.oak.plugins.observation.CommitRateLimiter; |
| import org.apache.jackrabbit.oak.plugins.observation.Filter; |
| import org.apache.jackrabbit.oak.plugins.observation.FilteringAwareObserver; |
| import org.apache.jackrabbit.oak.plugins.observation.FilteringDispatcher; |
| import org.apache.jackrabbit.oak.plugins.observation.FilteringObserver; |
| import org.apache.jackrabbit.oak.plugins.observation.filter.EventFilter; |
| import org.apache.jackrabbit.oak.plugins.observation.filter.FilterConfigMBean; |
| import org.apache.jackrabbit.oak.plugins.observation.filter.FilterProvider; |
| import org.apache.jackrabbit.oak.plugins.observation.filter.Filters; |
| import org.apache.jackrabbit.oak.plugins.observation.filter.ChangeSetFilter; |
| import org.apache.jackrabbit.oak.spi.commit.BackgroundObserver; |
| import org.apache.jackrabbit.oak.spi.commit.BackgroundObserverMBean; |
| import org.apache.jackrabbit.oak.spi.commit.CommitContext; |
| import org.apache.jackrabbit.oak.spi.commit.CommitInfo; |
| import org.apache.jackrabbit.oak.spi.commit.Observer; |
| import org.apache.jackrabbit.oak.spi.observation.ChangeSet; |
| import org.apache.jackrabbit.oak.spi.state.NodeState; |
| import org.apache.jackrabbit.oak.spi.whiteboard.CompositeRegistration; |
| import org.apache.jackrabbit.oak.spi.whiteboard.Registration; |
| import org.apache.jackrabbit.oak.spi.whiteboard.Whiteboard; |
| import org.apache.jackrabbit.oak.spi.whiteboard.WhiteboardExecutor; |
| import org.apache.jackrabbit.oak.stats.Clock; |
| import org.apache.jackrabbit.oak.stats.MeterStats; |
| import org.apache.jackrabbit.oak.stats.StatisticManager; |
| import org.apache.jackrabbit.oak.stats.TimerStats; |
| import org.apache.jackrabbit.stats.TimeSeriesMax; |
| import org.jetbrains.annotations.NotNull; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import com.google.common.collect.ImmutableMap; |
| import com.google.common.util.concurrent.Monitor; |
| import com.google.common.util.concurrent.Monitor.Guard; |
| |
| /** |
| * A {@code ChangeProcessor} generates observation {@link javax.jcr.observation.Event}s |
| * based on a {@link FilterProvider filter} and delivers them to an {@link EventListener}. |
| * <p> |
| * After instantiation a {@code ChangeProcessor} must be started in order to start |
| * delivering observation events and stopped to stop doing so. |
| */ |
| class ChangeProcessor implements FilteringAwareObserver { |
| private static final Logger LOG = LoggerFactory.getLogger(ChangeProcessor.class); |
| private static final PerfLogger PERF_LOGGER = new PerfLogger( |
| LoggerFactory.getLogger(ChangeProcessor.class.getName() + ".perf")); |
| |
| private enum FilterResult { |
| /** marks a commit as to be included, ie delivered. |
| * It's okay to falsely mark a commit as included, |
| * since filtering (as part of converting to events) |
| * will be applied at a later stage again. */ |
| INCLUDE, |
| /** mark a commit as not of interest to this ChangeProcessor. |
| * Exclusion is definite, ie it's not okay to falsely |
| * mark a commit as excluded */ |
| EXCLUDE, |
| /** mark a commit as included but indicate that this |
| * is not a result of prefiltering but that prefiltering |
| * was skipped/not applicable for some reason */ |
| PREFILTERING_SKIPPED |
| } |
| |
| /** |
| * Fill ratio of the revision queue at which commits should be delayed |
| * (conditional of {@code commitRateLimiter} being non {@code null}). |
| */ |
| public static final double DELAY_THRESHOLD; |
| |
| /** |
| * Maximal number of milli seconds a commit is delayed once {@code DELAY_THRESHOLD} |
| * kicks in. |
| */ |
| public static final int MAX_DELAY; |
| |
| //It'd would have been more useful to have following 2 properties as instance variables |
| //which got set by tests. But, the tests won't get a handle to the actual instance, so |
| //static-members it is. |
| /** |
| * Number of milliseconds to wait before issuing consecutive queue full warn messages |
| * Controlled by command line property "oak.observation.full-queue.warn.interval". |
| * Note, the command line parameter is wait interval in minutes. |
| */ |
| static long QUEUE_FULL_WARN_INTERVAL = TimeUnit.MINUTES.toMillis(Integer |
| .getInteger("oak.observation.full-queue.warn.interval", 30)); |
| static Clock clock = Clock.SIMPLE; |
| |
| // OAK-4533: make DELAY_THRESHOLD and MAX_DELAY adjustable - using System.properties for now |
| static { |
| final String delayThresholdStr = System.getProperty("oak.commitRateLimiter.delayThreshold"); |
| final String maxDelayStr = System.getProperty("oak.commitRateLimiter.maxDelay"); |
| double delayThreshold = 0.8; /* default is 0.8 still */ |
| int maxDelay = 10000; /* default is 10000 still */ |
| try{ |
| if (delayThresholdStr != null && delayThresholdStr.length() != 0) { |
| delayThreshold = Double.parseDouble(delayThresholdStr); |
| LOG.info("<clinit> using oak.commitRateLimiter.delayThreshold of " + delayThreshold); |
| } |
| } catch(RuntimeException e) { |
| LOG.warn("<clinit> could not parse oak.commitRateLimiter.delayThreshold, using default(" + delayThreshold + "): " + e, e); |
| } |
| try{ |
| if (maxDelayStr != null && maxDelayStr.length() != 0) { |
| maxDelay = Integer.parseInt(maxDelayStr); |
| LOG.info("<clinit> using oak.commitRateLimiter.maxDelay of " + maxDelay + "ms"); |
| } |
| } catch(RuntimeException e) { |
| LOG.warn("<clinit> could not parse oak.commitRateLimiter.maxDelay, using default(" + maxDelay + "): " + e, e); |
| } |
| DELAY_THRESHOLD = delayThreshold; |
| MAX_DELAY = maxDelay; |
| } |
| |
| private static final AtomicInteger COUNTER = new AtomicInteger(); |
| |
| /** |
| * JMX ObjectName property storing the listenerId which allows |
| * to correlate various mbeans |
| */ |
| static final String LISTENER_ID = "listenerId"; |
| |
| private final ContentSession contentSession; |
| private final NamePathMapper namePathMapper; |
| private final ListenerTracker tracker; |
| private final EventListener eventListener; |
| private final AtomicReference<FilterProvider> filterProvider; |
| private final MeterStats eventCount; |
| private final TimerStats eventDuration; |
| private final TimeSeriesMax maxQueueLengthRecorder; |
| private final int queueLength; |
| private final CommitRateLimiter commitRateLimiter; |
| private final BlobAccessProvider blobAccessProvider; |
| |
| /** |
| * Lazy initialization via the {@link #start(Whiteboard)} method |
| */ |
| private String listenerId; |
| |
| /** |
| * Lazy initialization via the {@link #start(Whiteboard)} method |
| */ |
| private CompositeRegistration registration; |
| |
| /** |
| * for statistics: tracks how many times prefiltering excluded a commit |
| */ |
| private int prefilterExcludeCount; |
| |
| /** |
| * for statistics: tracks how many times prefiltering included a commit |
| */ |
| private int prefilterIncludeCount; |
| |
| /** |
| * for statistics: tracks how many times prefiltering was ignored (not evaluated at all), |
| * either because it was disabled, queue too small, CommitInfo null or CommitContext null |
| */ |
| private int prefilterSkipCount; |
| |
| public ChangeProcessor( |
| ContentSession contentSession, |
| NamePathMapper namePathMapper, |
| ListenerTracker tracker, |
| FilterProvider filter, |
| StatisticManager statisticManager, |
| int queueLength, |
| CommitRateLimiter commitRateLimiter, |
| BlobAccessProvider blobAccessProvider) { |
| this.contentSession = contentSession; |
| this.namePathMapper = namePathMapper; |
| this.tracker = tracker; |
| eventListener = tracker.getTrackedListener(); |
| filterProvider = new AtomicReference<FilterProvider>(filter); |
| this.eventCount = statisticManager.getMeter(OBSERVATION_EVENT_COUNTER); |
| this.eventDuration = statisticManager.getTimer(OBSERVATION_EVENT_DURATION); |
| this.maxQueueLengthRecorder = statisticManager.maxQueLengthRecorder(); |
| this.queueLength = queueLength; |
| this.commitRateLimiter = commitRateLimiter; |
| this.blobAccessProvider = blobAccessProvider; |
| } |
| |
| /** |
| * Set the filter for the events this change processor will generate. |
| * @param filter |
| */ |
| public void setFilterProvider(FilterProvider filter) { |
| filterProvider.set(filter); |
| } |
| |
| /** for testing only - hence package protected **/ |
| FilterProvider getFilterProvider() { |
| return filterProvider.get(); |
| } |
| |
| @NotNull |
| public ChangeProcessorMBean getMBean() { |
| return new ChangeProcessorMBean() { |
| |
| @Override |
| public int getPrefilterExcludeCount() { |
| return prefilterExcludeCount; |
| } |
| |
| @Override |
| public int getPrefilterIncludeCount() { |
| return prefilterIncludeCount; |
| } |
| |
| @Override |
| public int getPrefilterSkipCount() { |
| return prefilterSkipCount; |
| } |
| |
| }; |
| } |
| |
| |
| /** |
| * Start this change processor |
| * @param whiteboard the whiteboard instance to used for scheduling individual |
| * runs of this change processor. |
| * @throws IllegalStateException if started already |
| */ |
| public synchronized void start(Whiteboard whiteboard) { |
| checkState(registration == null, "Change processor started already"); |
| final WhiteboardExecutor executor = new WhiteboardExecutor(); |
| executor.start(whiteboard); |
| final FilteringObserver filteringObserver = createObserver(executor); |
| listenerId = COUNTER.incrementAndGet() + ""; |
| Map<String, String> attrs = ImmutableMap.of(LISTENER_ID, listenerId); |
| String name = tracker.toString(); |
| registration = new CompositeRegistration( |
| whiteboard.register(Observer.class, filteringObserver, emptyMap()), |
| registerMBean(whiteboard, EventListenerMBean.class, |
| tracker.getListenerMBean(), "EventListener", name, attrs), |
| registerMBean(whiteboard, BackgroundObserverMBean.class, |
| filteringObserver.getBackgroundObserver().getMBean(), BackgroundObserverMBean.TYPE, name, attrs), |
| registerMBean(whiteboard, ChangeProcessorMBean.class, |
| getMBean(), ChangeProcessorMBean.TYPE, name, attrs), |
| //TODO If FilterProvider gets changed later then MBean would need to be |
| // re-registered |
| registerMBean(whiteboard, FilterConfigMBean.class, |
| filterProvider.get().getConfigMBean(), FilterConfigMBean.TYPE, name, attrs), |
| new Registration() { |
| @Override |
| public void unregister() { |
| filteringObserver.close(); |
| } |
| }, |
| new Registration() { |
| @Override |
| public void unregister() { |
| executor.stop(); |
| } |
| }, |
| scheduleWithFixedDelay(whiteboard, new Runnable() { |
| @Override |
| public void run() { |
| tracker.recordOneSecond(); |
| } |
| }, 1) |
| ); |
| } |
| |
| private FilteringObserver createObserver(final WhiteboardExecutor executor) { |
| FilteringDispatcher fd = new FilteringDispatcher(this); |
| BackgroundObserver bo = new BackgroundObserver(fd, executor, queueLength) { |
| private volatile long delay; |
| private volatile boolean blocking; |
| |
| private long lastQueueFullWarnTimestamp = -1; |
| |
| @Override |
| protected void added(int newQueueSize) { |
| queueSizeChanged(newQueueSize); |
| } |
| |
| @Override |
| protected void removed(int newQueueSize, long created) { |
| queueSizeChanged(newQueueSize); |
| } |
| |
| private void queueSizeChanged(int newQueueSize) { |
| maxQueueLengthRecorder.recordValue(newQueueSize); |
| tracker.recordQueueLength(newQueueSize); |
| if (newQueueSize >= queueLength) { |
| if (commitRateLimiter != null) { |
| if (!blocking) { |
| logQueueFullWarning("Revision queue is full. Further commits will be blocked."); |
| } |
| commitRateLimiter.blockCommits(); |
| } else if (!blocking) { |
| logQueueFullWarning("Revision queue is full. Further revisions will be compacted."); |
| } |
| blocking = true; |
| } else { |
| double fillRatio = (double) newQueueSize / queueLength; |
| if (fillRatio > DELAY_THRESHOLD) { |
| if (commitRateLimiter != null) { |
| if (delay == 0) { |
| LOG.warn("Revision queue is becoming full. Further commits will be delayed."); |
| } |
| |
| // Linear backoff proportional to the number of items exceeding |
| // DELAY_THRESHOLD. Offset by 1 to trigger the log message in the |
| // else branch once the queue falls below DELAY_THRESHOLD again. |
| int newDelay = 1 + (int) ((fillRatio - DELAY_THRESHOLD) / (1 - DELAY_THRESHOLD) * MAX_DELAY); |
| if (newDelay > delay) { |
| delay = newDelay; |
| commitRateLimiter.setDelay(delay); |
| } |
| } |
| } else { |
| if (commitRateLimiter != null) { |
| if (delay > 0) { |
| LOG.debug("Revision queue becoming empty. Unblocking commits"); |
| commitRateLimiter.setDelay(0); |
| delay = 0; |
| } |
| if (blocking) { |
| LOG.debug("Revision queue becoming empty. Stop delaying commits."); |
| commitRateLimiter.unblockCommits(); |
| blocking = false; |
| } |
| } else { |
| blocking = false; |
| } |
| } |
| } |
| } |
| |
| private void logQueueFullWarning(String message) { |
| long currTime = clock.getTime(); |
| if (lastQueueFullWarnTimestamp + QUEUE_FULL_WARN_INTERVAL < currTime) { |
| LOG.warn("{} Suppressing further such cases for {} minutes.", |
| message, |
| TimeUnit.MILLISECONDS.toMinutes(QUEUE_FULL_WARN_INTERVAL)); |
| lastQueueFullWarnTimestamp = currTime; |
| } else { |
| LOG.debug(message); |
| } |
| } |
| |
| @Override |
| public String toString() { |
| return "Prefiltering BackgroundObserver for "+ChangeProcessor.this; |
| } |
| }; |
| return new FilteringObserver(bo, new Filter() { |
| |
| @Override |
| public boolean excludes(NodeState root, CommitInfo info) { |
| final FilterResult filterResult = evalPrefilter(root, info, getChangeSet(info)); |
| switch (filterResult) { |
| case PREFILTERING_SKIPPED: { |
| prefilterSkipCount++; |
| return false; |
| } |
| case EXCLUDE: { |
| prefilterExcludeCount++; |
| return true; |
| } |
| case INCLUDE: { |
| prefilterIncludeCount++; |
| return false; |
| } |
| default: { |
| LOG.info("isExcluded: unknown/unsupported filter result: " + filterResult); |
| prefilterSkipCount++; |
| return false; |
| } |
| } |
| } |
| }); |
| } |
| |
| private final Monitor runningMonitor = new Monitor(); |
| private final RunningGuard running = new RunningGuard(runningMonitor); |
| |
| /** |
| * Try to stop this change processor if running. This method will wait |
| * the specified time for a pending event listener to complete. If |
| * no timeout occurred no further events will be delivered after this |
| * method returns. |
| * <p> |
| * Does nothing if stopped already. |
| * |
| * @param timeOut time this method will wait for an executing event |
| * listener to complete. |
| * @param unit time unit for {@code timeOut} |
| * @return {@code true} if no time out occurred and this change processor |
| * could be stopped, {@code false} otherwise. |
| * @throws IllegalStateException if not yet started |
| */ |
| public synchronized boolean stopAndWait(int timeOut, TimeUnit unit) { |
| checkState(registration != null, "Change processor not started"); |
| if (running.stop()) { |
| if (runningMonitor.enter(timeOut, unit)) { |
| registration.unregister(); |
| runningMonitor.leave(); |
| return true; |
| } else { |
| // Timed out |
| return false; |
| } |
| } else { |
| // Stopped already |
| return true; |
| } |
| } |
| |
| /** |
| * Stop this change processor after all pending events have been |
| * delivered. In contrast to {@link #stopAndWait(int, java.util.concurrent.TimeUnit)} |
| * this method returns immediately without waiting for pending listeners to |
| * complete. |
| */ |
| public synchronized void stop() { |
| checkState(registration != null, "Change processor not started"); |
| if (running.stop()) { |
| registration.unregister(); |
| runningMonitor.leave(); |
| } |
| } |
| |
| /** |
| * Utility method that extracts the ChangeSet from a CommitInfo if possible. |
| * @param info |
| * @return |
| */ |
| public static ChangeSet getChangeSet(CommitInfo info) { |
| if (info == null) { |
| return null; |
| } |
| CommitContext context = (CommitContext) info.getInfo().get(CommitContext.NAME); |
| if (context == null) { |
| return null; |
| } |
| return (ChangeSet) context.get(COMMIT_CONTEXT_OBSERVATION_CHANGESET); |
| } |
| |
| @Override |
| public void contentChanged(@NotNull NodeState before, |
| @NotNull NodeState after, |
| @NotNull CommitInfo info) { |
| checkNotNull(before); // OAK-5160 before is now guaranteed to be non-null |
| checkNotNull(after); |
| checkNotNull(info); |
| try { |
| long start = PERF_LOGGER.start(); |
| FilterProvider provider = filterProvider.get(); |
| // FIXME don't rely on toString for session id |
| if (provider.includeCommit(contentSession.toString(), info)) { |
| EventFilter filter = provider.getFilter(before, after); |
| EventIterator events = new EventQueue(namePathMapper, |
| blobAccessProvider, info, before, after, |
| provider.getSubTrees(), Filters.all(filter, VISIBLE_FILTER), |
| provider.getEventAggregator()); |
| |
| long time = System.nanoTime(); |
| boolean hasEvents = events.hasNext(); |
| tracker.recordProducerTime(System.nanoTime() - time, TimeUnit.NANOSECONDS); |
| if (hasEvents && runningMonitor.enterIf(running)) { |
| if (commitRateLimiter != null) { |
| commitRateLimiter.beforeNonBlocking(); |
| } |
| try { |
| CountingIterator countingEvents = new CountingIterator(events); |
| eventListener.onEvent(countingEvents); |
| countingEvents.updateCounters(eventCount, eventDuration); |
| } finally { |
| if (commitRateLimiter != null) { |
| commitRateLimiter.afterNonBlocking(); |
| } |
| runningMonitor.leave(); |
| } |
| } |
| } |
| PERF_LOGGER.end(start, 100, |
| "Generated events (before: {}, after: {})", |
| before, after); |
| } catch (Exception e) { |
| LOG.warn("Error while dispatching observation events for " + tracker, e); |
| } |
| } |
| |
| private static class CountingIterator implements EventIterator { |
| private final long t0 = System.nanoTime(); |
| private final EventIterator events; |
| private long eventCount; |
| private long sysTime; |
| |
| public CountingIterator(EventIterator events) { |
| this.events = events; |
| } |
| |
| public void updateCounters(MeterStats eventCount, TimerStats eventDuration) { |
| checkState(this.eventCount >= 0); |
| eventCount.mark(this.eventCount); |
| eventDuration.update(System.nanoTime() - t0 - sysTime, TimeUnit.NANOSECONDS); |
| this.eventCount = -1; |
| } |
| |
| @Override |
| public Event next() { |
| if (eventCount == -1) { |
| LOG.warn("Access to EventIterator outside the onEvent callback detected. This will " + |
| "cause observation related values in RepositoryStatistics to become unreliable."); |
| eventCount = -2; |
| } |
| |
| long t0 = System.nanoTime(); |
| try { |
| return events.nextEvent(); |
| } finally { |
| eventCount++; |
| sysTime += System.nanoTime() - t0; |
| } |
| } |
| |
| @Override |
| public boolean hasNext() { |
| long t0 = System.nanoTime(); |
| try { |
| return events.hasNext(); |
| } finally { |
| sysTime += System.nanoTime() - t0; |
| } |
| } |
| |
| @Override |
| public Event nextEvent() { |
| return next(); |
| } |
| |
| @Override |
| public void skip(long skipNum) { |
| long t0 = System.nanoTime(); |
| try { |
| events.skip(skipNum); |
| } finally { |
| sysTime += System.nanoTime() - t0; |
| } |
| } |
| |
| @Override |
| public long getSize() { |
| return events.getSize(); |
| } |
| |
| @Override |
| public long getPosition() { |
| return events.getPosition(); |
| } |
| |
| @Override |
| public void remove() { |
| throw new UnsupportedOperationException(); |
| } |
| } |
| |
| private static class RunningGuard extends Guard { |
| private boolean stopped; |
| |
| public RunningGuard(Monitor monitor) { |
| super(monitor); |
| } |
| |
| @Override |
| public boolean isSatisfied() { |
| return !stopped; |
| } |
| |
| /** |
| * @return {@code true} if this call set this guard to stopped, |
| * {@code false} if another call set this guard to stopped before. |
| */ |
| public boolean stop() { |
| boolean wasStopped = stopped; |
| stopped = true; |
| return !wasStopped; |
| } |
| } |
| |
| @Override |
| public String toString() { |
| return "ChangeProcessor [" |
| + "listenerId=" + listenerId |
| + ", tracker=" + tracker |
| + ", contentSession=" + contentSession |
| + ", eventCount=" + eventCount |
| + ", eventDuration=" + eventDuration |
| + ", commitRateLimiter=" + commitRateLimiter |
| + ", running=" + running.isSatisfied() + "]"; |
| } |
| |
| /** for logging only **/ |
| String getListenerToString() { |
| if (tracker == null) { |
| return "null"; |
| } |
| EventListenerMBean listenerMBean = tracker.getListenerMBean(); |
| if (listenerMBean == null) { |
| return "null (no listener mbean)"; |
| } |
| return listenerMBean.getToString(); |
| } |
| |
| /** |
| * Evaluate the prefilter for a given commit. |
| * @param changeSet |
| * |
| * @return a FilterResult indicating either inclusion, exclusion or |
| * inclusion-due-to-skipping. The latter is used to reflect |
| * prefilter evaluation better in statistics (as it could also have |
| * been reported just as include) |
| */ |
| private FilterResult evalPrefilter(NodeState root, CommitInfo info, ChangeSet changeSet) { |
| if (info == null) { |
| return FilterResult.PREFILTERING_SKIPPED; |
| } |
| if (root == null) { |
| // likely only occurs at startup |
| // we can't do any diffing etc, so just not exclude it |
| return FilterResult.PREFILTERING_SKIPPED; |
| } |
| |
| final FilterProvider fp = filterProvider.get(); |
| // FIXME don't rely on toString for session id |
| if (!fp.includeCommit(contentSession.toString(), info)) { |
| // 'classic' (and cheap pre-) filtering |
| return FilterResult.EXCLUDE; |
| } |
| if (changeSet == null) { |
| // then can't do any prefiltering since it was not |
| // able to complete the sets (within the given boundaries) |
| // (this corresponds to a large commit, which thus can't |
| // go through prefiltering) |
| return FilterResult.PREFILTERING_SKIPPED; |
| } |
| |
| final ChangeSetFilter prefilter = fp; |
| if (prefilter.excludes(changeSet)) { |
| return FilterResult.EXCLUDE; |
| } else { |
| return FilterResult.INCLUDE; |
| } |
| } |
| } |
