package org.infinispan.distribution;

import org.infinispan.remoting.transport.Address;
import java.util.ArrayList;
import java.util.Arrays;
import org.infinispan.marshall.Ids;
import java.io.IOException;
import java.util.List;
import org.infinispan.marshall.Marshallable;
import static java.lang.Math.min;
import java.io.ObjectInput;
import java.io.ObjectOutput;

/**
 * Adapter around a generic consistant hash to allow hashing of jgroups addresses.
 */
@Marshallable(externalizer = DefaultConsistentHash.Externalizer.class, id = Ids.DEFAULT_CONSISTENT_HASH)
public class AddressHash extends AbstractConsistentHash {
    /** A Weight and weight factor of 1 gives one node per address. */
    private static final int WEIGHT        = 1;
    /** Weightfactor and weight control the number of virtual nodes in the consistent hash. */
    private static final int WEIGHT_FACTOR = 1;

    // make sure all threads see the current list
    ArrayList<Address>             addresses;
    GenericConsistentHash<Address> chash;

    public void setCaches(List<Address> caches) {

      addresses = new ArrayList<Address>(caches.size());

      Address[] rawAddresses = caches.toArray(new Address[caches.size()]);
      int[]     weights      = new int[rawAddresses.length];
      Arrays.fill(weights, WEIGHT);

      // A consistent hash of the addresses, with no virtual nodes.
      chash = new GenericConsistentHash.Builder<Address>(rawAddresses)
                  .weights(weights)
                  .weightFactor(WEIGHT_FACTOR).build();

      int poolSize              = chash.getPoolSize();
      Address[] hashedAddresses = new Address[poolSize];
      hashedAddresses           = chash.getPoolValues(hashedAddresses);

      for (int i=0; i<poolSize; i++)
      {
          addresses.add(hashedAddresses[i]);
      }
      addresses.trimToSize();
   }

   public List<Address> getCaches() {
      return addresses;
   }

   public List<Address> locate(Object key, int replCount) {
      int clusterSize     = addresses.size();
      int numCopiesToFind = min(replCount, clusterSize);

      // Loads the primary owner of the key, and replication servers into the owners array.
      Address[]  owners = chash.locate(key, new Address[numCopiesToFind], numCopiesToFind);

      return Arrays.asList(owners);
   }

   /**
    * The distance between the first entries in the address array for two caches, a1 and a2.
    * Of questionable use when virtual nodes are employed.
    *
    * @param a1 The address of the first cache.
    * @param a2 The address of the second cache.
    *
    * @return Am int containing the difference between these two indices.
    */
   public int getDistance(Address a1, Address a2) {
      if (a1 == null || a2 == null) throw new NullPointerException("Cannot find the distance between null servers.");

      int p1 = addresses.indexOf(a1);
      if (p1 < 0)
         throw new IllegalArgumentException("Address " + a1 + " not in the addresses list of this consistent hash impl!");

      int p2 = addresses.indexOf(a2);
      if (p2 < 0)
         throw new IllegalArgumentException("Address " + a2 + " not in the addresses list of this consistent hash impl!");

      if (p1 <= p2)
         return p2 - p1;
      else
         return addresses.size() - (p1 - p2);
   }
   
   /**
    * Two hashes are adjacent if they are next to each other in the consistent hash.
    * @param a1 The address of the first cache.
    * @param a2 The address of the second cache.
    * 
    * @return A boolean, true if they are adjacent, false if not.
    */
   public boolean isAdjacent(Address a1, Address a2) {
      int distance = getDistance(a1, a2);
      return distance == 1 || distance == addresses.size() - 1;
   }

   @Override
   public boolean equals(Object o) {
      if (this == o) return true;
      if (o == null || getClass() != o.getClass()) return false;

      AddressHash that = (AddressHash) o;

      if (addresses != null ? !addresses.equals(that.addresses) : that.addresses != null) return false;
      if (chash != null ? !chash.equals(that.chash) : that.chash != null) return false;

      return true;
   }

   @Override
   public int hashCode() {
      int result = addresses != null ? addresses.hashCode() : 0;
      result = 31 * result + (chash != null ? chash.hashCode() : 0);
      return result;
   }

   public static class Externalizer implements org.infinispan.marshall.Externalizer {
      public void writeObject(ObjectOutput output, Object subject) throws IOException {
         DefaultConsistentHash dch = (DefaultConsistentHash) subject;
         output.writeObject(dch.addresses);
         output.writeObject(dch.positions);
      }

      @SuppressWarnings("unchecked")
      public Object readObject(ObjectInput unmarshaller) throws IOException, ClassNotFoundException {
         AddressHash dch = new AddressHash();
         dch.addresses = (ArrayList<Address>) unmarshaller.readObject();
         dch.chash = (GenericConsistentHash<Address>) unmarshaller.readObject();
         return dch;
      }
   }

}