/*
 * 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.solr.client.solrj.routing;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.ListIterator;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.params.SolrParams;
import org.apache.solr.common.util.Hash;

/**
 * Allows better caching by establishing deterministic evenly-distributed replica routing
 * preferences according to either explicitly configured hash routing parameter, or the hash of a
 * query parameter (configurable, usually related to the main query).
 */
public class AffinityReplicaListTransformer implements ReplicaListTransformer {

  private final int routingDividend;

  private AffinityReplicaListTransformer(String hashVal) {
    this.routingDividend = Math.abs(Hash.lookup3ycs(hashVal, 0, hashVal.length(), 0));
  }

  private AffinityReplicaListTransformer(int routingDividend) {
    this.routingDividend = routingDividend;
  }

  /**
   * @param dividendParam int param to be used directly for mod-based routing
   * @param hashParam String param to be hashed into an int for mod-based routing
   * @param requestParams the parameters of the Solr request
   * @return null if specified routing vals are not able to be parsed properly
   */
  public static ReplicaListTransformer getInstance(
      String dividendParam, String hashParam, SolrParams requestParams) {
    Integer dividendVal;
    if (dividendParam != null && (dividendVal = requestParams.getInt(dividendParam)) != null) {
      return new AffinityReplicaListTransformer(dividendVal);
    }
    String hashVal;
    if (hashParam != null
        && (hashVal = requestParams.get(hashParam)) != null
        && !hashVal.isEmpty()) {
      return new AffinityReplicaListTransformer(hashVal);
    } else {
      return null;
    }
  }

  @Override
  public <T> void transform(List<T> choices) {
    int size = choices.size();
    if (size > 1) {
      int i = 0;
      ArrayList<SortableChoice<T>> sortableChoices = new ArrayList<>(choices.size());
      for (T choice : choices) {
        sortableChoices.add(new SortableChoice<>(choice));
      }
      sortableChoices.sort(SORTABLE_CHOICE_COMPARATOR);
      ListIterator<T> iter = choices.listIterator();
      i = routingDividend % size;
      final int limit = i + size;
      do {
        iter.next();
        iter.set(sortableChoices.get(i % size).choice);
      } while (++i < limit);
    }
  }

  private static final class SortableChoice<T> {

    private final T choice;
    private final String sortableCoreLabel;

    private SortableChoice(T choice) {
      this.choice = choice;
      if (choice instanceof Replica) {
        this.sortableCoreLabel = ((Replica) choice).getCoreUrl();
      } else if (choice instanceof String) {
        this.sortableCoreLabel = (String) choice;
      } else {
        throw new IllegalArgumentException("can't handle type " + choice.getClass());
      }
    }
  }

  private static final Comparator<SortableChoice<?>> SORTABLE_CHOICE_COMPARATOR =
      Comparator.comparing(o -> o.sortableCoreLabel);
}
