package org.apache.commons.jcs.access;
   
   /*
    * 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.
   */
  
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Map;
  import java.util.Set;
  
  import org.apache.commons.jcs.JCS;
  import org.apache.commons.jcs.access.behavior.ICacheAccess;
  import org.apache.commons.jcs.access.exception.CacheException;
  import org.apache.commons.jcs.access.exception.ConfigurationException;
  import org.apache.commons.jcs.engine.behavior.ICacheElement;
  import org.apache.commons.jcs.engine.behavior.ICompositeCacheAttributes;
  import org.apache.commons.jcs.engine.behavior.IElementAttributes;
  import org.apache.commons.jcs.engine.stats.behavior.ICacheStats;
  import org.apache.commons.jcs.utils.props.AbstractPropertyContainer;
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  
  /**
   * TODO:  Add new methods that will allow you to provide a partition indicator for all major calls.  Add an interface as well.
   * <p>
   * This handles dividing puts and gets.
   * <p>
   * There are two required properties.
   * <p>
   * <ol>
   * <li>.numberOfPartitions</li>
   * <li>.partitionRegionNamePrefix</li>
   * </ol>
   * System properties will override values in the properties file.
   * <p>
   * We use a JCS region name for each partition that looks like this: partitionRegionNamePrefix + "_"
   * + partitionNumber. The number is 0 indexed based.
   * <p>
   * @author Aaron Smuts
   */
  public class PartitionedCacheAccess<K, V>
      extends AbstractPropertyContainer
      implements ICacheAccess<K, V>
  {
      /** the logger. */
      private static final Log log = LogFactory.getLog( PartitionedCacheAccess.class );
  
      /** The number of partitions. */
      private int numberOfPartitions = 1;
  
      /**
       * We use a JCS region name for each partition that looks like this: partitionRegionNamePrefix +
       * "_" + partitionNumber
       */
      private String partitionRegionNamePrefix;
  
      /** An array of partitions built during initialization. */
      private ICacheAccess<K, V>[] partitions;
  
      /** Is the class initialized. */
      private boolean initialized = false;
  
      /** Sets default properties heading and group. */
      public PartitionedCacheAccess()
      {
          setPropertiesHeading( "PartitionedCacheAccess" );
          setPropertiesGroup( "cache" );
      }
  
      /**
       * Puts the value into the appropriate cache partition.
       * <p>
       * @param key key
       * @param object object
       * @throws CacheException on configuration problem
       */
      @Override
      public void put( K key, V object )
          throws CacheException
      {
          if ( key == null || object == null )
          {
             log.warn( "Bad input key [" + key + "].  Cannot put null into the cache." );
             return;
         }
 
         if (!ensureInit())
         {
             return;
         }
 
         int partition = getPartitionNumberForKey( key );
         try
         {
             partitions[partition].put( key, object );
         }
         catch ( CacheException e )
         {
             log.error( "Problem putting value for key [" + key + "] in cache [" + partitions[partition] + "]" );
             throw e;
         }
     }
 
     /**
      * Puts in cache if an item does not exist with the name in that region.
      * <p>
      * @param key
      * @param object
      * @throws CacheException
      */
     @Override
     public void putSafe( K key, V object )
         throws CacheException
     {
         if ( key == null || object == null )
         {
             log.warn( "Bad input key [" + key + "].  Cannot putSafe null into the cache." );
         }
 
         if (!ensureInit())
         {
             return;
         }
 
         int partition = getPartitionNumberForKey( key );
         partitions[partition].putSafe( key, object );
     }
 
     /**
      * Puts the value into the appropriate cache partition.
      * <p>
      * @param key key
      * @param object object
      * @param attr
      * @throws CacheException on configuration problem
      */
     @Override
     public void put( K key, V object, IElementAttributes attr )
         throws CacheException
     {
         if ( key == null || object == null )
         {
             log.warn( "Bad input key [" + key + "].  Cannot put null into the cache." );
             return;
         }
 
         if (!ensureInit())
         {
             return;
         }
 
         int partition = getPartitionNumberForKey( key );
 
         try
         {
             partitions[partition].put( key, object, attr );
         }
         catch ( CacheException e )
         {
             log.error( "Problem putting value for key [" + key + "] in cache [" + partitions[partition] + "]" );
             throw e;
         }
     }
 
     /**
      * Gets the object for the key from the desired partition.
      * <p>
      * @param key key
      * @return result, null if not found.
      */
     @Override
     public V get( K key )
     {
         if ( key == null )
         {
             log.warn( "Input key is null." );
             return null;
         }
 
         if (!ensureInit())
         {
             return null;
         }
 
         int partition = getPartitionNumberForKey( key );
 
         return partitions[partition].get( key );
     }
 
     /**
      * Gets the ICacheElement&lt;K, V&gt; (the wrapped object) for the key from the desired partition.
      * <p>
      * @param key key
      * @return result, null if not found.
      */
     @Override
     public ICacheElement<K, V> getCacheElement( K key )
     {
         if ( key == null )
         {
             log.warn( "Input key is null." );
             return null;
         }
 
         if (!ensureInit())
         {
             return null;
         }
 
         int partition = getPartitionNumberForKey( key );
 
         return partitions[partition].getCacheElement( key );
     }
 
     /**
      * This is a getMultiple. We try to group the keys so that we make as few calls as needed.
      * <p>
      * @param names
      * @return Map of keys to ICacheElement
      */
     @Override
     public Map<K, ICacheElement<K, V>> getCacheElements( Set<K> names )
     {
         if ( names == null )
         {
             log.warn( "Bad input names cannot be null." );
             return Collections.emptyMap();
         }
 
         if (!ensureInit())
         {
             return Collections.emptyMap();
         }
 
         @SuppressWarnings("unchecked") // No generic arrays in java
         Set<K>[] dividedNames = new Set[this.getNumberOfPartitions()];
 
         for (K key : names)
         {
             int partition = getPartitionNumberForKey( key );
             if ( dividedNames[partition] == null )
             {
                 dividedNames[partition] = new HashSet<K>();
             }
             dividedNames[partition].add( key );
         }
 
         Map<K, ICacheElement<K, V>> result = new HashMap<K, ICacheElement<K, V>>();
         for ( int i = 0; i < partitions.length; i++ )
         {
             if ( dividedNames[i] != null && !dividedNames[i].isEmpty() )
             {
                 result.putAll( partitions[i].getCacheElements( dividedNames[i] ) );
             }
         }
         return result;
     }
 
     /**
      * This is tricky. Do we need to get from all the partitions?
      * <p>
      * If this interface took an object, we could use the hashcode to determine the partition. Then
      * we could use the toString for the pattern.
      * <p>
      * @param pattern
      * @return HashMap key to value
      */
     @Override
     public Map<K, V> getMatching( String pattern )
     {
         if ( pattern == null )
         {
             log.warn( "Input pattern is null." );
             return null;
         }
 
         if (!ensureInit())
         {
             return null;
         }
 
         Map<K, V> result = new HashMap<K, V>();
         for (ICacheAccess<K, V> partition : partitions)
         {
             result.putAll( partition.getMatching( pattern ) );
         }
 
         return result;
     }
 
     /**
      * This is tricky. Do we need to get from all the partitions?
      * <p>
      * @param pattern
      * @return HashMap key to ICacheElement
      */
     @Override
     public Map<K, ICacheElement<K, V>> getMatchingCacheElements( String pattern )
     {
         if ( pattern == null )
         {
             log.warn( "Input pattern is null." );
             return null;
         }
 
         if (!ensureInit())
         {
             return null;
         }
 
         Map<K, ICacheElement<K, V>> result = new HashMap<K, ICacheElement<K, V>>();
         for (ICacheAccess<K, V> partition : partitions)
         {
             result.putAll( partition.getMatchingCacheElements( pattern ) );
         }
         return result;
     }
 
     /**
      * Removes the item from the appropriate partition.
      * <p>
      * @param key
      * @throws CacheException
      */
     @Override
     public void remove( K key )
         throws CacheException
     {
         if ( key == null )
         {
             log.warn( "Input key is null. Cannot remove null from the cache." );
             return;
         }
 
         if (!ensureInit())
         {
             return;
         }
 
         int partition = getPartitionNumberForKey( key );
         try
         {
             partitions[partition].remove( key );
         }
         catch ( CacheException e )
         {
             log.error( "Problem removing value for key [" + key + "] in cache [" + partitions[partition] + "]" );
             throw e;
         }
     }
 
     /**
      * Calls free on each partition.
      * <p>
      * @param numberToFree
      * @return number removed
      * @throws CacheException
      */
     @Override
     public int freeMemoryElements( int numberToFree )
         throws CacheException
     {
         if (!ensureInit())
         {
             return 0;
         }
 
         int count = 0;
         for (ICacheAccess<K, V> partition : partitions)
         {
             count += partition.freeMemoryElements( numberToFree );
         }
         return count;
     }
 
     /**
      * @return ICompositeCacheAttributes from the first partition.
      */
     @Override
     public ICompositeCacheAttributes getCacheAttributes()
     {
         if (!ensureInit())
         {
             return null;
         }
 
         if ( partitions.length == 0 )
         {
             return null;
         }
 
         return partitions[0].getCacheAttributes();
     }
 
     /**
      * @return IElementAttributes from the first partition.
      * @throws CacheException
      */
     @Override
     public IElementAttributes getDefaultElementAttributes()
         throws CacheException
     {
         if (!ensureInit())
         {
             return null;
         }
 
         if ( partitions.length == 0 )
         {
             return null;
         }
 
         return partitions[0].getDefaultElementAttributes();
     }
 
     /**
      * This is no more efficient than simply getting the cache element.
      * <p>
      * @param key
      * @return IElementAttributes
      * @throws CacheException
      */
     @Override
     public IElementAttributes getElementAttributes( K key )
         throws CacheException
     {
         if ( key == null )
         {
             log.warn( "Input key is null. Cannot getElementAttributes for null from the cache." );
             return null;
         }
 
         if (!ensureInit())
         {
             return null;
         }
 
         int partition = getPartitionNumberForKey( key );
 
         return partitions[partition].getElementAttributes( key );
     }
 
     /**
      * Resets the attributes for this item. This has the same effect as an update, in most cases.
      * None of the auxiliaries are optimized to do this more efficiently than a simply update.
      * <p>
      * @param key
      * @param attributes
      * @throws CacheException
      */
     @Override
     public void resetElementAttributes( K key, IElementAttributes attributes )
         throws CacheException
     {
         if ( key == null )
         {
             log.warn( "Input key is null. Cannot resetElementAttributes for null." );
             return;
         }
 
         if (!ensureInit())
         {
             return;
         }
 
         int partition = getPartitionNumberForKey( key );
 
         partitions[partition].resetElementAttributes( key, attributes );
     }
 
     /**
      * Sets the attributes on all the partitions.
      * <p>
      * @param cattr
      */
     @Override
     public void setCacheAttributes( ICompositeCacheAttributes cattr )
     {
         if (!ensureInit())
         {
             return;
         }
 
         for (ICacheAccess<K, V> partition : partitions)
         {
             partition.setCacheAttributes( cattr );
         }
     }
 
     /**
      * Removes all of the elements from a region.
      * <p>
      * @throws CacheException
      */
     @Override
     public void clear()
         throws CacheException
     {
         if (!ensureInit())
         {
             return;
         }
 
         for (ICacheAccess<K, V> partition : partitions)
         {
             partition.clear();
         }
     }
 
     /**
      * This method is does not reset the attributes for items already in the cache. It could
      * potentially do this for items in memory, and maybe on disk (which would be slow) but not
      * remote items. Rather than have unpredictable behavior, this method just sets the default
      * attributes. Items subsequently put into the cache will use these defaults if they do not
      * specify specific attributes.
      * <p>
      * @param attr the default attributes.
      * @throws CacheException if something goes wrong.
      */
     @Override
     public void setDefaultElementAttributes( IElementAttributes attr )
         throws CacheException
     {
         if (!ensureInit())
         {
             return;
         }
 
         for (ICacheAccess<K, V> partition : partitions)
         {
             partition.setDefaultElementAttributes(attr);
         }
     }
 
     /**
      * This returns the ICacheStats object with information on this region and its auxiliaries.
      * <p>
      * This data can be formatted as needed.
      * <p>
      * @return ICacheStats
      */
     @Override
     public ICacheStats getStatistics()
     {
         if (!ensureInit())
         {
             return null;
         }
 
         if ( partitions.length == 0 )
         {
             return null;
         }
 
         return partitions[0].getStatistics();
     }
 
     /**
      * @return A String version of the stats.
      */
     @Override
     public String getStats()
     {
         if (!ensureInit())
         {
             return "";
         }
 
         StringBuilder stats = new StringBuilder();
         for (ICacheAccess<K, V> partition : partitions)
         {
             stats.append(partition.getStats());
             stats.append("\n");
         }
 
         return stats.toString();
     }
 
     /**
      * Dispose this region. Flushes objects to and closes auxiliary caches. This is a shutdown
      * command!
      * <p>
      * To simply remove all elements from the region use clear().
      */
     @Override
     public synchronized void dispose()
     {
         if (!ensureInit())
         {
             return;
         }
 
         for (ICacheAccess<K, V> partition : partitions)
         {
             partition.dispose();
         }
 
         initialized = false;
     }
 
     /**
      * This expects a numeric key. If the key cannot be converted into a number, we will return 0.
      * TODO we could md5 it or get the hashcode.
      * <p>
      * We determine the partition by taking the mod of the number of partitions.
      * <p>
      * @param key key
      * @return the partition number.
      */
     protected int getPartitionNumberForKey( K key )
     {
         if ( key == null )
         {
             return 0;
         }
 
         long keyNum = getNumericValueForKey( key );
 
         int partition = (int) ( keyNum % getNumberOfPartitions() );
 
         if ( log.isDebugEnabled() )
         {
             log.debug( "Using partition [" + partition + "] for key [" + key + "]" );
         }
 
         return partition;
     }
 
     /**
      * This can be overridden for special purposes.
      * <p>
      * @param key key
      * @return long
      */
     public long getNumericValueForKey( K key )
     {
         String keyString = key.toString();
         long keyNum = -1;
         try
         {
             keyNum = Long.parseLong( keyString );
         }
         catch ( NumberFormatException e )
         {
             // THIS IS UGLY, but I can't think of a better failsafe right now.
             keyNum = key.hashCode();
             log.warn( "Couldn't convert [" + key + "] into a number.  Will use hashcode [" + keyNum + "]" );
         }
         return keyNum;
     }
 
     /**
      * Initialize if we haven't already.
      * <p>
      * @throws ConfigurationException on configuration problem
      */
     protected synchronized boolean ensureInit()
     {
         if ( !initialized )
         {
             try
             {
                 initialize();
             }
             catch ( ConfigurationException e )
             {
                 log.error( "Couldn't configure partioned access.", e );
                 return false;
             }
         }
 
         return true;
     }
 
     /**
      * Use the partition prefix and the number of partitions to get JCS regions.
      * <p>
      * @throws ConfigurationException on configuration problem
      */
     protected synchronized void initialize()
         throws ConfigurationException
     {
         ensureProperties();
 
         @SuppressWarnings("unchecked") // No generic arrays in java
         ICacheAccess<K, V>[] tempPartitions = new ICacheAccess[this.getNumberOfPartitions()];
         for ( int i = 0; i < this.getNumberOfPartitions(); i++ )
         {
             String regionName = this.getPartitionRegionNamePrefix() + "_" + i;
             try
             {
                 tempPartitions[i] = JCS.getInstance( regionName );
             }
             catch ( CacheException e )
             {
                 log.error( "Problem getting cache for region [" + regionName + "]" );
             }
         }
 
         partitions = tempPartitions;
         initialized = true;
     }
 
     /**
      * Loads in the needed configuration settings. System properties are checked first. A system
      * property will override local property value.
      * <p>
      * Loads the following JCS Cache specific properties:
      * <ul>
      * <li>heading.numberOfPartitions</li>
      * <li>heading.partitionRegionNamePrefix</li>
      * </ul>
      * @throws ConfigurationException on configuration problem
      */
     @Override
     protected void handleProperties()
         throws ConfigurationException
     {
         // Number of Partitions.
         String numberOfPartitionsPropertyName = this.getPropertiesHeading() + ".numberOfPartitions";
         String numberOfPartitionsPropertyValue = getPropertyForName( numberOfPartitionsPropertyName, true );
         try
         {
             this.setNumberOfPartitions( Integer.parseInt( numberOfPartitionsPropertyValue ) );
         }
         catch ( NumberFormatException e )
         {
             String message = "Could not convert [" + numberOfPartitionsPropertyValue + "] into a number for ["
                 + numberOfPartitionsPropertyName + "]";
             log.error( message );
             throw new ConfigurationException( message );
         }
 
         // Partition Name Prefix.
         String prefixPropertyName = this.getPropertiesHeading() + ".partitionRegionNamePrefix";
         String prefix = getPropertyForName( prefixPropertyName, true );
         this.setPartitionRegionNamePrefix( prefix );
     }
 
     /**
      * Checks the system properties before the properties.
      * <p>
      * @param propertyName name
      * @param required is it required?
      * @return the property value if one is found
      * @throws ConfigurationException thrown if it is required and not found.
      */
     protected String getPropertyForName( String propertyName, boolean required )
         throws ConfigurationException
     {
         String propertyValue = null;
         propertyValue = System.getProperty( propertyName );
         if ( propertyValue != null )
         {
             if ( log.isInfoEnabled() )
             {
                 log.info( "Found system property override: Name [" + propertyName + "] Value [" + propertyValue + "]" );
             }
         }
         else
         {
             propertyValue = this.getProperties().getProperty( propertyName );
             if ( required && propertyValue == null )
             {
                 String message = "Could not find required property [" + propertyName + "] in propertiesGroup ["
                     + this.getPropertiesGroup() + "]";
                 log.error( message );
                 throw new ConfigurationException( message );
             }
             else
             {
                 if ( log.isInfoEnabled() )
                 {
                     log.info( "Name [" + propertyName + "] Value [" + propertyValue + "]" );
                 }
             }
         }
         return propertyValue;
     }
 
     /**
      * @param numberOfPartitions The numberOfPartitions to set.
      */
     protected void setNumberOfPartitions( int numberOfPartitions )
     {
         this.numberOfPartitions = numberOfPartitions;
     }
 
     /**
      * @return Returns the numberOfPartitions.
      */
     protected int getNumberOfPartitions()
     {
         return numberOfPartitions;
     }
 
     /**
      * @param partitionRegionNamePrefix The partitionRegionNamePrefix to set.
      */
     protected void setPartitionRegionNamePrefix( String partitionRegionNamePrefix )
     {
         this.partitionRegionNamePrefix = partitionRegionNamePrefix;
     }
 
     /**
      * @return Returns the partitionRegionNamePrefix.
      */
     protected String getPartitionRegionNamePrefix()
     {
         return partitionRegionNamePrefix;
     }
 
     /**
      * @param partitions The partitions to set.
      */
     protected void setPartitions( ICacheAccess<K, V>[] partitions )
     {
         this.partitions = partitions;
     }
 
     /**
      * @return Returns the partitions.
      */
     protected ICacheAccess<K, V>[] getPartitions()
     {
         return partitions;
     }
 }