commons-jcs-core/src/main/java/org/apache/commons/jcs3/engine/AbstractCacheEventQueue.java - commons-jcs - Git at Google

 package org.apache.commons.jcs3.engine;

 /*
  * 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.io.IOException;
 import java.util.concurrent.atomic.AtomicBoolean;

 import org.apache.commons.jcs3.engine.behavior.ICacheElement;
 import org.apache.commons.jcs3.engine.behavior.ICacheEventQueue;
 import org.apache.commons.jcs3.engine.behavior.ICacheListener;
 import org.apache.commons.jcs3.log.Log;
 import org.apache.commons.jcs3.log.LogManager;

 /**
  * An abstract base class to the different implementations
  */
 public abstract class AbstractCacheEventQueue<K, V>
     implements ICacheEventQueue<K, V>
 {
     /** The logger. */
     private static final Log log = LogManager.getLog( AbstractCacheEventQueue.class );

     /** default */
     protected static final int DEFAULT_WAIT_TO_DIE_MILLIS = 10000;

     /**
      * time to wait for an event before snuffing the background thread if the queue is empty. make
      * configurable later
      */
     private int waitToDieMillis = DEFAULT_WAIT_TO_DIE_MILLIS;

     /**
      * When the events are pulled off the queue, then tell the listener to handle the specific event
      * type. The work is done by the listener.
      */
     private ICacheListener<K, V> listener;

     /** Id of the listener registered with this queue */
     private long listenerId;

     /** The cache region name, if applicable. */
     private String cacheName;

     /** Maximum number of failures before we buy the farm. */
     private int maxFailure;

     /** in milliseconds */
     private int waitBeforeRetry;

     /**
      * This means that the queue is functional. If we reached the max number of failures, the queue
      * is marked as non functional and will never work again.
      */
     private final AtomicBoolean working = new AtomicBoolean(true);

     /**
      * Returns the time to wait for events before killing the background thread.
      * <p>
      * @return int
      */
     public int getWaitToDieMillis()
     {
         return waitToDieMillis;
     }

     /**
      * Sets the time to wait for events before killing the background thread.
      * <p>
      * @param wtdm the ms for the q to sit idle.
      */
     public void setWaitToDieMillis( final int wtdm )
     {
         waitToDieMillis = wtdm;
     }

     /**
      * Creates a brief string identifying the listener and the region.
      * <p>
      * @return String debugging info.
      */
     @Override
     public String toString()
     {
         return "CacheEventQueue [listenerId=" + listenerId + ", cacheName=" + cacheName + "]";
     }

     /**
      * @return The listenerId value
      */
     @Override
     public long getListenerId()
     {
         return listenerId;
     }

     /**
      * @return the cacheName
      */
     protected String getCacheName()
     {
         return cacheName;
     }

     /**
      * Initializes the queue.
      * <p>
      * @param listener
      * @param listenerId
      * @param cacheName
      * @param maxFailure
      * @param waitBeforeRetry
      */
     protected void initialize( final ICacheListener<K, V> listener, final long listenerId, final String cacheName, final int maxFailure,
                             final int waitBeforeRetry)
     {
         if ( listener == null )
         {
             throw new IllegalArgumentException( "listener must not be null" );
         }

         this.listener = listener;
         this.listenerId = listenerId;
         this.cacheName = cacheName;
         this.maxFailure = maxFailure <= 0 ? 3 : maxFailure;
         this.waitBeforeRetry = waitBeforeRetry <= 0 ? 500 : waitBeforeRetry;

         log.debug( "Constructed: {0}", this );
     }

     /**
      * This adds a put event to the queue. When it is processed, the element will be put to the
      * listener.
      * <p>
      * @param ce The feature to be added to the PutEvent attribute
      * @throws IOException
      */
     @Override
     public void addPutEvent( final ICacheElement<K, V> ce )
     {
         put( new PutEvent( ce ) );
     }

     /**
      * This adds a remove event to the queue. When processed the listener's remove method will be
      * called for the key.
      * <p>
      * @param key The feature to be added to the RemoveEvent attribute
      * @throws IOException
      */
     @Override
     public void addRemoveEvent( final K key )
     {
         put( new RemoveEvent( key ) );
     }

     /**
      * This adds a remove all event to the queue. When it is processed, all elements will be removed
      * from the cache.
      */
     @Override
     public void addRemoveAllEvent()
     {
         put( new RemoveAllEvent() );
     }

     /**
      * This adds a dispose event to the queue. When it is processed, the cache is shut down
      */
     @Override
     public void addDisposeEvent()
     {
         put( new DisposeEvent() );
     }

     /**
      * Adds an event to the queue.
      * <p>
      * @param event
      */
     protected abstract void put( AbstractCacheEvent event );


     // /////////////////////////// Inner classes /////////////////////////////
     /**
      * Retries before declaring failure.
      * <p>
      * @author asmuts
      */
     protected abstract class AbstractCacheEvent implements Runnable
     {
         /** Number of failures encountered processing this event. */
         int failures = 0;

         /**
          * Main processing method for the AbstractCacheEvent object
          */
         @Override
         @SuppressWarnings("synthetic-access")
         public void run()
         {
             try
             {
                 doRun();
             }
             catch ( final IOException e )
             {
                 log.warn( e );
                 if ( ++failures >= maxFailure )
                 {
                     log.warn( "Error while running event from Queue: {0}. "
                             + "Dropping Event and marking Event Queue as "
                             + "non-functional.", this );
                     destroy();
                     return;
                 }
                 log.info( "Error while running event from Queue: {0}. "
                         + "Retrying...", this );
                 try
                 {
                     Thread.sleep( waitBeforeRetry );
                     run();
                 }
                 catch ( final InterruptedException ie )
                 {
                     log.warn( "Interrupted while sleeping for retry on event "
                             + "{0}.", this );
                     destroy();
                 }
             }
         }

         /**
          * @throws IOException
          */
         protected abstract void doRun()
             throws IOException;
     }

     /**
      * An element should be put in the cache.
      * <p>
      * @author asmuts
      */
     protected class PutEvent
         extends AbstractCacheEvent
     {
         /** The element to put to the listener */
         private final ICacheElement<K, V> ice;

         /**
          * Constructor for the PutEvent object.
          * <p>
          * @param ice
          */
         PutEvent( final ICacheElement<K, V> ice )
         {
             this.ice = ice;
         }

         /**
          * Call put on the listener.
          * <p>
          * @throws IOException
          */
         @Override
         protected void doRun()
             throws IOException
         {
             listener.handlePut( ice );
         }

         /**
          * For debugging.
          * <p>
          * @return Info on the key and value.
          */
         @Override
         public String toString()
         {
             return new StringBuilder( "PutEvent for key: " )
                     .append( ice.getKey() )
                     .append( " value: " )
                     .append( ice.getVal() )
                     .toString();
         }

     }

     /**
      * An element should be removed from the cache.
      * <p>
      * @author asmuts
      */
     protected class RemoveEvent
         extends AbstractCacheEvent
     {
         /** The key to remove from the listener */
         private final K key;

         /**
          * Constructor for the RemoveEvent object
          * <p>
          * @param key
          */
         RemoveEvent( final K key )
         {
             this.key = key;
         }

         /**
          * Call remove on the listener.
          * <p>
          * @throws IOException
          */
         @Override
         protected void doRun()
             throws IOException
         {
             listener.handleRemove( cacheName, key );
         }

         /**
          * For debugging.
          * <p>
          * @return Info on the key to remove.
          */
         @Override
         public String toString()
         {
             return new StringBuilder( "RemoveEvent for " )
                     .append( key )
                     .toString();
         }

     }

     /**
      * All elements should be removed from the cache when this event is processed.
      * <p>
      * @author asmuts
      */
     protected class RemoveAllEvent
         extends AbstractCacheEvent
     {
         /**
          * Call removeAll on the listener.
          * <p>
          * @throws IOException
          */
         @Override
         protected void doRun()
             throws IOException
         {
             listener.handleRemoveAll( cacheName );
         }

         /**
          * For debugging.
          * <p>
          * @return The name of the event.
          */
         @Override
         public String toString()
         {
             return "RemoveAllEvent";
         }
     }

     /**
      * The cache should be disposed when this event is processed.
      * <p>
      * @author asmuts
      */
     protected class DisposeEvent
         extends AbstractCacheEvent
     {
         /**
          * Called when gets to the end of the queue
          * <p>
          * @throws IOException
          */
         @Override
         protected void doRun()
             throws IOException
         {
             listener.handleDispose( cacheName );
         }

         /**
          * For debugging.
          * <p>
          * @return The name of the event.
          */
         @Override
         public String toString()
         {
             return "DisposeEvent";
         }
     }

     /**
      * @return whether the queue is functional.
      */
     @Override
     public boolean isWorking()
     {
         return working.get();
     }

     /**
      * This means that the queue is functional. If we reached the max number of failures, the queue
      * is marked as non functional and will never work again.
      * <p>
      * @param b
      */
     public void setWorking( final boolean b )
     {
         working.set(b);
     }
 }
	package org.apache.commons.jcs3.engine;

	/*
	* 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.io.IOException;
	import java.util.concurrent.atomic.AtomicBoolean;

	import org.apache.commons.jcs3.engine.behavior.ICacheElement;
	import org.apache.commons.jcs3.engine.behavior.ICacheEventQueue;
	import org.apache.commons.jcs3.engine.behavior.ICacheListener;
	import org.apache.commons.jcs3.log.Log;
	import org.apache.commons.jcs3.log.LogManager;

	/**
	* An abstract base class to the different implementations
	*/
	public abstract class AbstractCacheEventQueue<K, V>
	implements ICacheEventQueue<K, V>
	{
	/** The logger. */
	private static final Log log = LogManager.getLog( AbstractCacheEventQueue.class );

	/** default */
	protected static final int DEFAULT_WAIT_TO_DIE_MILLIS = 10000;

	/**
	* time to wait for an event before snuffing the background thread if the queue is empty. make
	* configurable later
	*/
	private int waitToDieMillis = DEFAULT_WAIT_TO_DIE_MILLIS;

	/**
	* When the events are pulled off the queue, then tell the listener to handle the specific event
	* type. The work is done by the listener.
	*/
	private ICacheListener<K, V> listener;

	/** Id of the listener registered with this queue */
	private long listenerId;

	/** The cache region name, if applicable. */
	private String cacheName;

	/** Maximum number of failures before we buy the farm. */
	private int maxFailure;

	/** in milliseconds */
	private int waitBeforeRetry;

	/**
	* This means that the queue is functional. If we reached the max number of failures, the queue
	* is marked as non functional and will never work again.
	*/
	private final AtomicBoolean working = new AtomicBoolean(true);

	/**
	* Returns the time to wait for events before killing the background thread.
	* <p>
	* @return int
	*/
	public int getWaitToDieMillis()
	{
	return waitToDieMillis;
	}

	/**
	* Sets the time to wait for events before killing the background thread.
	* <p>
	* @param wtdm the ms for the q to sit idle.
	*/
	public void setWaitToDieMillis( final int wtdm )
	{
	waitToDieMillis = wtdm;
	}

	/**
	* Creates a brief string identifying the listener and the region.
	* <p>
	* @return String debugging info.
	*/
	@Override
	public String toString()
	{
	return "CacheEventQueue [listenerId=" + listenerId + ", cacheName=" + cacheName + "]";
	}

	/**
	* @return The listenerId value
	*/
	@Override
	public long getListenerId()
	{
	return listenerId;
	}

	/**
	* @return the cacheName
	*/
	protected String getCacheName()
	{
	return cacheName;
	}

	/**
	* Initializes the queue.
	* <p>
	* @param listener
	* @param listenerId
	* @param cacheName
	* @param maxFailure
	* @param waitBeforeRetry
	*/
	protected void initialize( final ICacheListener<K, V> listener, final long listenerId, final String cacheName, final int maxFailure,
	final int waitBeforeRetry)
	{
	if ( listener == null )
	{
	throw new IllegalArgumentException( "listener must not be null" );
	}

	this.listener = listener;
	this.listenerId = listenerId;
	this.cacheName = cacheName;
	this.maxFailure = maxFailure <= 0 ? 3 : maxFailure;
	this.waitBeforeRetry = waitBeforeRetry <= 0 ? 500 : waitBeforeRetry;

	log.debug( "Constructed: {0}", this );
	}

	/**
	* This adds a put event to the queue. When it is processed, the element will be put to the
	* listener.
	* <p>
	* @param ce The feature to be added to the PutEvent attribute
	* @throws IOException
	*/
	@Override
	public void addPutEvent( final ICacheElement<K, V> ce )
	{
	put( new PutEvent( ce ) );
	}

	/**
	* This adds a remove event to the queue. When processed the listener's remove method will be
	* called for the key.
	* <p>
	* @param key The feature to be added to the RemoveEvent attribute
	* @throws IOException
	*/
	@Override
	public void addRemoveEvent( final K key )
	{
	put( new RemoveEvent( key ) );
	}

	/**
	* This adds a remove all event to the queue. When it is processed, all elements will be removed
	* from the cache.
	*/
	@Override
	public void addRemoveAllEvent()
	{
	put( new RemoveAllEvent() );
	}

	/**
	* This adds a dispose event to the queue. When it is processed, the cache is shut down
	*/
	@Override
	public void addDisposeEvent()
	{
	put( new DisposeEvent() );
	}

	/**
	* Adds an event to the queue.
	* <p>
	* @param event
	*/
	protected abstract void put( AbstractCacheEvent event );


	// /////////////////////////// Inner classes /////////////////////////////
	/**
	* Retries before declaring failure.
	* <p>
	* @author asmuts
	*/
	protected abstract class AbstractCacheEvent implements Runnable
	{
	/** Number of failures encountered processing this event. */
	int failures = 0;

	/**
	* Main processing method for the AbstractCacheEvent object
	*/
	@Override
	@SuppressWarnings("synthetic-access")
	public void run()
	{
	try
	{
	doRun();
	}
	catch ( final IOException e )
	{
	log.warn( e );
	if ( ++failures >= maxFailure )
	{
	log.warn( "Error while running event from Queue: {0}. "
	+ "Dropping Event and marking Event Queue as "
	+ "non-functional.", this );
	destroy();
	return;
	}
	log.info( "Error while running event from Queue: {0}. "
	+ "Retrying...", this );
	try
	{
	Thread.sleep( waitBeforeRetry );
	run();
	}
	catch ( final InterruptedException ie )
	{
	log.warn( "Interrupted while sleeping for retry on event "
	+ "{0}.", this );
	destroy();
	}
	}
	}

	/**
	* @throws IOException
	*/
	protected abstract void doRun()
	throws IOException;
	}

	/**
	* An element should be put in the cache.
	* <p>
	* @author asmuts
	*/
	protected class PutEvent
	extends AbstractCacheEvent
	{
	/** The element to put to the listener */
	private final ICacheElement<K, V> ice;

	/**
	* Constructor for the PutEvent object.
	* <p>
	* @param ice
	*/
	PutEvent( final ICacheElement<K, V> ice )
	{
	this.ice = ice;
	}

	/**
	* Call put on the listener.
	* <p>
	* @throws IOException
	*/
	@Override
	protected void doRun()
	throws IOException
	{
	listener.handlePut( ice );
	}

	/**
	* For debugging.
	* <p>
	* @return Info on the key and value.
	*/
	@Override
	public String toString()
	{
	return new StringBuilder( "PutEvent for key: " )
	.append( ice.getKey() )
	.append( " value: " )
	.append( ice.getVal() )
	.toString();
	}

	}

	/**
	* An element should be removed from the cache.
	* <p>
	* @author asmuts
	*/
	protected class RemoveEvent
	extends AbstractCacheEvent
	{
	/** The key to remove from the listener */
	private final K key;

	/**
	* Constructor for the RemoveEvent object
	* <p>
	* @param key
	*/
	RemoveEvent( final K key )
	{
	this.key = key;
	}

	/**
	* Call remove on the listener.
	* <p>
	* @throws IOException
	*/
	@Override
	protected void doRun()
	throws IOException
	{
	listener.handleRemove( cacheName, key );
	}

	/**
	* For debugging.
	* <p>
	* @return Info on the key to remove.
	*/
	@Override
	public String toString()
	{
	return new StringBuilder( "RemoveEvent for " )
	.append( key )
	.toString();
	}

	}

	/**
	* All elements should be removed from the cache when this event is processed.
	* <p>
	* @author asmuts
	*/
	protected class RemoveAllEvent
	extends AbstractCacheEvent
	{
	/**
	* Call removeAll on the listener.
	* <p>
	* @throws IOException
	*/
	@Override
	protected void doRun()
	throws IOException
	{
	listener.handleRemoveAll( cacheName );
	}

	/**
	* For debugging.
	* <p>
	* @return The name of the event.
	*/
	@Override
	public String toString()
	{
	return "RemoveAllEvent";
	}
	}

	/**
	* The cache should be disposed when this event is processed.
	* <p>
	* @author asmuts
	*/
	protected class DisposeEvent
	extends AbstractCacheEvent
	{
	/**
	* Called when gets to the end of the queue
	* <p>
	* @throws IOException
	*/
	@Override
	protected void doRun()
	throws IOException
	{
	listener.handleDispose( cacheName );
	}

	/**
	* For debugging.
	* <p>
	* @return The name of the event.
	*/
	@Override
	public String toString()
	{
	return "DisposeEvent";
	}
	}

	/**
	* @return whether the queue is functional.
	*/
	@Override
	public boolean isWorking()
	{
	return working.get();
	}

	/**
	* This means that the queue is functional. If we reached the max number of failures, the queue
	* is marked as non functional and will never work again.
	* <p>
	* @param b
	*/
	public void setWorking( final boolean b )
	{
	working.set(b);
	}
	}