content: /*
   * 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.bookkeeper.common.conf;
  
  import java.math.BigDecimal;
  import java.math.BigInteger;
  import java.net.URL;
  import java.time.Duration;
  import java.util.Collection;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.Properties;
  import org.apache.bookkeeper.common.util.JsonUtil;
  import org.apache.bookkeeper.common.util.JsonUtil.ParseJsonException;
  import org.apache.commons.configuration2.CompositeConfiguration;
  import org.apache.commons.configuration2.Configuration;
  import org.apache.commons.configuration2.ConfigurationDecoder;
  import org.apache.commons.configuration2.ImmutableConfiguration;
  import org.apache.commons.configuration2.ex.ConfigurationException;
  import org.apache.commons.configuration2.interpol.ConfigurationInterpolator;
  import org.apache.commons.configuration2.interpol.Lookup;
  import org.apache.commons.configuration2.sync.LockMode;
  import org.apache.commons.configuration2.sync.Synchronizer;
  
  /**
   * Component Configuration.
   */
  public abstract class ComponentConfiguration implements Configuration {
  
      protected static final String DELIMITER = ".";
  
      private final String componentPrefix;
      private final CompositeConfiguration underlyingConf;
      private final Configuration conf;
  
      protected ComponentConfiguration(CompositeConfiguration underlyingConf,
                                       String componentPrefix) {
          super();
          this.underlyingConf = underlyingConf;
          this.conf = new ConcurrentConfiguration();
          this.componentPrefix = componentPrefix;
  
          // load the component keys
          loadConf(underlyingConf);
      }
  
      protected String getKeyName(String name) {
          return name;
      }
  
      public CompositeConfiguration getUnderlyingConf() {
          return underlyingConf;
      }
  
      /**
       * Load configuration from a given {@code confURL}.
       *
       * @param confURL the url points to the configuration.
       * @throws ConfigurationException when failed to load configuration.
       */
      public void loadConf(URL confURL) throws ConfigurationException {
          Configuration loadedConf = ConfigurationUtil.newConfiguration(conf -> conf.propertiesBuilder(confURL));
          loadConf(loadedConf);
      }
  
      protected void loadConf(Configuration loadedConf) {
          loadedConf.getKeys().forEachRemaining(fullKey -> {
              if (fullKey.startsWith(componentPrefix)) {
                  String componentKey = fullKey.substring(componentPrefix.length());
                  setProperty(componentKey, loadedConf.getProperty(fullKey));
              }
          });
      }
  
      public void validate() throws ConfigurationException {
          // do nothing by default.
      }
  
      @Override
      public Configuration subset(String prefix) {
          return conf.subset(getKeyName(prefix));
      }
  
      @Override
      public boolean isEmpty() {
          return conf.isEmpty();
      }
  
      @Override
      public boolean containsKey(String key) {
          return conf.containsKey(getKeyName(key));
      }
  
      @Override
      public void addProperty(String key, Object value) {
          conf.addProperty(getKeyName(key), value);
      }
  
      @Override
      public void setProperty(String key, Object value) {
          conf.setProperty(getKeyName(key), value);
      }
  
      @Override
      public void clearProperty(String key) {
          conf.clearProperty(getKeyName(key));
      }
  
      @Override
      public void clear() {
          conf.clear();
      }
  
      @Override
      public Object getProperty(String key) {
          return conf.getProperty(getKeyName(key));
      }
  
      @Override
      public Iterator<String> getKeys(String prefix) {
          return conf.getKeys(getKeyName(prefix));
      }
  
      @Override
      public Iterator<String> getKeys() {
          return conf.getKeys();
      }
  
      @Override
      public Properties getProperties(String key) {
          return conf.getProperties(getKeyName(key));
      }
  
      @Override
      public boolean getBoolean(String key) {
          return conf.getBoolean(getKeyName(key));
      }
  
      @Override
      public boolean getBoolean(String key, boolean defaultValue) {
          return conf.getBoolean(getKeyName(key), defaultValue);
      }
  
      @Override
      public Boolean getBoolean(String key, Boolean defaultValue) {
          return conf.getBoolean(getKeyName(key), defaultValue);
      }
  
      @Override
      public byte getByte(String key) {
          return conf.getByte(getKeyName(key));
      }
  
      @Override
      public byte getByte(String key, byte defaultValue) {
          return conf.getByte(getKeyName(key), defaultValue);
      }
  
      @Override
      public Byte getByte(String key, Byte defaultValue) {
          return conf.getByte(getKeyName(key), defaultValue);
      }
  
      @Override
      public double getDouble(String key) {
          return conf.getDouble(getKeyName(key));
      }
  
      @Override
      public double getDouble(String key, double defaultValue) {
          return conf.getDouble(getKeyName(key), defaultValue);
      }
  
      @Override
      public Double getDouble(String key, Double defaultValue) {
          return conf.getDouble(getKeyName(key), defaultValue);
      }
  
      @Override
      public float getFloat(String key) {
          return conf.getFloat(getKeyName(key));
      }
  
      @Override
      public float getFloat(String key, float defaultValue) {
          return conf.getFloat(getKeyName(key), defaultValue);
      }
  
      @Override
      public Float getFloat(String key, Float defaultValue) {
          return conf.getFloat(getKeyName(key), defaultValue);
      }
  
      @Override
      public int getInt(String key) {
          return conf.getInt(getKeyName(key));
      }
  
      @Override
      public int getInt(String key, int defaultValue) {
          return conf.getInt(getKeyName(key), defaultValue);
      }
  
      @Override
      public Integer getInteger(String key, Integer defaultValue) {
          return conf.getInt(getKeyName(key), defaultValue);
      }
  
      @Override
      public long getLong(String key) {
          return conf.getLong(getKeyName(key));
      }
  
      @Override
      public long getLong(String key, long defaultValue) {
          return conf.getLong(getKeyName(key), defaultValue);
      }
  
      @Override
      public Long getLong(String key, Long defaultValue) {
          return conf.getLong(getKeyName(key), defaultValue);
      }
  
      @Override
      public short getShort(String key) {
          return conf.getShort(getKeyName(key));
      }
  
      @Override
      public short getShort(String key, short defaultValue) {
          return conf.getShort(getKeyName(key), defaultValue);
      }
  
      @Override
      public Short getShort(String key, Short defaultValue) {
          return conf.getShort(getKeyName(key), defaultValue);
      }
  
      @Override
      public BigDecimal getBigDecimal(String key) {
          return conf.getBigDecimal(getKeyName(key));
      }
  
      @Override
      public BigDecimal getBigDecimal(String key, BigDecimal defaultValue) {
          return conf.getBigDecimal(getKeyName(key), defaultValue);
      }
  
      @Override
      public BigInteger getBigInteger(String key) {
          return conf.getBigInteger(getKeyName(key));
      }
  
      @Override
      public BigInteger getBigInteger(String key, BigInteger defaultValue) {
          return conf.getBigInteger(getKeyName(key), defaultValue);
      }
  
      @Override
      public String getString(String key) {
          return conf.getString(getKeyName(key));
      }
  
      @Override
      public String getString(String key, String defaultValue) {
          return conf.getString(getKeyName(key), defaultValue);
      }
  
      @Override
      public String[] getStringArray(String key) {
          return conf.getStringArray(getKeyName(key));
      }
  
      @Override
      public List<Object> getList(String key) {
          return conf.getList(getKeyName(key));
      }
  
      @Override
      public List<Object> getList(String key, List<?> defaultValue) {
          return conf.getList(getKeyName(key), defaultValue);
      }
  
      @Override
      public ConfigurationInterpolator getInterpolator() {
          return conf.getInterpolator();
      }
  
      @Override
      public void installInterpolator(Map<String, ? extends Lookup> prefixLookups,
                                      Collection<? extends Lookup> defLookups) {
          conf.installInterpolator(prefixLookups, defLookups);
      }
  
      @Override
      public void setInterpolator(ConfigurationInterpolator ci) {
          conf.setInterpolator(ci);
      }
  
      @Override
      public <T> T get(Class<T> cls, String key) {
          return conf.get(cls, key);
      }
  
      @Override
      public <T> T get(Class<T> cls, String key, T defaultValue) {
          return conf.get(cls, key, defaultValue);
      }
  
      @Override
      public Object getArray(Class<?> cls, String key) {
          return conf.getArray(cls, key);
      }
  
      @Override
      @Deprecated
      public Object getArray(Class<?> cls, String key, Object defaultValue) {
          return conf.getArray(cls, key, defaultValue);
      }
  
      @Override
      public <T> Collection<T> getCollection(Class<T> cls, String key, Collection<T> target) {
          return conf.getCollection(cls, key, target);
      }
  
      @Override
      public <T> Collection<T> getCollection(Class<T> cls, String key, Collection<T> target, Collection<T> defaultValue) {
          return conf.getCollection(cls, key, target, defaultValue);
      }
  
      @Override
      public Duration getDuration(String key) {
          return conf.getDuration(key);
      }
  
      @Override
      public Duration getDuration(String key, Duration defaultValue) {
          return conf.getDuration(key, defaultValue);
      }
  
      @Override
      public String getEncodedString(String key) {
          return conf.getEncodedString(key);
      }
  
      @Override
      public String getEncodedString(String key, ConfigurationDecoder decoder) {
          return conf.getEncodedString(key, decoder);
      }
  
      @Override
      public <T extends Enum<T>> T getEnum(String key, Class<T> enumType) {
          return conf.getEnum(key, enumType);
      }
  
      @Override
      public <T extends Enum<T>> T getEnum(String key, Class<T> enumType, T defaultValue) {
          return conf.getEnum(key, enumType, defaultValue);
      }
  
      @Override
      public Iterator<String> getKeys(String prefix, String delimiter) {
          return conf.getKeys(prefix, delimiter);
      }
  
      @Override
      public <T> List<T> getList(Class<T> cls, String key) {
          return conf.getList(cls, key);
      }
  
      @Override
      public <T> List<T> getList(Class<T> cls, String key, List<T> defaultValue) {
          return conf.getList(cls, key, defaultValue);
      }
  
      @Override
      public ImmutableConfiguration immutableSubset(String prefix) {
          return conf.immutableSubset(prefix);
      }
  
      @Override
      public int size() {
          return conf.size();
      }
  
      @Override
      public Synchronizer getSynchronizer() {
          return conf.getSynchronizer();
      }
  
      @Override
      public void lock(LockMode mode) {
          conf.lock(mode);
      }
  
      @Override
      public void setSynchronizer(Synchronizer sync) {
          conf.setSynchronizer(sync);
      }
  
      @Override
      public void unlock(LockMode mode) {
          conf.unlock(mode);
      }
  
      /**
       * returns the string representation of json format of this config.
       *
       * @return
       * @throws ParseJsonException
       */
      public String asJson() {
          try {
              return JsonUtil.toJson(toMap());
          } catch (ParseJsonException e) {
              throw new RuntimeException("Failed to serialize the configuration as json", e);
          }
      }
  
      private Map<String, Object> toMap() {
          Map<String, Object> configMap = new HashMap<>();
          Iterator<String> iterator = this.getKeys();
          while (iterator.hasNext()) {
              String key = iterator.next();
              Object property = this.getProperty(key);
              if (property != null) {
                  configMap.put(key, property.toString());
              }
          }
          return configMap;
      }
  }
  