Author: jason.greene(a)jboss.com Date: 2008-10-15 11:44:27 -0400 (Wed, 15 Oct 2008) New Revision: 6957 Added: core/branches/flat/src/main/java/org/jboss/starobrno/loader/AbstractCacheLoader.java core/branches/flat/src/main/java/org/jboss/starobrno/loader/Modification.java Log: missed files Added: core/branches/flat/src/main/java/org/jboss/starobrno/loader/AbstractCacheLoader.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/loader/AbstractCacheLoader.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/loader/AbstractCacheLoader.java 2008-10-15 15:44:27 UTC (rev 6957) @@ -0,0 +1,192 @@ +/* + * JBoss, Home of Professional Open Source. + * Copyright 2000 - 2008, Red Hat Middleware LLC, and individual contributors + * as indicated by the @author tags. See the copyright.txt file in the + * distribution for a full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jboss.starobrno.loader; + +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.jboss.cache.RegionManager; +import org.jboss.cache.buddyreplication.BuddyFqnTransformer; +import org.jboss.cache.marshall.Marshaller; +import org.jboss.starobrno.CacheException; +import org.jboss.starobrno.CacheSPI; +import org.jboss.starobrno.marshall.EntryData; +import org.jboss.starobrno.marshall.EntryDataExceptionMarker; +import org.jboss.starobrno.marshall.EntryDataMarker; + +/** + * A convenience abstract implementation of a {@link CacheLoader}. Specific methods to note are methods like + * {@link #storeState(org.jboss.cache.Fqn,java.io.ObjectInputStream)}, {@link #loadState(org.jboss.cache.Fqn,java.io.ObjectOutputStream)}, + * {@link #storeEntireState(java.io.ObjectInputStream)} and {@link #loadEntireState(java.io.ObjectOutputStream)} which have convenience + * implementations here. + * <p/> + * Also useful to note is the implementation of {@link #put(java.util.List)}, used during the prepare phase of a transaction. + * <p/> + * + * @author <a href="mailto:manik@jboss.org">Manik Surtani (manik(a)jboss.org)&lt;/a&gt; + * @author <a href="mailto:galder.zamarreno@jboss.com">Galder Zamarreno</a> + * @since 2.0.0 + */ +public abstract class AbstractCacheLoader<K, V> implements CacheLoader<K,V> +{ + protected CacheSPI<K,V> cache; + protected RegionManager regionManager; + private static final Log log = LogFactory.getLog(AbstractCacheLoader.class); + private static final boolean trace = log.isTraceEnabled(); + protected BuddyFqnTransformer buddyFqnTransformer = new BuddyFqnTransformer(); + /** + * HashMap<Object,List<Modification>>. List of open transactions. Note that this is purely transient, as + * we don't use a log, recovery is not available + */ + protected Map<Object, List<Modification>> transactions = new ConcurrentHashMap<Object, List<Modification>>(); + + public void storeEntireState(ObjectInputStream in) + { + // store new state + Object objectFromStream; + try + { + objectFromStream = cache.getMarshaller().objectFromObjectStream(in); + } + catch (Exception e) + { + throw new CacheException(e.getMessage(), e); + } + + if (objectFromStream instanceof EntryDataMarker) + { + // no persistent state sent across; return? + if (trace) log.trace("Empty persistent stream?"); + return; + } + if (objectFromStream instanceof EntryDataExceptionMarker) + { + EntryDataExceptionMarker ndem = (EntryDataExceptionMarker) objectFromStream; + throw new CacheException("State provider cacheloader at node " + ndem.getCacheNodeIdentity() + + " threw exception during loadState (see Caused by)", ndem.getCause()); + } + + List<EntryData<K, V>> data = (List<EntryData<K, V>>) objectFromStream; + for (EntryData<K, V> datem : data) + { + put(datem.getKey(), datem.getValue()); + } + } + + public void loadEntireState(ObjectOutputStream os) + { + List<EntryData<K,V>> list = getAllEntries(); + if (trace) log.trace("Loading state of " + list.size() + " nodes into stream"); + try + { + cache.getMarshaller().objectToObjectStream(list, os); + } + catch (Exception e) + { + throw new CacheException(e.getMessage(), e); + } + } + + + public void setCache(CacheSPI<K,V> c) + { + this.cache = c; + } + + public void put(List<Modification> modifications) + { + for (Modification m : modifications) + { + switch (m.getType()) + { + case PUT: + put(m.getKey(), (V)m.getValue()); + break; + case REMOVE: + remove(m.getKey()); + break; + case CLEAR: + clear(); + break; + default: + throw new CacheException("Unknown modification " + m.getType()); + } + } + } + + protected Marshaller getMarshaller() + { + return cache.getMarshaller(); + } + + // empty implementations for loaders that do not wish to implement lifecycle. + public void create() + { + } + + public void start() + { + } + + public void stop() + { + } + + public void destroy() + { + } + + // Adds simple transactional capabilities to cache loaders that are inherently non-transactional. If your cache loader implementation + // is tansactional though, then override these. + + public void prepare(Object tx, List<Modification> modifications, boolean one_phase) + { + if (one_phase) + { + put(modifications); + } + else + { + transactions.put(tx, modifications); + } + } + + public void commit(Object tx) + { + List<Modification> modifications = transactions.remove(tx); + if (modifications == null) + { + throw new CacheException("transaction " + tx + " not found in transaction table"); + } + put(modifications); + } + + public void rollback(Object tx) + { + transactions.remove(tx); + } +} Added: core/branches/flat/src/main/java/org/jboss/starobrno/loader/Modification.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/loader/Modification.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/loader/Modification.java 2008-10-15 15:44:27 UTC (rev 6957) @@ -0,0 +1,112 @@ +/* + * JBoss, Home of Professional Open Source. + * Copyright 2000 - 2008, Red Hat Middleware LLC, and individual contributors + * as indicated by the @author tags. See the copyright.txt file in the + * distribution for a full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jboss.starobrno.loader; + + +import java.io.Serializable; + + +/** + * Represents a modification in the cache. Contains the nature of the modification + * (e.g. PUT, REMOVE), the fqn of the node, the new value and the previous value. + * A list of modifications will be sent to all nodes in a cluster when a transaction + * has been committed (PREPARE phase). A Modification is also used to roll back changes, + * e.g. since we know the previous value, we can reconstruct the previous state by + * applying the changes in a modification listin reverse order. + * + * @author <a href="mailto:bela@jboss.org">Bela Ban</a> Apr 12, 2003 + * @version $Revision: 6776 $ + */ +public final class Modification implements Serializable +{ + + private static final long serialVersionUID = 7463314130283897197L; + + public static enum ModificationType + { + PUT, + REMOVE, + CLEAR + } + + private final ModificationType type; + private final Object key; + private final Object value; + + /** + * Constructs a new modification with details. + */ + public Modification(ModificationType type, Object key, Object value) + { + if (type == null) + throw new IllegalArgumentException(); + this.type = type; + this.key = key; + this.value = value; + } + + + /** + * Returns the type of modification. + */ + public ModificationType getType() + { + return type; + } + + /** + * Returns the modification key. + */ + public Object getKey() + { + return key; + } + + /** + * Returns the modification value. + */ + public Object getValue() + { + return value; + } + + + /** + * Returns debug information about this modification. + */ + @Override + public String toString() + { + StringBuilder sb = new StringBuilder(); + sb.append(type.toString()).append("{"); + if (key != null) + { + sb.append("\nkey=").append(key); + } + if (value != null) + { + sb.append("\nvalue=").append(value); + } + return sb.append("}").toString(); + } + +}