/**
 * 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.hadoop.hbase.quotas;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.hbase.TableName;
import org.apache.yetus.audience.InterfaceAudience;
import org.apache.yetus.audience.InterfaceStability;
import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.Quotas;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;

/**
 * In-Memory state of the user quotas
 */
@InterfaceAudience.Private
@InterfaceStability.Evolving
@edu.umd.cs.findbugs.annotations.SuppressWarnings(value="IS2_INCONSISTENT_SYNC",
  justification="FindBugs seems confused; says bypassGlobals, namepaceLimiters, and " +
    "tableLimiters are mostly synchronized...but to me it looks like they are totally synchronized")
public class UserQuotaState extends QuotaState {
  private Map<String, QuotaLimiter> namespaceLimiters = null;
  private Map<TableName, QuotaLimiter> tableLimiters = null;
  private boolean bypassGlobals = false;

  public UserQuotaState() {
    super();
  }

  public UserQuotaState(final long updateTs) {
    super(updateTs);
  }

  @Override
  public synchronized String toString() {
    StringBuilder builder = new StringBuilder();
    builder.append("UserQuotaState(ts=" + getLastUpdate());
    if (bypassGlobals) builder.append(" bypass-globals");

    if (isBypass()) {
      builder.append(" bypass");
    } else {
      if (getGlobalLimiterWithoutUpdatingLastQuery() != NoopQuotaLimiter.get()) {
        builder.append(" global-limiter");
      }

      if (tableLimiters != null && !tableLimiters.isEmpty()) {
        builder.append(" [");
        for (TableName table: tableLimiters.keySet()) {
          builder.append(" " + table);
        }
        builder.append(" ]");
      }

      if (namespaceLimiters != null && !namespaceLimiters.isEmpty()) {
        builder.append(" [");
        for (String ns: namespaceLimiters.keySet()) {
          builder.append(" " + ns);
        }
        builder.append(" ]");
      }
    }
    builder.append(')');
    return builder.toString();
  }

  /**
   * @return true if there is no quota information associated to this object
   */
  @Override
  public synchronized boolean isBypass() {
    return !bypassGlobals &&
        getGlobalLimiterWithoutUpdatingLastQuery() == NoopQuotaLimiter.get() &&
        (tableLimiters == null || tableLimiters.isEmpty()) &&
        (namespaceLimiters == null || namespaceLimiters.isEmpty());
  }

  public synchronized boolean hasBypassGlobals() {
    return bypassGlobals;
  }

  @Override
  public synchronized void setQuotas(final Quotas quotas) {
    super.setQuotas(quotas);
    bypassGlobals = quotas.getBypassGlobals();
  }

  /**
   * Add the quota information of the specified table.
   * (This operation is part of the QuotaState setup)
   */
  public synchronized void setQuotas(final TableName table, Quotas quotas) {
    tableLimiters = setLimiter(tableLimiters, table, quotas);
  }

  /**
   * Add the quota information of the specified namespace.
   * (This operation is part of the QuotaState setup)
   */
  public void setQuotas(final String namespace, Quotas quotas) {
    namespaceLimiters = setLimiter(namespaceLimiters, namespace, quotas);
  }

  private <K> Map<K, QuotaLimiter> setLimiter(Map<K, QuotaLimiter> limiters,
      final K key, final Quotas quotas) {
    if (limiters == null) {
      limiters = new HashMap<>();
    }

    QuotaLimiter limiter = quotas.hasThrottle() ?
      QuotaLimiterFactory.fromThrottle(quotas.getThrottle()) : null;
    if (limiter != null && !limiter.isBypass()) {
      limiters.put(key, limiter);
    } else {
      limiters.remove(key);
    }
    return limiters;
  }

  /**
   * Perform an update of the quota state based on the other quota state object.
   * (This operation is executed by the QuotaCache)
   */
  @Override
  public synchronized void update(final QuotaState other) {
    super.update(other);

    if (other instanceof UserQuotaState) {
      UserQuotaState uOther = (UserQuotaState)other;
      tableLimiters = updateLimiters(tableLimiters, uOther.tableLimiters);
      namespaceLimiters = updateLimiters(namespaceLimiters, uOther.namespaceLimiters);
      bypassGlobals = uOther.bypassGlobals;
    } else {
      tableLimiters = null;
      namespaceLimiters = null;
      bypassGlobals = false;
    }
  }

  private static <K> Map<K, QuotaLimiter> updateLimiters(final Map<K, QuotaLimiter> map,
      final Map<K, QuotaLimiter> otherMap) {
    if (map == null) {
      return otherMap;
    }

    if (otherMap != null) {
      // To Remove
      Set<K> toRemove = new HashSet<>(map.keySet());
      toRemove.removeAll(otherMap.keySet());
      map.keySet().removeAll(toRemove);

      // To Update/Add
      for (final Map.Entry<K, QuotaLimiter> entry: otherMap.entrySet()) {
        QuotaLimiter limiter = map.get(entry.getKey());
        if (limiter == null) {
          limiter = entry.getValue();
        } else {
          limiter = QuotaLimiterFactory.update(limiter, entry.getValue());
        }
        map.put(entry.getKey(), limiter);
      }
      return map;
    }
    return null;
  }

  /**
   * Return the limiter for the specified table associated with this quota.
   * If the table does not have its own quota limiter the global one will be returned.
   * In case there is no quota limiter associated with this object a noop limiter will be returned.
   *
   * @return the quota limiter for the specified table
   */
  public synchronized QuotaLimiter getTableLimiter(final TableName table) {
    lastQuery = EnvironmentEdgeManager.currentTime();
    if (tableLimiters != null) {
      QuotaLimiter limiter = tableLimiters.get(table);
      if (limiter != null) return limiter;
    }
    if (namespaceLimiters != null) {
      QuotaLimiter limiter = namespaceLimiters.get(table.getNamespaceAsString());
      if (limiter != null) return limiter;
    }
    return getGlobalLimiterWithoutUpdatingLastQuery();
  }
}