/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.collections;

import com.google.common.base.Preconditions;
import com.google.common.collect.Iterables;

import java.util.AbstractSet;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import javax.annotation.concurrent.ThreadSafe;

/**
 * A set of objects that are indexed and thus can be queried by specific fields of the object.
 * Different {@link IndexedSet} instances may specify different fields to index. The field type
 * must be comparable. The field value must not be changed after an object is added to the set,
 * otherwise, behavior for all operations is not specified.
 *
 * If concurrent adds or removes for objects which are equivalent, but not the same exact object,
 * the behavior is undefined. Therefore, do not add or remove "clones" objects in the
 * {@link IndexedSet}.
 *
 * <p>
 * Example usage:
 *
 * We have a set of puppies:
 *
 * <pre>
 *   class Puppy {
 *     private final String mName;
 *     private final long mId;
 *
 *     public Puppy(String name, long id) {
 *       mName = name;
 *       mId = id;
 *     }
 *
 *     public String name() {
 *       return mName;
 *     }
 *
 *     public long id() {
 *       return mId;
 *     }
 *   }
 * </pre>
 *
 * We want to be able to retrieve the set of puppies via a puppy's id or name, one way is to have
 * two maps like {@code Map<String, Puppy> nameToPuppy} and {@code Map<Long, Puppy> idToPuppy},
 * another way is to use a single instance of {@link IndexedSet}!
 *
 * First, define the fields to be indexed:
 * <pre>
 *  IndexDefinition&lt;Puppy&gt; idIndex = new IndexDefinition&lt;Puppy&gt;(true) {
 *    {@literal @Override}
 *    Object getFieldValue(Puppy o) {
 *      return o.id();
 *    }
 *  }
 *
 *  IndexDefinition&lt;Puppy&gt; nameIndex = new IndexDefinition&lt;Puppy&gt;(true) {
 *    {@literal @Override}
 *    Object getFieldValue(Puppy o) {
 *      return o.name();
 *    }
 *  }
 * </pre>
 *
 * Then create an {@link IndexedSet} and add puppies:
 * <pre>
 *  IndexedSet&lt;Puppy&gt; puppies = new IndexedSet&lt;Puppy&gt;(idIndex, nameIndex);
 *  puppies.add(new Puppy("sweet", 0));
 *  puppies.add(new Puppy("heart", 1));
 * </pre>
 *
 * Then retrieve the puppy named sweet:
 * <pre>
 *   Puppy sweet = puppies.getFirstByField(nameIndex, "sweet");
 * </pre>
 * and retrieve the puppy with id 1:
 * <pre>
 *   Puppy heart = puppies.getFirstByField(idIndex, 1L);
 * </pre>
 *
 * @param <T> the type of object
 */
@ThreadSafe
public class IndexedSet<T> extends AbstractSet<T> {
  /**
   * The first index of the indexed set. This index is used to guarantee uniqueness of objects,
   * support iterator and provide quick lookup.
   */
  private final FieldIndex<T, ?> mPrimaryIndex;
  /**
   * Map from index definition to the index. An index is a map from index value to one or a set of
   * objects with that index value.
   */
  private final Map<IndexDefinition<T, ?>, FieldIndex<T, ?>> mIndices;

  /**
   * Constructs a new {@link IndexedSet} instance with at least one field as the index.
   *
   * @param primaryIndexDefinition at least one field is needed to index the set of objects. This
   *        primaryIndexDefinition is used to initialize {@link IndexedSet#mPrimaryIndex} and is
   *        recommended to be unique in consideration of performance.
   * @param otherIndexDefinitions other index definitions to index the set
   */
  @SafeVarargs
  public IndexedSet(IndexDefinition<T, ?> primaryIndexDefinition,
      IndexDefinition<T, ?>... otherIndexDefinitions) {
    Iterable<IndexDefinition<T, ?>> indexDefinitions =
        Iterables.concat(Collections.singletonList(primaryIndexDefinition),
            Arrays.asList(otherIndexDefinitions));

    // initialization
    Map<IndexDefinition<T, ?>, FieldIndex<T, ?>> indices = new HashMap<>();

    for (IndexDefinition<T, ?> indexDefinition : indexDefinitions) {
      FieldIndex<T, ?> index = indexDefinition.isUnique()
          ? new UniqueFieldIndex<>(indexDefinition) : new NonUniqueFieldIndex<>(indexDefinition);

      indices.put(indexDefinition, index);
    }

    mPrimaryIndex = indices.get(primaryIndexDefinition);
    mIndices = Collections.unmodifiableMap(indices);
  }

  /**
   * Removes all the entries in this set.
   *
   * This is an expensive operation, and concurrent adds are permitted.
   */
  @Override
  public void clear() {
    for (T obj : mPrimaryIndex) {
      remove(obj);
    }
  }

  /**
   * Adds an object o to the set if there is no other object o2 such that
   * {@code (o == null ? o2 == null : o.equals(o2))}. If this set already contains the object, the
   * call leaves the set unchanged.
   *
   * @param object the object to add
   * @return true if this set did not already contain the specified element
   */
  @Override
  public boolean add(T object) {
    Preconditions.checkNotNull(object, "object");

    // Locking this object protects against removing the exact object, but does not protect against
    // removing a distinct, but equivalent object.
    synchronized (object) {
      // add() will atomically add the object to the index, if it doesn't exist.
      if (!mPrimaryIndex.add(object)) {
        // This object is already added, possibly by another concurrent thread.
        return false;
      }

      for (FieldIndex<T, ?> fieldIndex : mIndices.values()) {
        fieldIndex.add(object);
      }
    }
    return true;
  }

  /**
   * Returns an iterator over the elements in this set. The elements are returned in no particular
   * order. It is to implement {@link Iterable} so that users can foreach the {@link IndexedSet}
   * directly.
   *
   * Note that the behavior of the iterator is unspecified if the underlying collection is
   * modified while a thread is going through the iterator.
   *
   * @return an iterator over the elements in this {@link IndexedSet}
   */
  @Override
  public Iterator<T> iterator() {
    return new IndexedSetIterator();
  }

  /**
   * Specialized iterator for {@link IndexedSet}.
   *
   * This is needed to support consistent removal from the set and the indices.
   */
  private class IndexedSetIterator implements Iterator<T> {
    private final Iterator<T> mSetIterator;
    private T mObject;

    public IndexedSetIterator() {
      mSetIterator = mPrimaryIndex.iterator();
      mObject = null;
    }

    @Override
    public boolean hasNext() {
      return mSetIterator.hasNext();
    }

    @Override
    public T next() {
      final T next = mSetIterator.next();
      mObject = next;
      return next;
    }

    @Override
    public void remove() {
      if (mObject != null) {
        IndexedSet.this.remove(mObject);
        mObject = null;
      } else {
        throw new IllegalStateException("next() was not called before remove()");
      }
    }
  }

  /**
   * Whether there is an object with the specified index field value in the set.
   *
   * @param indexDefinition the field index definition
   * @param value the field value
   * @param <V> the field type
   * @return true if there is one such object, otherwise false
   */
  public <V> boolean contains(IndexDefinition<T, V> indexDefinition, V value) {
    FieldIndex<T, V> index = (FieldIndex<T, V>) mIndices.get(indexDefinition);
    if (index == null) {
      throw new IllegalStateException("the given index isn't defined for this IndexedSet");
    }
    return index.containsField(value);
  }

  /**
   * Gets a subset of objects with the specified field value. If there is no object with the
   * specified field value, a newly created empty set is returned.
   *
   * @param indexDefinition the field index definition
   * @param value the field value to be satisfied
   * @param <V> the field type
   * @return the set of objects or an empty set if no such object exists
   */
  public <V> Set<T> getByField(IndexDefinition<T, V> indexDefinition, V value) {
    FieldIndex<T, V> index = (FieldIndex<T, V>) mIndices.get(indexDefinition);
    if (index == null) {
      throw new IllegalStateException("the given index isn't defined for this IndexedSet");
    }
    return index.getByField(value);
  }

  /**
   * Gets the object from the set of objects with the specified field value.
   *
   * @param indexDefinition the field index definition
   * @param value the field value
   * @param <V> the field type
   * @return the object or null if there is no such object
   */
  public <V> T getFirstByField(IndexDefinition<T, V> indexDefinition, V value) {
    FieldIndex<T, V> index = (FieldIndex<T, V>) mIndices.get(indexDefinition);
    if (index == null) {
      throw new IllegalStateException("the given index isn't defined for this IndexedSet");
    }
    return index.getFirst(value);
  }

  /**
   * Removes an object from the set.
   *
   * @param object the object to remove
   * @return true if the object is in the set and removed successfully, otherwise false
   */
  @Override
  public boolean remove(Object object) {
    if (object == null) {
      return false;
    }
    // Locking this object protects against removing the exact object that might be in the
    // process of being added, but does not protect against removing a distinct, but equivalent
    // object.
    synchronized (object) {
      if (mPrimaryIndex.containsObject((T) object)) {
        // This isn't technically typesafe. However, given that success is true, it's very unlikely
        // that the object passed to remove is not of type <T>.
        @SuppressWarnings("unchecked")
        T tObj = (T) object;
        removeFromIndices(tObj);
        return true;
      } else {
        return false;
      }
    }
  }

  /**
   * Helper method that removes an object from the indices.
   *
   * @param object the object to be removed
   */
  private void removeFromIndices(T object) {
    for (FieldIndex<T, ?> fieldValue : mIndices.values()) {
      fieldValue.remove(object);
    }
  }

  /**
   * Removes the subset of objects with the specified index field value.
   *
   * @param indexDefinition the field index
   * @param value the field value
   * @param <V> the field type
   * @return the number of objects removed
   */
  public <V> int removeByField(IndexDefinition<T, V> indexDefinition, V value) {
    Set<T> toRemove = getByField(indexDefinition, value);

    int removed = 0;
    for (T o : toRemove) {
      if (remove(o)) {
        removed++;
      }
    }
    return removed;
  }

  /**
   * @return the number of objects in this indexed set
   */
  @Override
  public int size() {
    return mPrimaryIndex.size();
  }
}
