library/src/main/java/com/datatorrent/lib/db/cache/CacheManager.java - apex-malhar - Git at Google

 /**
  * 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 com.datatorrent.lib.db.cache;

 import java.io.Closeable;
 import java.io.IOException;
 import java.util.Calendar;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.Timer;
 import java.util.TimerTask;

 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 import javax.validation.constraints.NotNull;

 import org.codehaus.jackson.annotate.JsonIgnore;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import com.google.common.base.Strings;
 import com.google.common.collect.Lists;

 import com.datatorrent.api.Attribute;
 import com.datatorrent.api.Component;
 import com.datatorrent.api.Context;
 import com.datatorrent.lib.db.KeyValueStore;

 /**
  * Manages primary and secondary stores.<br/>
  * It firsts checks the primary store for a key. If the primary store doesn't have the key, it queries the backup
  * store and retrieves the value.<br/>
  * If the key was present in the backup store, its value is returned and also saved in the primary store.
  * <p/>
  * Typically primary store is faster but has limited size like memory and backup store is slower but unlimited like
  * databases.<br/>
  * Store Manager can also refresh the values of keys at a specified time every day.
  * This time is in format HH:mm:ss Z.<br/>
  * This is not thread-safe.
  *
  * @since 0.9.2
  */
 public class CacheManager implements Closeable
 {
   @NotNull
   protected transient Primary primary;
   @NotNull
   protected transient Backup backup;
   protected String refreshTime;
   private transient Timer refresher;

   private transient Attribute.AttributeMap attributeMap = new Attribute.AttributeMap.DefaultAttributeMap();
   private boolean readOnlyData = false;
   private int numInitCachedLines = -1;

   public CacheManager()
   {
     this.primary = new CacheStore();
   }

   public void initialize() throws IOException
   {
     Class[] parameters = new Class[1];
     parameters[0] = CacheContext.class;

     attributeMap.put(CacheContext.READ_ONLY_ATTR, readOnlyData);
     attributeMap.put(CacheContext.NUM_INIT_CACHED_LINES_ATTR, numInitCachedLines);
     CacheContext cacheContext = new CacheContext(attributeMap);

     if (primary instanceof Component) {
       try {
         //check Component for method of type setup(CacheManager.CacheContext context)
         primary.getClass().getMethod("setup", parameters);
         ((Component)primary).setup(cacheContext);
       } catch (NoSuchMethodException e) {
         //Store implements Component, but not Component<CacheManager.CacheContext>
       }
     }

     if (backup instanceof Component) {
       try {
         //check Component for method of type setup(CacheManager.CacheContext context)
         backup.getClass().getMethod("setup", parameters);
         ((Component)backup).setup(cacheContext);
       } catch (NoSuchMethodException e) {
         //Store implements Component, but not Component<CacheManager.CacheContext>
       }
     }

     primary.connect();
     backup.connect();
     Map<Object, Object> initialEntries = backup.loadInitialData();
     if (initialEntries != null) {
       primary.putAll(initialEntries);
     }

     if (!readOnlyData && !Strings.isNullOrEmpty(refreshTime)) {

       String[] parts = refreshTime.split("[:\\s]");

       Calendar timeToRefresh = Calendar.getInstance();
       timeToRefresh.set(Calendar.HOUR_OF_DAY, Integer.parseInt(parts[0]));
       if (parts.length >= 2) {
         timeToRefresh.set(Calendar.MINUTE, Integer.parseInt(parts[1]));
       }
       if (parts.length >= 3) {
         timeToRefresh.set(Calendar.SECOND, Integer.parseInt(parts[2]));
       }
       long initialDelay = timeToRefresh.getTimeInMillis() - Calendar.getInstance().getTimeInMillis();

       TimerTask task = new TimerTask()
       {
         @Override
         public void run()
         {
           List<Object> keysToRefresh = Lists.newArrayList(primary.getKeys());
           if (keysToRefresh.size() > 0) {
             List<Object> refreshedValues = backup.getAll(keysToRefresh);
             if (refreshedValues != null) {
               for (int i = 0; i < keysToRefresh.size(); i++) {
                 primary.put(keysToRefresh.get(i), refreshedValues.get(i));
               }
             }
           }
         }
       };

       refresher = new Timer();
       if (initialDelay < 0) {
         refresher.schedule(task, 0);
         timeToRefresh.add(Calendar.DAY_OF_MONTH, 1);
         initialDelay = timeToRefresh.getTimeInMillis();
       }
       refresher.scheduleAtFixedRate(task, initialDelay, 86400000);
     }
   }

   @Nullable
   public Object get(@Nonnull Object key)
   {
     Object primaryVal = primary.get(key);
     if (primaryVal != null) {
       return primaryVal;
     }

     Object backupVal = backup.get(key);
     if (backupVal != null) {
       primary.put(key, backupVal);
     }
     return backupVal;
   }

   public void put(@Nonnull Object key, @Nonnull Object value)
   {
     primary.put(key, value);
     backup.put(key, value);
   }

   @Override
   public void close() throws IOException
   {
     if (refresher != null) {
       refresher.cancel();
     }
     primary.disconnect();
     backup.disconnect();
   }

   public void setPrimary(Primary primary)
   {
     this.primary = primary;
   }

   @JsonIgnore
   public Primary getPrimary()
   {
     return primary;
   }

   public void setBackup(Backup backup)
   {
     this.backup = backup;
   }

   @JsonIgnore
   public Backup getBackup()
   {
     return backup;
   }

   /**
    * The cache store can be refreshed every day at a specific time. This sets
    * the time. If the time is not set, cache is not refreshed.
    *
    * @param time time at which cache is refreshed everyday. Format is HH:mm:ss Z.
    */
   public void setRefreshTime(String time)
   {
     refreshTime = time;
   }

   public String getRefreshTime()
   {
     return refreshTime;
   }

   public int getNumInitCachedLines()
   {
     return numInitCachedLines;
   }

   /**
    * The cache store can be refreshed every day at a specific time. This sets
    * the time. If the time is not set, cache is not refreshed.
    *
    * @param numInitCachedLines number of lines to initially from
    */
   public void setNumInitCachedLines(int numInitCachedLines)
   {
     this.numInitCachedLines = numInitCachedLines;
   }

   public boolean getReadOnlyData()
   {
     return readOnlyData;
   }

   public void setReadOnlyData(boolean isReadOnly)
   {
     readOnlyData = isReadOnly;
   }

   /**
    * A primary store should also provide setting the value for a key.
    */
   public interface Primary extends KeyValueStore
   {

     /**
      * Get all the keys in the store.
      *
      * @return all present keys.
      */
     @JsonIgnore
     Set<Object> getKeys();
   }

   /**
    * Backup store is queried when {@link Primary} doesn't contain a key.<br/>
    * It also provides data needed at startup.<br/>
    */
   public interface Backup extends KeyValueStore
   {
     /**
      * <br>Backup stores are also used to initialize primary stores. This fetches initialization data.</br>
      *
      * @return map of key/value to initialize {@link CacheManager}
      */
     Map<Object, Object> loadInitialData();
   }


   /**
    * Used by {@link CacheManager} to pass properties to its stores which implement {@link Component}.
    */
   public static class CacheContext implements Context
   {
     public static final transient Attribute<Boolean> READ_ONLY_ATTR = new Attribute<>((false));
     public static final transient Attribute<Integer> NUM_INIT_CACHED_LINES_ATTR = new Attribute<>((-1));


     static {
       Attribute.AttributeMap.AttributeInitializer.initialize(CacheContext.class);
     }

     Attribute.AttributeMap attributeMap;

     public CacheContext(Attribute.AttributeMap attributeMap)
     {
       this.attributeMap = (Attribute.AttributeMap)(attributeMap == null ?
           new Attribute.AttributeMap.DefaultAttributeMap() : attributeMap);
     }

     @Override
     public Attribute.AttributeMap getAttributes()
     {
       return attributeMap;
     }

     @Override
     public <T> T getValue(Attribute<T> key)
     {
       T attr = attributeMap.get(key);
       return attr != null ? attr : key.defaultValue;
     }

     @Override
     public void setCounters(Object counters)
     {
       throw new UnsupportedOperationException();
     }

     @Override
     public void sendMetrics(Collection<String> metricNames)
     {
       throw new UnsupportedOperationException();
     }
   }

   @SuppressWarnings("unused")
   private static final Logger LOG = LoggerFactory.getLogger(CacheManager.class);

 }
	/**
	* 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 com.datatorrent.lib.db.cache;

	import java.io.Closeable;
	import java.io.IOException;
	import java.util.Calendar;
	import java.util.Collection;
	import java.util.List;
	import java.util.Map;
	import java.util.Set;
	import java.util.Timer;
	import java.util.TimerTask;

	import javax.annotation.Nonnull;
	import javax.annotation.Nullable;
	import javax.validation.constraints.NotNull;

	import org.codehaus.jackson.annotate.JsonIgnore;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import com.google.common.base.Strings;
	import com.google.common.collect.Lists;

	import com.datatorrent.api.Attribute;
	import com.datatorrent.api.Component;
	import com.datatorrent.api.Context;
	import com.datatorrent.lib.db.KeyValueStore;

	/**
	* Manages primary and secondary stores.<br/>
	* It firsts checks the primary store for a key. If the primary store doesn't have the key, it queries the backup
	* store and retrieves the value.<br/>
	* If the key was present in the backup store, its value is returned and also saved in the primary store.
	* <p/>
	* Typically primary store is faster but has limited size like memory and backup store is slower but unlimited like
	* databases.<br/>
	* Store Manager can also refresh the values of keys at a specified time every day.
	* This time is in format HH:mm:ss Z.<br/>
	* This is not thread-safe.
	*
	* @since 0.9.2
	*/
	public class CacheManager implements Closeable
	{
	@NotNull
	protected transient Primary primary;
	@NotNull
	protected transient Backup backup;
	protected String refreshTime;
	private transient Timer refresher;

	private transient Attribute.AttributeMap attributeMap = new Attribute.AttributeMap.DefaultAttributeMap();
	private boolean readOnlyData = false;
	private int numInitCachedLines = -1;

	public CacheManager()
	{
	this.primary = new CacheStore();
	}

	public void initialize() throws IOException
	{
	Class[] parameters = new Class[1];
	parameters[0] = CacheContext.class;

	attributeMap.put(CacheContext.READ_ONLY_ATTR, readOnlyData);
	attributeMap.put(CacheContext.NUM_INIT_CACHED_LINES_ATTR, numInitCachedLines);
	CacheContext cacheContext = new CacheContext(attributeMap);

	if (primary instanceof Component) {
	try {
	//check Component for method of type setup(CacheManager.CacheContext context)
	primary.getClass().getMethod("setup", parameters);
	((Component)primary).setup(cacheContext);
	} catch (NoSuchMethodException e) {
	//Store implements Component, but not Component<CacheManager.CacheContext>
	}
	}

	if (backup instanceof Component) {
	try {
	//check Component for method of type setup(CacheManager.CacheContext context)
	backup.getClass().getMethod("setup", parameters);
	((Component)backup).setup(cacheContext);
	} catch (NoSuchMethodException e) {
	//Store implements Component, but not Component<CacheManager.CacheContext>
	}
	}

	primary.connect();
	backup.connect();
	Map<Object, Object> initialEntries = backup.loadInitialData();
	if (initialEntries != null) {
	primary.putAll(initialEntries);
	}

	if (!readOnlyData && !Strings.isNullOrEmpty(refreshTime)) {

	String[] parts = refreshTime.split("[:\\s]");

	Calendar timeToRefresh = Calendar.getInstance();
	timeToRefresh.set(Calendar.HOUR_OF_DAY, Integer.parseInt(parts[0]));
	if (parts.length >= 2) {
	timeToRefresh.set(Calendar.MINUTE, Integer.parseInt(parts[1]));
	}
	if (parts.length >= 3) {
	timeToRefresh.set(Calendar.SECOND, Integer.parseInt(parts[2]));
	}
	long initialDelay = timeToRefresh.getTimeInMillis() - Calendar.getInstance().getTimeInMillis();

	TimerTask task = new TimerTask()
	{
	@Override
	public void run()
	{
	List<Object> keysToRefresh = Lists.newArrayList(primary.getKeys());
	if (keysToRefresh.size() > 0) {
	List<Object> refreshedValues = backup.getAll(keysToRefresh);
	if (refreshedValues != null) {
	for (int i = 0; i < keysToRefresh.size(); i++) {
	primary.put(keysToRefresh.get(i), refreshedValues.get(i));
	}
	}
	}
	}
	};

	refresher = new Timer();
	if (initialDelay < 0) {
	refresher.schedule(task, 0);
	timeToRefresh.add(Calendar.DAY_OF_MONTH, 1);
	initialDelay = timeToRefresh.getTimeInMillis();
	}
	refresher.scheduleAtFixedRate(task, initialDelay, 86400000);
	}
	}

	@Nullable
	public Object get(@Nonnull Object key)
	{
	Object primaryVal = primary.get(key);
	if (primaryVal != null) {
	return primaryVal;
	}

	Object backupVal = backup.get(key);
	if (backupVal != null) {
	primary.put(key, backupVal);
	}
	return backupVal;
	}

	public void put(@Nonnull Object key, @Nonnull Object value)
	{
	primary.put(key, value);
	backup.put(key, value);
	}

	@Override
	public void close() throws IOException
	{
	if (refresher != null) {
	refresher.cancel();
	}
	primary.disconnect();
	backup.disconnect();
	}

	public void setPrimary(Primary primary)
	{
	this.primary = primary;
	}

	@JsonIgnore
	public Primary getPrimary()
	{
	return primary;
	}

	public void setBackup(Backup backup)
	{
	this.backup = backup;
	}

	@JsonIgnore
	public Backup getBackup()
	{
	return backup;
	}

	/**
	* The cache store can be refreshed every day at a specific time. This sets
	* the time. If the time is not set, cache is not refreshed.
	*
	* @param time time at which cache is refreshed everyday. Format is HH:mm:ss Z.
	*/
	public void setRefreshTime(String time)
	{
	refreshTime = time;
	}

	public String getRefreshTime()
	{
	return refreshTime;
	}

	public int getNumInitCachedLines()
	{
	return numInitCachedLines;
	}

	/**
	* The cache store can be refreshed every day at a specific time. This sets
	* the time. If the time is not set, cache is not refreshed.
	*
	* @param numInitCachedLines number of lines to initially from
	*/
	public void setNumInitCachedLines(int numInitCachedLines)
	{
	this.numInitCachedLines = numInitCachedLines;
	}

	public boolean getReadOnlyData()
	{
	return readOnlyData;
	}

	public void setReadOnlyData(boolean isReadOnly)
	{
	readOnlyData = isReadOnly;
	}

	/**
	* A primary store should also provide setting the value for a key.
	*/
	public interface Primary extends KeyValueStore
	{

	/**
	* Get all the keys in the store.
	*
	* @return all present keys.
	*/
	@JsonIgnore
	Set<Object> getKeys();
	}

	/**
	* Backup store is queried when {@link Primary} doesn't contain a key.<br/>
	* It also provides data needed at startup.<br/>
	*/
	public interface Backup extends KeyValueStore
	{
	/**
	* <br>Backup stores are also used to initialize primary stores. This fetches initialization data.</br>
	*
	* @return map of key/value to initialize {@link CacheManager}
	*/
	Map<Object, Object> loadInitialData();
	}


	/**
	* Used by {@link CacheManager} to pass properties to its stores which implement {@link Component}.
	*/
	public static class CacheContext implements Context
	{
	public static final transient Attribute<Boolean> READ_ONLY_ATTR = new Attribute<>((false));
	public static final transient Attribute<Integer> NUM_INIT_CACHED_LINES_ATTR = new Attribute<>((-1));


	static {
	Attribute.AttributeMap.AttributeInitializer.initialize(CacheContext.class);
	}

	Attribute.AttributeMap attributeMap;

	public CacheContext(Attribute.AttributeMap attributeMap)
	{
	this.attributeMap = (Attribute.AttributeMap)(attributeMap == null ?
	new Attribute.AttributeMap.DefaultAttributeMap() : attributeMap);
	}

	@Override
	public Attribute.AttributeMap getAttributes()
	{
	return attributeMap;
	}

	@Override
	public <T> T getValue(Attribute<T> key)
	{
	T attr = attributeMap.get(key);
	return attr != null ? attr : key.defaultValue;
	}

	@Override
	public void setCounters(Object counters)
	{
	throw new UnsupportedOperationException();
	}

	@Override
	public void sendMetrics(Collection<String> metricNames)
	{
	throw new UnsupportedOperationException();
	}
	}

	@SuppressWarnings("unused")
	private static final Logger LOG = LoggerFactory.getLogger(CacheManager.class);

	}