/*
    * 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>
   *
   *
   * @param <T> the {@link BloomFilter} type.
   * @since 4.5.0-M1
   */
  public class LayerManager<T extends BloomFilter<T>> implements BloomFilterExtractor {
  
      /**
       * Builds new instances of {@link LayerManager}.
       *
       * @param <T> the {@link BloomFilter} type.
       */
      public static class Builder<T extends BloomFilter<T>> implements Supplier<LayerManager<T>> {
  
          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.
           */
          @Override
          public LayerManager<T> get() {
              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 {@code this} instance.
          */
         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 {@code this} instance.
          */
         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<T>> 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<T>> 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.  (for example {@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<T>> 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<T>> 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<T>> 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<T>> 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<T>> Predicate<LayerManager<T>> advanceOnSaturation(final double maxN) {
             if (maxN <= 0) {
                 throw new IllegalArgumentException("'maxN' must be greater than 0");
             }
             return manager -> {
                 final T 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<T>> Predicate<LayerManager<T>> neverAdvance() {
             return x -> false;
         }
 
         private ExtendCheck() {
         }
     }
 
     /**
      * Creates a new Builder with defaults of {@link ExtendCheck#neverAdvance()} and
      * {@link Cleanup#noCleanup()}.
      *
      * @param <T> Type of BloomFilter.
      * @return A builder.
      * @see ExtendCheck#neverAdvance()
      * @see Cleanup#noCleanup()
      */
     public static <T extends BloomFilter<T>> 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;
 
     /**
      * Constructs a new instance.
      *
      * @param filterSupplier the non-null supplier of new Bloom filters to add the the list
      *                       when necessary.
      * @param extendCheck    The non-null predicate that checks if a new filter should be
      *                       added to the list.
      * @param filterCleanup  the non-null 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 = Objects.requireNonNull(filterSupplier, "filterSupplier");
         this.extendCheck = Objects.requireNonNull(extendCheck, "extendCheck");
         this.filterCleanup = Objects.requireNonNull(filterCleanup, "filterCleanup");
         if (initialize) {
             addFilter();
         }
     }
 
     /**
      * Adds a new Bloom filter to the list.
      */
     private void addFilter() {
         filters.add(Objects.requireNonNull(filterSupplier.get(), "filterSupplier.get() returned null."));
     }
 
     /**
      * 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() {
         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 {@link 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 {@link LayerManager}.
      */
     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();
     }
 
     /**
      * 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);
     }
 
     /**
      * Gets 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();
     }
 
     /**
      * Gets 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() {
         filterCleanup.accept(filters);
         addFilter();
     }
 
     /**
      * 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) {
         return filters.stream().allMatch(bloomFilterPredicate);
     }
 }