/*
 * 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.conf.Configuration;
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;

/**
 * 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;

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

    if (isBypass()) {
      builder.append(" bypass");
    } else {
      if (getGlobalLimiter() != 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();
  }

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

  public synchronized boolean hasBypassGlobals() {
    return bypassGlobals;
  }

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

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

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

  public boolean hasTableLimiters() {
    return tableLimiters != null && !tableLimiters.isEmpty();
  }

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

    QuotaLimiter limiter =
      quotas.hasThrottle() ? QuotaLimiterFactory.fromThrottle(conf, 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) {
    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 getGlobalLimiter();
  }
}