/**
 * Copyright The Apache Software Foundation
 *
 * 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;

import java.io.DataOutput;
import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.yetus.audience.InterfaceAudience;

import org.apache.hbase.thirdparty.org.apache.commons.collections4.iterators.UnmodifiableIterator;

/**
 * Do a shallow merge of multiple KV configuration pools. This is a very useful
 * utility class to easily add per-object configurations in addition to wider
 * scope settings. This is different from Configuration.addResource()
 * functionality, which performs a deep merge and mutates the common data
 * structure.
 * <p>
 * The iterator on CompoundConfiguration is unmodifiable. Obtaining iterator is an expensive
 * operation.
 * <p>
 * For clarity: the shallow merge allows the user to mutate either of the
 * configuration objects and have changes reflected everywhere. In contrast to a
 * deep merge, that requires you to explicitly know all applicable copies to
 * propagate changes.
 * 
 * WARNING: The values set in the CompoundConfiguration are do not handle Property variable
 * substitution.  However, if they are set in the underlying configuration substitutions are
 * done. 
 */
@InterfaceAudience.Private
public class CompoundConfiguration extends Configuration {

  private Configuration mutableConf = null;

  /**
   * Default Constructor. Initializes empty configuration
   */
  public CompoundConfiguration() {
  }

  // Devs: these APIs are the same contract as their counterparts in
  // Configuration.java
  private interface ImmutableConfigMap extends Iterable<Map.Entry<String,String>> {
    String get(String key);
    String getRaw(String key);
    Class<?> getClassByName(String name) throws ClassNotFoundException;
    int size();
  }

  private final List<ImmutableConfigMap> configs = new ArrayList<>();

  static class ImmutableConfWrapper implements  ImmutableConfigMap {
   private final Configuration c;
    
    ImmutableConfWrapper(Configuration conf) {
      c = conf;
    }

    @Override
    public Iterator<Map.Entry<String,String>> iterator() {
      return c.iterator();
    }
    
    @Override
    public String get(String key) {
      return c.get(key);
    }

    @Override
    public String getRaw(String key) {
      return c.getRaw(key);
    }

    @Override
    public Class<?> getClassByName(String name)
        throws ClassNotFoundException {
      return c.getClassByName(name);
    }

    @Override
    public int size() {
      return c.size();
    }

    @Override
    public String toString() {
      return c.toString();
    }
  }

  /**
   * If set has been called, it will create a mutableConf.  This converts the mutableConf to an
   * immutable one and resets it to allow a new mutable conf.  This is used when a new map or
   * conf is added to the compound configuration to preserve proper override semantics.
   */
  void freezeMutableConf() {
    if (mutableConf == null) {
      // do nothing if there is no current mutableConf
      return;
    }

    this.configs.add(0, new ImmutableConfWrapper(mutableConf));
    mutableConf = null;
  }

  /**
   * Add Hadoop Configuration object to config list.
   * The added configuration overrides the previous ones if there are name collisions.
   * @param conf configuration object
   * @return this, for builder pattern
   */
  public CompoundConfiguration add(final Configuration conf) {
    freezeMutableConf();

    if (conf instanceof CompoundConfiguration) {
      this.configs.addAll(0, ((CompoundConfiguration) conf).configs);
      return this;
    }
    // put new config at the front of the list (top priority)
    this.configs.add(0, new ImmutableConfWrapper(conf));
    return this;
  }

  /**
   * Add Bytes map to config list. This map is generally
   * created by HTableDescriptor or HColumnDescriptor, but can be abstractly
   * used. The added configuration overrides the previous ones if there are
   * name collisions.
   *
   * @param map
   *          Bytes map
   * @return this, for builder pattern
   */
  public CompoundConfiguration addBytesMap(
      final Map<Bytes, Bytes> map) {
    freezeMutableConf();

    // put new map at the front of the list (top priority)
    this.configs.add(0, new ImmutableConfigMap() {
      private final Map<Bytes, Bytes> m = map;

      @Override
      public Iterator<Map.Entry<String,String>> iterator() {
        Map<String, String> ret = new HashMap<>();
        for (Map.Entry<Bytes, Bytes> entry : map.entrySet()) {
          String key = Bytes.toString(entry.getKey().get());
          String val = entry.getValue() == null ? null : Bytes.toString(entry.getValue().get());
          ret.put(key, val);
        }
        return ret.entrySet().iterator();
      }
      
      @Override
      public String get(String key) {
        Bytes ibw = new Bytes(Bytes
            .toBytes(key));
        if (!m.containsKey(ibw))
          return null;
        Bytes value = m.get(ibw);
        if (value == null || value.get() == null)
          return null;
        return Bytes.toString(value.get());
      }

      @Override
      public String getRaw(String key) {
        return get(key);
      }

      @Override
      public Class<?> getClassByName(String name)
      throws ClassNotFoundException {
        return null;
      }

      @Override
      public int size() {
        return m.size();
      }

      @Override
      public String toString() {
        return m.toString();
      }
    });
    return this;
  }

  /**
   * Add String map to config list. This map is generally created by HTableDescriptor
   * or HColumnDescriptor, but can be abstractly used. The added configuration
   * overrides the previous ones if there are name collisions.
   *
   * @return this, for builder pattern
   */
  public CompoundConfiguration addStringMap(final Map<String, String> map) {
    freezeMutableConf();

    // put new map at the front of the list (top priority)
    this.configs.add(0, new ImmutableConfigMap() {
      private final Map<String, String> m = map;

      @Override
      public Iterator<Map.Entry<String,String>> iterator() {
        return map.entrySet().iterator();
      }

      @Override
      public String get(String key) {
        return m.get(key);
      }

      @Override
      public String getRaw(String key) {
        return get(key);
      }

      @Override
      public Class<?> getClassByName(String name)
      throws ClassNotFoundException {
        return null;
      }

      @Override
      public int size() {
        return m.size();
      }

      @Override
      public String toString() {
        return m.toString();
      }
    });
    return this;
  }

  @Override
  public String toString() {
    StringBuilder sb = new StringBuilder();
    sb.append("CompoundConfiguration: " + this.configs.size() + " configs");
    for (ImmutableConfigMap m : this.configs) {
      sb.append(m);
    }
    return sb.toString();
  }

  @Override
  public String get(String key) {
    if (mutableConf != null) {
      String value = mutableConf.get(key);
      if (value != null) {
        return value;
      }
    }

    for (ImmutableConfigMap m : this.configs) {
      String value = m.get(key);
      if (value != null) {
        return value;
      }
    }
    return null;
  }

  @Override
  public String getRaw(String key) {
    if (mutableConf != null) {
      String value = mutableConf.getRaw(key);
      if (value != null) {
        return value;
      }
    }

    for (ImmutableConfigMap m : this.configs) {
      String value = m.getRaw(key);
      if (value != null) {
        return value;
      }
    }
    return null;
  }

  @Override
  public Class<?> getClassByName(String name) throws ClassNotFoundException {
    if (mutableConf != null) {
      Class<?> value = mutableConf.getClassByName(name);
      if (value != null) {
        return value;
      }
    }

    for (ImmutableConfigMap m : this.configs) {
      Class<?> value = m.getClassByName(name);
      if (value != null) {
        return value;
      }
    }
    throw new ClassNotFoundException();
  }

  // TODO: This method overestimates the number of configuration settings -- if a value is masked
  // by an overriding config or map, it will be counted multiple times. 
  @Override
  public int size() {
    int ret = 0;

    if (mutableConf != null) {
      ret += mutableConf.size();
    }

    for (ImmutableConfigMap m : this.configs) {
      ret += m.size();
    }
    return ret;
  }

  /**
   * Get the value of the <code>name</code>. If the key is deprecated,
   * it returns the value of the first key which replaces the deprecated key
   * and is not null.
   * If no such property exists,
   * then <code>defaultValue</code> is returned.

   * The CompooundConfiguration does not do property substitution.  To do so we need
   * Configuration.getProps to be protected or package visible.  Though in hadoop2 it is
   * protected, in hadoop1 the method is private and not accessible.
   * 
   * All of the get* methods call this overridden get method.
   * 
   * @param name property name.
   * @param defaultValue default value.
   * @return property value, or <code>defaultValue</code> if the property 
   *         doesn't exist.                    
   **/
  @Override
  public String get(String name, String defaultValue) {
    String ret = get(name);
    return ret == null ? defaultValue : ret;
  }

  @Override
  public Iterator<Map.Entry<String, String>> iterator() {
    Map<String, String> ret = new HashMap<>();

    // add in reverse order so that oldest get overridden.
    if (!configs.isEmpty()) {
      for (int i = configs.size() - 1; i >= 0; i--) {
        ImmutableConfigMap map = configs.get(i);
        Iterator<Map.Entry<String, String>> iter = map.iterator();
        while (iter.hasNext()) {
          Map.Entry<String, String> entry = iter.next();
          ret.put(entry.getKey(), entry.getValue());
        }
      }
    }

    // add mutations to this CompoundConfiguration last.
    if (mutableConf != null) {
      Iterator<Map.Entry<String, String>> miter = mutableConf.iterator();
      while (miter.hasNext()) {
        Map.Entry<String, String> entry = miter.next();
        ret.put(entry.getKey(), entry.getValue());
      }
    }

    return UnmodifiableIterator.unmodifiableIterator(ret.entrySet().iterator());
  }

  @Override
  public void set(String name, String value) {
    if (mutableConf == null) {
      // not thread safe
      mutableConf = new Configuration(false); // an empty configuration
    }
    mutableConf.set(name,  value);
  }

  /***********************************************************************************************
   * These methods are unsupported, and no code using CompoundConfiguration depend upon them.
   * Quickly abort upon any attempts to use them. 
   **********************************************************************************************/

  @Override
  public void clear() {
    throw new UnsupportedOperationException("Immutable Configuration");
  }

  @Override
  public void write(DataOutput out) throws IOException {
    throw new UnsupportedOperationException("Immutable Configuration");
  }

  @Override
  public void writeXml(OutputStream out) throws IOException {
    throw new UnsupportedOperationException("Immutable Configuration");
  }
}