LayerManager.java
/*
* 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.commons.collections4.bloomfilter;
import java.util.Deque;
import java.util.LinkedList;
import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.function.Supplier;
/**
* Implementation of the methods to manage the layers in a layered Bloom filter.
* <p>
* The manager comprises a list of Bloom filters that are managed based on
* various rules. The last filter in the list is known as the {@code target} and
* is the filter into which merges are performed. The Layered manager utilizes
* three methods to manage the list.
* </p>
* <ul>
* <li>ExtendCheck - A Predicate that if true causes a new Bloom filter to be
* created as the new target.</li>
* <li>FilterSupplier - A Supplier that produces empty Bloom filters to be used
* as a new target.</li>
* <li>Cleanup - A Consumer of a {@code LinkedList} of BloomFilter that removes any
* expired or out dated filters from the list.</li>
* </ul>
* <p>
* When extendCheck returns {@code true} the following steps are taken:
* </p>
* <ol>
* <li>{@code Cleanup} is called</li>
* <li>{@code FilterSuplier} is executed and the new filter added to the list as
* the {@code target} filter.</li>
* </ol>
*
* @since 4.5
*/
public class LayerManager<T extends BloomFilter> implements BloomFilterExtractor {
/**
* Builder to create Layer Manager
*/
public static class Builder<T extends BloomFilter> {
private Predicate<LayerManager<T>> extendCheck;
private Supplier<T> supplier;
private Consumer<Deque<T>> cleanup;
private Builder() {
extendCheck = ExtendCheck.neverAdvance();
cleanup = Cleanup.noCleanup();
}
/**
* Builds the layer manager with the specified properties.
*
* @return a new LayerManager.
*/
public LayerManager<T> build() {
Objects.requireNonNull(supplier, "Supplier must not be null");
Objects.requireNonNull(extendCheck, "ExtendCheck must not be null");
Objects.requireNonNull(cleanup, "Cleanup must not be null");
return new LayerManager<>(supplier, extendCheck, cleanup, true);
}
/**
* Sets the Consumer that cleans the list of Bloom filters.
*
* @param cleanup the Consumer that will modify the list of filters removing out
* dated or stale filters.
* @return {@code this} instance.
*/
public Builder<T> setCleanup(final Consumer<Deque<T>> cleanup) {
this.cleanup = cleanup;
return this;
}
/**
* Sets the extendCheck predicate. When the predicate returns {@code true} a new
* target will be created.
*
* @param extendCheck The predicate to determine if a new target should be
* created.
* @return this for chaining.
*/
public Builder<T> setExtendCheck(final Predicate<LayerManager<T>> extendCheck) {
this.extendCheck = extendCheck;
return this;
}
/**
* Sets the supplier of Bloom filters. When extendCheck creates a new target,
* the supplier provides the instance of the Bloom filter.
*
* @param supplier The supplier of new Bloom filter instances.
* @return this for chaining.
*/
public Builder<T> setSupplier(final Supplier<T> supplier) {
this.supplier = supplier;
return this;
}
}
/**
* Static methods to create a Consumer of a List of BloomFilter perform
* tests on whether to reduce the collection of Bloom filters.
*/
public static final class Cleanup {
/**
* A Cleanup that never removes anything.
*
* @param <T> Type of BloomFilter.
* @return A Consumer suitable for the LayerManager {@code cleanup} parameter.
*/
public static <T extends BloomFilter> Consumer<Deque<T>> noCleanup() {
return x -> {
// empty
};
}
/**
* Removes the earliest filters in the list when the the number of filters
* exceeds maxSize.
*
* @param <T> Type of BloomFilter.
* @param maxSize the maximum number of filters for the list. Must be greater
* than 0
* @return A Consumer suitable for the LayerManager {@code cleanup} parameter.
* @throws IllegalArgumentException if {@code maxSize <= 0}.
*/
public static <T extends BloomFilter> Consumer<Deque<T>> onMaxSize(final int maxSize) {
if (maxSize <= 0) {
throw new IllegalArgumentException("'maxSize' must be greater than 0");
}
return ll -> {
while (ll.size() > maxSize) {
ll.removeFirst();
}
};
}
/**
* Removes the last added target if it is empty. Useful as the first in a chain
* of cleanup consumers. (e.g. {@code Cleanup.removeEmptyTarget.andThen( otherConsumer )})
*
* @param <T> Type of BloomFilter.
* @return A Consumer suitable for the LayerManager {@code cleanup} parameter.
*/
public static <T extends BloomFilter> Consumer<Deque<T>> removeEmptyTarget() {
return x -> {
if (!x.isEmpty() && x.getLast().isEmpty()) {
x.removeLast();
}
};
}
/**
* Removes any layer identified by the predicate.
*
* @param <T> Type of BloomFilter.
* @param test Predicate.
* @return A Consumer suitable for the LayerManager {@code cleanup} parameter.
*/
public static <T extends BloomFilter> Consumer<Deque<T>> removeIf(final Predicate<? super T> test) {
return x -> x.removeIf(test);
}
private Cleanup() {
}
}
/**
* A collection of common ExtendCheck implementations to test whether to extend
* the depth of a LayerManager.
*/
public static final class ExtendCheck {
/**
* Creates a new target after a specific number of filters have been added to
* the current target.
*
* @param <T> Type of BloomFilter.
* @param breakAt the number of filters to merge into each filter in the list.
* @return A Predicate suitable for the LayerManager {@code extendCheck} parameter.
* @throws IllegalArgumentException if {@code breakAt <= 0}
*/
public static <T extends BloomFilter> Predicate<LayerManager<T>> advanceOnCount(final int breakAt) {
if (breakAt <= 0) {
throw new IllegalArgumentException("'breakAt' must be greater than 0");
}
return new Predicate<LayerManager<T>>() {
int count;
@Override
public boolean test(final LayerManager<T> filter) {
if (++count == breakAt) {
count = 0;
return true;
}
return false;
}
};
}
/**
* Advances the target once a merge has been performed.
*
* @param <T> Type of BloomFilter.
* @return A Predicate suitable for the LayerManager {@code extendCheck} parameter.
*/
public static <T extends BloomFilter> Predicate<LayerManager<T>> advanceOnPopulated() {
return lm -> !lm.last().isEmpty();
}
/**
* Creates a new target after the current target is saturated. Saturation is
* defined as the {@code Bloom filter estimated N >= maxN}.
*
* <p>An example usage is advancing on a calculated saturation by calling:
* {@code ExtendCheck.advanceOnSaturation(shape.estimateMaxN()) }</p>
*
* @param <T> Type of BloomFilter.
* @param maxN the maximum number of estimated items in the filter.
* @return A Predicate suitable for the LayerManager {@code extendCheck} parameter.
* @throws IllegalArgumentException if {@code maxN <= 0}
*/
public static <T extends BloomFilter> Predicate<LayerManager<T>> advanceOnSaturation(final double maxN) {
if (maxN <= 0) {
throw new IllegalArgumentException("'maxN' must be greater than 0");
}
return manager -> {
final BloomFilter bf = manager.last();
return maxN <= bf.getShape().estimateN(bf.cardinality());
};
}
/**
* Does not automatically advance the target. @{code next()} must be called directly to
* perform the advance.
*
* @param <T> Type of BloomFilter.
* @return A Predicate suitable for the LayerManager {@code extendCheck} parameter.
*/
public static <T extends BloomFilter> Predicate<LayerManager<T>> neverAdvance() {
return x -> false;
}
private ExtendCheck() {
}
}
/**
* Creates a new Builder with defaults of {@code ExtendCheck.neverAdvance()} and
* {@code Cleanup.noCleanup()}.
*
* @param <T> Type of BloomFilter.
* @return A builder.
* @see ExtendCheck#neverAdvance()
* @see Cleanup#noCleanup()
*/
public static <T extends BloomFilter> Builder<T> builder() {
return new Builder<>();
}
private final LinkedList<T> filters = new LinkedList<>();
private final Consumer<Deque<T>> filterCleanup;
private final Predicate<LayerManager<T>> extendCheck;
private final Supplier<T> filterSupplier;
/**
* Constructor.
*
* @param filterSupplier the supplier of new Bloom filters to add the the list
* when necessary.
* @param extendCheck The predicate that checks if a new filter should be
* added to the list.
* @param filterCleanup the consumer that removes any old filters from the
* list.
* @param initialize true if the filter list should be initialized.
*/
private LayerManager(final Supplier<T> filterSupplier, final Predicate<LayerManager<T>> extendCheck,
final Consumer<Deque<T>> filterCleanup, final boolean initialize) {
this.filterSupplier = filterSupplier;
this.extendCheck = extendCheck;
this.filterCleanup = filterCleanup;
if (initialize) {
addFilter();
}
}
/**
* Adds a new Bloom filter to the list.
*/
private void addFilter() {
final T bf = filterSupplier.get();
if (bf == null) {
throw new NullPointerException("filterSupplier returned null.");
}
filters.add(bf);
}
/**
* Forces execution the configured cleanup without creating a new filter except in cases
* where the cleanup removes all the layers.
* @see LayerManager.Builder#setCleanup(Consumer)
*/
void cleanup() {
this.filterCleanup.accept(filters);
if (filters.isEmpty()) {
addFilter();
}
}
/**
* Removes all the filters from the layer manager, and sets up a new one as the
* target.
*/
public final void clear() {
filters.clear();
addFilter();
}
/**
* Creates a deep copy of this LayerManager.
* <p><em>Filters in the copy are deep copies, not references, so changes in the copy
* are NOT reflected in the original.</em></p>
* <p>The {@code filterSupplier}, {@code extendCheck}, and the {@code filterCleanup} are shared between
* the copy and this instance.</p>
*
* @return a copy of this layer Manager.
*/
public LayerManager<T> copy() {
final LayerManager<T> newMgr = new LayerManager<>(filterSupplier, extendCheck, filterCleanup, false);
for (final T bf : filters) {
newMgr.filters.add(bf.copy());
}
return newMgr;
}
/**
* Gets the Bloom filter from the first layer.
* No extension check is performed during this call.
* @return The Bloom filter from the first layer.
* @see #getTarget()
*/
public final T first() {
return filters.getFirst();
}
/**
* Executes a Bloom filter Predicate on each Bloom filter in the manager in
* depth order. Oldest filter first.
*
* @param bloomFilterPredicate the predicate to evaluate each Bloom filter with.
* @return {@code false} when the a filter fails the predicate test. Returns
* {@code true} if all filters pass the test.
*/
@Override
public boolean processBloomFilters(final Predicate<BloomFilter> bloomFilterPredicate) {
for (final BloomFilter bf : filters) {
if (!bloomFilterPredicate.test(bf)) {
return false;
}
}
return true;
}
/**
* Gets the Bloom filter at the specified depth. The filter at depth 0 is the
* oldest filter.
*
* @param depth the depth at which the desired filter is to be found.
* @return the filter.
* @throws NoSuchElementException if depth is not in the range
* [0,filters.size())
*/
public final T get(final int depth) {
if (depth < 0 || depth >= filters.size()) {
throw new NoSuchElementException(String.format("Depth must be in the range [0,%s)", filters.size()));
}
return filters.get(depth);
}
/**
* Returns the number of filters in the LayerManager. In the default LayerManager implementation
* there is always at least one layer.
*
* @return the current depth.
*/
public final int getDepth() {
return filters.size();
}
/**
* Returns the current target filter. If a new filter should be created based on
* {@code extendCheck} it will be created before this method returns.
*
* @return the current target filter after any extension.
*/
public final T getTarget() {
if (extendCheck.test(this)) {
next();
}
return last();
}
/**
* Gets the Bloom filter from the last layer.
* No extension check is performed during this call.
* @return The Bloom filter from the last layer.
* @see #getTarget()
*/
public final T last() {
return filters.getLast();
}
/**
* Forces an advance to the next depth. This method will clean-up the current
* layers and generate a new filter layer. In most cases is it unnecessary to
* call this method directly.
* <p>
* Ths method is used within {@link #getTarget()} when the configured
* {@code ExtendCheck} returns {@code true}.
* </p>
* @see LayerManager.Builder#setExtendCheck(Predicate)
* @see LayerManager.Builder#setCleanup(Consumer)
*/
void next() {
this.filterCleanup.accept(filters);
addFilter();
}
}