/*
    * 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.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 implements BloomFilterProducer {
  
      /**
       * Builder to create Layer Manager
       */
      public static class Builder {
          private Predicate<LayerManager> extendCheck;
          private Supplier<BloomFilter> supplier;
          private Consumer<LinkedList<BloomFilter>> cleanup;
  
          private Builder() {
              extendCheck = ExtendCheck.neverAdvance();
              cleanup = Cleanup.noCleanup();
          }
  
          /**
           * Builds the layer manager with the specified properties.
           *
           * @return a new LayerManager.
           */
          public LayerManager 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 this for chaining.
           */
          public Builder setCleanup(Consumer<LinkedList<BloomFilter>> 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 setExtendCheck(Predicate<LayerManager> 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 setSupplier(Supplier<BloomFilter> supplier) {
             this.supplier = supplier;
             return this;
         }
     }
 
     /**
      * Static methods to create a Consumer of a LinkedList of BloomFilter perform
      * tests on whether to reduce the collection of Bloom filters.
      */
     public static final class Cleanup {
         /**
          * A Cleanup that never removes anything.
          * @return A Consumer suitable for the LayerManager {@code cleanup} parameter.
          */
         public static Consumer<LinkedList<BloomFilter>> noCleanup() {
             return x -> {};
         }
 
         /**
          * Removes the earliest filters in the list when the the number of filters
          * exceeds maxSize.
          *
          * @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 Consumer<LinkedList<BloomFilter>> onMaxSize(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 )})
          *
          * @return A Consumer suitable for the LayerManager {@code cleanup} parameter.
          */
         public static Consumer<LinkedList<BloomFilter>> removeEmptyTarget() {
             return x -> {
                 if (x.getLast().cardinality() == 0) {
                     x.removeLast();
                 }
             };
         }
 
         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 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 Predicate<LayerManager> advanceOnCount(int breakAt) {
             if (breakAt <= 0) {
                 throw new IllegalArgumentException("'breakAt' must be greater than 0");
             }
             return new Predicate<LayerManager>() {
                 int count;
 
                 @Override
                 public boolean test(LayerManager filter) {
                     return ++count % breakAt == 0;
                 }
             };
         }
 
         /**
          * Advances the target once a merge has been performed.
          * @return A Predicate suitable for the LayerManager {@code extendCheck} parameter.
          */
         public static Predicate<LayerManager> advanceOnPopulated() {
             return lm -> !lm.filters.peekLast().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 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 Predicate<LayerManager> advanceOnSaturation(double maxN) {
             if (maxN <= 0) {
                 throw new IllegalArgumentException("'maxN' must be greater than 0");
             }
             return manager -> {
                 BloomFilter bf = manager.filters.peekLast();
                 return maxN <= bf.getShape().estimateN(bf.cardinality());
             };
         }
 
         /**
          * Does not automatically advance the target. @{code next()} must be called directly to
          * perform the advance.
          * @return A Predicate suitable for the LayerManager {@code extendCheck} parameter.
          */
         public static Predicate<LayerManager> neverAdvance() {
             return x -> false;
         }
 
         private ExtendCheck() {
         }
     }
     /**
      * Creates a new Builder with defaults of {@code ExtendCheck.neverAdvance()} and
      * {@code Cleanup.noCleanup()}.
      *
      * @return A builder.
      * @see ExtendCheck#neverAdvance()
      * @see Cleanup#noCleanup()
      */
     public static Builder builder() {
         return new Builder();
     }
     private final LinkedList<BloomFilter> filters = new LinkedList<>();
     private final Consumer<LinkedList<BloomFilter>> filterCleanup;
 
     private final Predicate<LayerManager> extendCheck;
 
     private final Supplier<BloomFilter> 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(Supplier<BloomFilter> filterSupplier, Predicate<LayerManager> extendCheck,
             Consumer<LinkedList<BloomFilter>> filterCleanup, 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() {
         BloomFilter bf = filterSupplier.get();
         if (bf == null) {
             throw new NullPointerException("filterSupplier returned null.");
         }
         filters.add(bf);
     }
 
     /**
      * 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 copy() {
         LayerManager newMgr = new LayerManager(filterSupplier, extendCheck, filterCleanup, false);
         for (BloomFilter bf : filters) {
             newMgr.filters.add(bf.copy());
         }
         return newMgr;
     }
 
     /**
      * 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 forEachBloomFilter(Predicate<BloomFilter> bloomFilterPredicate) {
         for (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 BloomFilter get(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 alwasy 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 BloomFilter getTarget() {
         if (extendCheck.test(this)) {
             next();
         }
         return filters.peekLast();
     }
 
     /**
      * 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>
      */
     void next() {
         this.filterCleanup.accept(filters);
         addFilter();
     }
 }