Author: manik.surtani(a)jboss.com Date: 2008-10-09 09:45:15 -0400 (Thu, 09 Oct 2008) New Revision: 6898 Added: core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/CacheMgmtInterceptor.java core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/MarshalledValueInterceptor.java core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/NotificationInterceptor.java core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/base/JmxStatsCommandInterceptor.java core/branches/flat/src/main/java/org/jboss/starobrno/marshall/ core/branches/flat/src/main/java/org/jboss/starobrno/marshall/MarshalledValue.java core/branches/flat/src/main/java/org/jboss/starobrno/marshall/MarshalledValueHelper.java core/branches/flat/src/main/java/org/jboss/starobrno/marshall/MarshalledValueMap.java Removed: core/branches/flat/src/main/java/org/jboss/cache/interceptors/CallInterceptor.java core/branches/flat/src/main/java/org/jboss/cache/interceptors/MVCCLockingInterceptor.java core/branches/flat/src/main/java/org/jboss/cache/interceptors/MarshalledValueInterceptor.java core/branches/flat/src/main/java/org/jboss/cache/interceptors/NotificationInterceptor.java core/branches/flat/src/main/java/org/jboss/cache/interceptors/TxInterceptor.java Modified: core/branches/flat/src/main/java/org/jboss/cache/UnversionedNode.java core/branches/flat/src/main/java/org/jboss/cache/marshall/CacheMarshaller200.java core/branches/flat/src/main/java/org/jboss/cache/notifications/NotifierImpl.java core/branches/flat/src/main/java/org/jboss/starobrno/commands/write/PutKeyValueCommand.java core/branches/flat/src/main/java/org/jboss/starobrno/commands/write/PutMapCommand.java core/branches/flat/src/main/java/org/jboss/starobrno/factories/InterceptorChainFactory.java core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/BatchingInterceptor.java core/branches/flat/src/main/java/org/jboss/starobrno/notifications/NotifierImpl.java Log: More interceptors migrated Modified: core/branches/flat/src/main/java/org/jboss/cache/UnversionedNode.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/cache/UnversionedNode.java 2008-10-09 13:22:09 UTC (rev 6897) +++ core/branches/flat/src/main/java/org/jboss/cache/UnversionedNode.java 2008-10-09 13:45:15 UTC (rev 6898) @@ -24,11 +24,11 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import static org.jboss.cache.AbstractNode.NodeFlags.*; -import org.jboss.cache.marshall.MarshalledValue; import org.jboss.cache.util.FastCopyHashMap; import org.jboss.cache.util.Immutables; import org.jboss.cache.util.concurrent.SelfInitializingConcurrentHashMap; import org.jboss.starobrno.CacheException; +import org.jboss.starobrno.marshall.MarshalledValue; import java.util.Collections; import java.util.HashMap; Deleted: core/branches/flat/src/main/java/org/jboss/cache/interceptors/CallInterceptor.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/cache/interceptors/CallInterceptor.java 2008-10-09 13:22:09 UTC (rev 6897) +++ core/branches/flat/src/main/java/org/jboss/cache/interceptors/CallInterceptor.java 2008-10-09 13:45:15 UTC (rev 6898) @@ -1,182 +0,0 @@ -/* - * 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.cache.interceptors; - -import org.jboss.cache.InvocationContext; -import org.jboss.cache.commands.ReplicableCommand; -import org.jboss.cache.commands.VisitableCommand; -import org.jboss.cache.commands.WriteCommand; -import org.jboss.cache.commands.tx.CommitCommand; -import org.jboss.cache.commands.tx.PrepareCommand; -import org.jboss.cache.commands.tx.RollbackCommand; -import org.jboss.cache.commands.write.ClearDataCommand; -import org.jboss.cache.commands.write.MoveCommand; -import org.jboss.cache.commands.write.PutDataMapCommand; -import org.jboss.cache.commands.write.PutForExternalReadCommand; -import org.jboss.cache.commands.write.PutKeyValueCommand; -import org.jboss.cache.commands.write.RemoveKeyCommand; -import org.jboss.cache.commands.write.RemoveNodeCommand; -import org.jboss.cache.interceptors.base.CommandInterceptor; -import org.jboss.cache.transaction.GlobalTransaction; -import org.jboss.starobrno.factories.annotations.Start; - -import javax.transaction.Transaction; - -/** - * Always at the end of the chain, directly in front of the cache. Simply calls into the cache using reflection. - * If the call resulted in a modification, add the Modification to the end of the modification list - * keyed by the current transaction. - * <p/> - * Although always added to the end of an optimistically locked chain as well, calls should not make it down to - * this interceptor unless it is a call the OptimisticNodeInterceptor knows nothing about. - * - * @author Bela Ban - * @version $Id$ - */ -public class CallInterceptor extends CommandInterceptor -{ - private boolean notOptimisticLocking; - - @Start - protected void start() - { - notOptimisticLocking = true; - } - - @Override - public Object visitPrepareCommand(InvocationContext ctx, PrepareCommand command) throws Throwable - { - if (trace) log.trace("Suppressing invocation of method handlePrepareCommand."); - return null; - } - - @Override - public Object visitCommitCommand(InvocationContext ctx, CommitCommand command) throws Throwable - { - if (trace) log.trace("Suppressing invocation of method handleCommitCommand."); - return null; - } - - @Override - public Object visitRollbackCommand(InvocationContext ctx, RollbackCommand command) throws Throwable - { - if (trace) log.trace("Suppressing invocation of method handleRollbackCommand."); - return null; - } - - @Override - public Object handleDefault(InvocationContext ctx, VisitableCommand command) throws Throwable - { - if (trace) log.trace("Executing command: " + command + "."); - return invokeCommand(ctx, command); - } - - private Object invokeCommand(InvocationContext ctx, ReplicableCommand command) - throws Throwable - { - Object retval; - try - { - retval = command.perform(ctx); - } - catch (Throwable t) - { - Transaction tx = ctx.getTransaction(); - if (ctx.isValidTransaction()) - { - tx.setRollbackOnly(); - } - throw t; - } - return retval; - } - - @Override - public Object visitPutDataMapCommand(InvocationContext ctx, PutDataMapCommand command) throws Throwable - { - return handleAlterCacheMethod(ctx, command); - } - - @Override - public Object visitPutKeyValueCommand(InvocationContext ctx, PutKeyValueCommand command) throws Throwable - { - return handleAlterCacheMethod(ctx, command); - } - - @Override - public Object visitPutForExternalReadCommand(InvocationContext ctx, PutForExternalReadCommand command) throws Throwable - { - return handleAlterCacheMethod(ctx, command); - } - - @Override - public Object visitRemoveNodeCommand(InvocationContext ctx, RemoveNodeCommand command) throws Throwable - { - return handleAlterCacheMethod(ctx, command); - } - - @Override - public Object visitClearDataCommand(InvocationContext ctx, ClearDataCommand command) throws Throwable - { - return handleAlterCacheMethod(ctx, command); - } - - @Override - public Object visitRemoveKeyCommand(InvocationContext ctx, RemoveKeyCommand command) throws Throwable - { - return handleAlterCacheMethod(ctx, command); - } - - @Override - public Object visitMoveCommand(InvocationContext ctx, MoveCommand command) throws Throwable - { - return handleAlterCacheMethod(ctx, command); - } - - /** - * only add the modification to the modification list if we are using pessimistic locking. - * Optimistic locking calls *should* not make it this far down the interceptor chain, but just - * in case a method has been invoked that the OptimisticNodeInterceptor knows nothing about, it will - * filter down here. - */ - private Object handleAlterCacheMethod(InvocationContext ctx, WriteCommand command) - throws Throwable - { - Object result = invokeCommand(ctx, command); - if (notOptimisticLocking && ctx.isValidTransaction()) - { - GlobalTransaction gtx = ctx.getGlobalTransaction(); - if (gtx == null) - { - if (log.isDebugEnabled()) - { - log.debug("didn't find GlobalTransaction for " + ctx.getTransaction() + "; won't add modification to transaction list"); - } - } - else - { - ctx.getTransactionContext().addModification(command); - } - } - return result; - } -} Deleted: core/branches/flat/src/main/java/org/jboss/cache/interceptors/MVCCLockingInterceptor.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/cache/interceptors/MVCCLockingInterceptor.java 2008-10-09 13:22:09 UTC (rev 6897) +++ core/branches/flat/src/main/java/org/jboss/cache/interceptors/MVCCLockingInterceptor.java 2008-10-09 13:45:15 UTC (rev 6898) @@ -1,402 +0,0 @@ -/* - * 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.cache.interceptors; - -import org.jboss.cache.DataContainer; -import org.jboss.cache.Fqn; -import org.jboss.cache.InternalNode; -import org.jboss.cache.InvocationContext; -import org.jboss.cache.NodeSPI; -import org.jboss.cache.commands.VisitableCommand; -import org.jboss.cache.commands.read.ExistsCommand; -import org.jboss.cache.commands.read.GetChildrenNamesCommand; -import org.jboss.cache.commands.read.GetDataMapCommand; -import org.jboss.cache.commands.read.GetKeyValueCommand; -import org.jboss.cache.commands.read.GetKeysCommand; -import org.jboss.cache.commands.read.GetNodeCommand; -import org.jboss.cache.commands.read.GravitateDataCommand; -import org.jboss.cache.commands.tx.CommitCommand; -import org.jboss.cache.commands.tx.PrepareCommand; -import org.jboss.cache.commands.tx.RollbackCommand; -import org.jboss.cache.commands.write.ClearDataCommand; -import org.jboss.cache.commands.write.EvictCommand; -import org.jboss.cache.commands.write.InvalidateCommand; -import org.jboss.cache.commands.write.MoveCommand; -import org.jboss.cache.commands.write.PutDataMapCommand; -import org.jboss.cache.commands.write.PutForExternalReadCommand; -import org.jboss.cache.commands.write.PutKeyValueCommand; -import org.jboss.cache.commands.write.RemoveKeyCommand; -import org.jboss.cache.commands.write.RemoveNodeCommand; -import org.jboss.cache.interceptors.base.PrePostProcessingCommandInterceptor; -import org.jboss.cache.lock.LockManager; -import org.jboss.cache.mvcc.MVCCNodeHelper; -import org.jboss.starobrno.factories.annotations.Inject; - -import java.util.Collections; -import java.util.LinkedList; -import java.util.List; -import java.util.ListIterator; -import java.util.Map; - -/** - * Interceptor to implement <a href="http://wiki.jboss.org/wiki/JBossCacheMVCC">MVCC</a> functionality. - * - * @author Manik Surtani (<a href="mailto:manik@jboss.org">manik@jboss.org</a>) - * @see <a href="http://wiki.jboss.org/wiki/JBossCacheMVCC">MVCC designs</a> - * @since 3.0 - */ -public class MVCCLockingInterceptor extends PrePostProcessingCommandInterceptor -{ - LockManager lockManager; - DataContainer dataContainer; - MVCCNodeHelper helper; - - @Inject - public void setDependencies(LockManager lockManager, DataContainer dataContainer, MVCCNodeHelper helper) - { - this.lockManager = lockManager; - this.dataContainer = dataContainer; - this.helper = helper; - } - - @Override - protected boolean doBeforeCall(InvocationContext ctx, VisitableCommand command) - { - if (ctx.getOptionOverrides().isSuppressLocking()) - { - if (log.isWarnEnabled()) log.warn("Lock suppression not supported with MVCC!"); - } - return true; - } - - @Override - public Object handlePutDataMapCommand(InvocationContext ctx, PutDataMapCommand command) throws Throwable - { - helper.wrapNodeForWriting(ctx, command.getFqn(), true, true, false, false, false); // get the node and stick it in the context. - return invokeNextInterceptor(ctx, command); - } - - @Override - public Object handlePutKeyValueCommand(InvocationContext ctx, PutKeyValueCommand command) throws Throwable - { - helper.wrapNodeForWriting(ctx, command.getFqn(), true, true, false, false, false); // get the node and stick it in the context. - return invokeNextInterceptor(ctx, command); - } - - @Override - public Object handlePutForExternalReadCommand(InvocationContext ctx, PutForExternalReadCommand command) throws Throwable - { - helper.wrapNodeForWriting(ctx, command.getFqn(), true, true, false, false, false); // get the node and stick it in the context. - return invokeNextInterceptor(ctx, command); - } - - @Override - public Object handleRemoveNodeCommand(InvocationContext ctx, RemoveNodeCommand command) throws Throwable - { - helper.wrapNodesRecursivelyForRemoval(ctx, command.getFqn()); - return invokeNextInterceptor(ctx, command); - } - - @Override - public Object handleClearDataCommand(InvocationContext ctx, ClearDataCommand command) throws Throwable - { - helper.wrapNodeForWriting(ctx, command.getFqn(), true, false, false, false, false); - return invokeNextInterceptor(ctx, command); - } - - @Override - public Object handleEvictFqnCommand(InvocationContext ctx, EvictCommand command) throws Throwable - { - // set lock acquisition timeout to 0 - we need to fail fast. - ctx.getOptionOverrides().setLockAcquisitionTimeout(0); - if (command.isRecursive()) - { - handleRecursiveEvict(ctx, command); - } - else - { - handleNonrecursiveEvict(ctx, command); - } - - return invokeNextInterceptor(ctx, command); - } - - @SuppressWarnings("unchecked") - private void handleRecursiveEvict(InvocationContext ctx, EvictCommand command) throws InterruptedException - { - List<Fqn> fqnsToEvict; - if (command.getFqn().isRoot()) - { - // if this is the root node, do not attempt to lock this for writing but instead just get all direct children of root. - Map<Object, InternalNode> children = dataContainer.peekInternalNode(Fqn.ROOT, false).getChildrenMap(); - if (!children.isEmpty()) - { - fqnsToEvict = new LinkedList<Fqn>(); - // lock recursively. - for (InternalNode child : children.values()) - fqnsToEvict.addAll(helper.wrapNodesRecursivelyForRemoval(ctx, child.getFqn())); - } - else - { - fqnsToEvict = Collections.emptyList(); - } - } - else - { - // lock current node recursively. - fqnsToEvict = helper.wrapNodesRecursivelyForRemoval(ctx, command.getFqn()); - } - - // set these in the evict command so that the command is aware of what needs to be evicted. - command.setNodesToEvict(fqnsToEvict); - } - - @SuppressWarnings("unchecked") - private void handleNonrecursiveEvict(InvocationContext ctx, EvictCommand command) throws InterruptedException - { - if (command.getFqn().isRoot()) - { - // if this is the root node, do not attempt to lock this for writing but instead just get all direct children of root. - Map<Object, InternalNode> children = dataContainer.peekInternalNode(Fqn.ROOT, false).getChildrenMap(); - if (!children.isEmpty()) - { - for (InternalNode child : children.values()) - helper.wrapNodeForWriting(ctx, child.getFqn(), true, false, false, true, true); - } - } - else - { - // just wrap the node for writing. Do not create if absent. - helper.wrapNodeForWriting(ctx, command.getFqn(), true, false, false, true, true); - } - } - - @Override - public Object handleInvalidateCommand(InvocationContext ctx, InvalidateCommand command) throws Throwable - { - // this should be handled the same as a recursive evict command. - ctx.getOptionOverrides().setLockAcquisitionTimeout(0); - if (!command.getFqn().isRoot()) helper.wrapNodesRecursivelyForRemoval(ctx, command.getFqn()); - return invokeNextInterceptor(ctx, command); - } - - @Override - public Object handleRemoveKeyCommand(InvocationContext ctx, RemoveKeyCommand command) throws Throwable - { - helper.wrapNodeForWriting(ctx, command.getFqn(), true, false, false, false, false); - return invokeNextInterceptor(ctx, command); - } - - @Override - public Object handleGetDataMapCommand(InvocationContext ctx, GetDataMapCommand command) throws Throwable - { - helper.wrapNodeForReading(ctx, command.getFqn(), true); - return invokeNextInterceptor(ctx, command); - } - - @Override - public Object handleExistsNodeCommand(InvocationContext ctx, ExistsCommand command) throws Throwable - { - helper.wrapNodeForReading(ctx, command.getFqn(), true); - return invokeNextInterceptor(ctx, command); - } - - @Override - public Object handleGetKeyValueCommand(InvocationContext ctx, GetKeyValueCommand command) throws Throwable - { - helper.wrapNodeForReading(ctx, command.getFqn(), true); - return invokeNextInterceptor(ctx, command); - } - - @Override - public Object handleGetNodeCommand(InvocationContext ctx, GetNodeCommand command) throws Throwable - { - helper.wrapNodeForReading(ctx, command.getFqn(), true); - return invokeNextInterceptor(ctx, command); - } - - @Override - public Object handleGetKeysCommand(InvocationContext ctx, GetKeysCommand command) throws Throwable - { - helper.wrapNodeForReading(ctx, command.getFqn(), true); - return invokeNextInterceptor(ctx, command); - } - - @Override - public Object handleGetChildrenNamesCommand(InvocationContext ctx, GetChildrenNamesCommand command) throws Throwable - { - helper.wrapNodeForReading(ctx, command.getFqn(), true); - return invokeNextInterceptor(ctx, command); - } - - @Override - public Object handleMoveCommand(InvocationContext ctx, MoveCommand command) throws Throwable - { - // Nodes we need to get WLs for: - - // 1) node we are moving FROM, and its parent and ALL children. Same as removeNode. - List<Fqn> nodeAndChildren = helper.wrapNodesRecursivelyForRemoval(ctx, command.getFqn()); - - Fqn newParent = command.getTo(); - Fqn oldParent = command.getFqn().getParent(); - - // 2) The new parent. - helper.wrapNodeForWriting(ctx, newParent, true, true, false, false, false); - - if (!oldParent.equals(newParent)) - { - // the nodeAndChildren list contains all child nodes, including the node itself. - // 3) now obtain locks on the new places these new nodes will occupy. - for (Fqn f : nodeAndChildren) - { - Fqn newChildFqn = f.replaceAncestor(oldParent, newParent); - helper.wrapNodeForWriting(ctx, newChildFqn, true, true, true, false, false); - } - } - - // now pass up the chain. - return invokeNextInterceptor(ctx, command); - } - - @Override - public Object handleGravitateDataCommand(InvocationContext ctx, GravitateDataCommand command) throws Throwable - { - helper.wrapNodeForReading(ctx, command.getFqn(), true); - return invokeNextInterceptor(ctx, command); - } - - @Override - public Object handleRollbackCommand(InvocationContext ctx, RollbackCommand command) throws Throwable - { - Object retval = null; - try - { - retval = invokeNextInterceptor(ctx, command); - } - finally - { - transactionalCleanup(false, ctx); - } - return retval; - } - - @Override - public Object handleCommitCommand(InvocationContext ctx, CommitCommand command) throws Throwable - { - Object retval = null; - try - { - retval = invokeNextInterceptor(ctx, command); - } - finally - { - transactionalCleanup(true, ctx); - } - return retval; - } - - @Override - public Object handlePrepareCommand(InvocationContext ctx, PrepareCommand command) throws Throwable - { - Object retval = null; - try - { - retval = invokeNextInterceptor(ctx, command); - } - finally - { - if (command.isOnePhaseCommit()) transactionalCleanup(true, ctx); - } - return retval; - } - - @SuppressWarnings("unchecked") - protected void doAfterCall(InvocationContext ctx, VisitableCommand command) - { - // for non-transactional stuff. - if (ctx.getTransactionContext() == null) - { - List<Fqn> locks; - if (!(locks = ctx.getLocks()).isEmpty()) - { - cleanupLocks(locks, ctx, Thread.currentThread(), true); - } - else - { - if (trace) log.trace("Nothing to do since there are no modifications in scope."); - } - } - else - { - if (trace) log.trace("Nothing to do since there is a transaction in scope."); - } - } - - private void cleanupLocks(List<Fqn> locks, InvocationContext ctx, Object owner, boolean commit) - { - // clean up. - // unlocking needs to be done in reverse order. - ListIterator<Fqn> it = locks.listIterator(locks.size()); - - if (commit) - { - while (it.hasPrevious()) - { - Fqn f = it.previous(); - NodeSPI rcn = ctx.lookUpNode(f); - // could be null with read-committed - if (rcn != null) rcn.commitUpdate(ctx, dataContainer); - // and then unlock - if (trace) log.trace("Releasing lock on [" + f + "] for owner " + owner); - lockManager.unlock(f, owner); - } - } - else - { - while (it.hasPrevious()) - { - Fqn f = it.previous(); - NodeSPI rcn = ctx.lookUpNode(f); - // could be null with read-committed - if (rcn != null) rcn.rollbackUpdate(); - // and then unlock - if (trace) log.trace("Releasing lock on [" + f + "] for owner " + owner); - lockManager.unlock(f, owner); - } - } - ctx.clearLocks(); - } - - @SuppressWarnings("unchecked") - private void transactionalCleanup(boolean commit, InvocationContext ctx) - { - if (ctx.getTransactionContext() != null) - { - List<Fqn> locks = ctx.getTransactionContext().getLocks(); - if (!locks.isEmpty()) cleanupLocks(locks, ctx, ctx.getGlobalTransaction(), commit); - } - else - { - throw new IllegalStateException("Attempting to do a commit or rollback but there is no transactional context in scope. " + ctx); - } - } -} Deleted: core/branches/flat/src/main/java/org/jboss/cache/interceptors/MarshalledValueInterceptor.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/cache/interceptors/MarshalledValueInterceptor.java 2008-10-09 13:22:09 UTC (rev 6897) +++ core/branches/flat/src/main/java/org/jboss/cache/interceptors/MarshalledValueInterceptor.java 2008-10-09 13:45:15 UTC (rev 6898) @@ -1,210 +0,0 @@ -/* - * 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.cache.interceptors; - -import org.jboss.cache.InvocationContext; -import org.jboss.cache.commands.ReplicableCommand; -import org.jboss.cache.commands.read.GetChildrenNamesCommand; -import org.jboss.cache.commands.read.GetDataMapCommand; -import org.jboss.cache.commands.read.GetKeyValueCommand; -import org.jboss.cache.commands.read.GetKeysCommand; -import org.jboss.cache.commands.read.GetNodeCommand; -import org.jboss.cache.commands.write.ClearDataCommand; -import org.jboss.cache.commands.write.PutDataMapCommand; -import org.jboss.cache.commands.write.PutForExternalReadCommand; -import org.jboss.cache.commands.write.PutKeyValueCommand; -import org.jboss.cache.commands.write.RemoveKeyCommand; -import org.jboss.cache.interceptors.base.CommandInterceptor; -import org.jboss.cache.marshall.MarshalledValue; -import org.jboss.cache.marshall.MarshalledValueHelper; -import org.jboss.cache.marshall.MarshalledValueMap; - -import java.io.IOException; -import java.io.NotSerializableException; -import java.util.Collections; -import java.util.HashMap; -import java.util.HashSet; -import java.util.Map; -import java.util.Set; - -/** - * Interceptor that handles the wrapping and unwrapping of cached data using {@link org.jboss.cache.marshall.MarshalledValue}s. - * Known "excluded" types are not wrapped/unwrapped, which at this time include {@link String}, Java primitives - * and their Object wrappers, as well as arrays of excluded types. - * <p/> - * The {@link org.jboss.cache.marshall.MarshalledValue} wrapper handles lazy deserialization from byte array representations. - * - * @author Manik Surtani (<a href="mailto:manik@jboss.org">manik@jboss.org</a>) - * @see org.jboss.cache.marshall.MarshalledValue - * @since 2.1.0 - */ -public class MarshalledValueInterceptor extends CommandInterceptor -{ - - @Override - public Object visitPutDataMapCommand(InvocationContext ctx, PutDataMapCommand command) throws Throwable - { - Set<MarshalledValue> marshalledValues = new HashSet<MarshalledValue>(); - command.setData(wrapMap(command.getData(), marshalledValues, ctx)); - Object retVal = invokeNextInterceptor(ctx, command); - return compactAndProcessRetVal(command, marshalledValues, retVal); - } - - @Override - public Object visitGetDataMapCommand(InvocationContext ctx, GetDataMapCommand command) throws Throwable - { - Object retVal = invokeNextInterceptor(ctx, command); - if (retVal instanceof Map) - { - if (trace) log.trace("Return value is a Map and we're retrieving data. Wrapping as a MarshalledValueMap."); - Map retValMap = (Map) retVal; - if (!retValMap.isEmpty()) retVal = new MarshalledValueMap(retValMap); - } - return retVal; - } - - @Override - public Object visitPutForExternalReadCommand(InvocationContext ctx, PutForExternalReadCommand command) throws Throwable - { - return visitPutKeyValueCommand(ctx, command); - } - - @Override - public Object visitPutKeyValueCommand(InvocationContext ctx, PutKeyValueCommand command) throws Throwable - { - Set<MarshalledValue> marshalledValues = new HashSet<MarshalledValue>(); - if (!MarshalledValueHelper.isTypeExcluded(command.getKey().getClass())) - { - Object newKey = createAndAddMarshalledValue(command.getKey(), marshalledValues, ctx); - command.setKey(newKey); - } - if (!MarshalledValueHelper.isTypeExcluded(command.getValue().getClass())) - { - Object value = createAndAddMarshalledValue(command.getValue(), marshalledValues, ctx); - command.setValue(value); - } - Object retVal = invokeNextInterceptor(ctx, command); - return compactAndProcessRetVal(command, marshalledValues, retVal); - } - - @Override - public Object visitGetNodeCommand(InvocationContext ctx, GetNodeCommand command) throws Throwable - { - Object retVal = invokeNextInterceptor(ctx, command); - return processRetVal(retVal); - } - - @Override - public Object visitClearDataCommand(InvocationContext ctx, ClearDataCommand command) throws Throwable - { - Object retVal = invokeNextInterceptor(ctx, command); - return processRetVal(retVal); - } - - @Override - public Object visitRemoveKeyCommand(InvocationContext ctx, RemoveKeyCommand command) throws Throwable - { - Set<MarshalledValue> marshalledValues = new HashSet<MarshalledValue>(); - if (!MarshalledValueHelper.isTypeExcluded(command.getKey().getClass())) - { - Object value = createAndAddMarshalledValue(command.getKey(), marshalledValues, ctx); - command.setKey(value); - } - Object retVal = invokeNextInterceptor(ctx, command); - return compactAndProcessRetVal(command, marshalledValues, retVal); - } - - @Override - public Object visitGetChildrenNamesCommand(InvocationContext ctx, GetChildrenNamesCommand command) throws Throwable - { - Object retVal = invokeNextInterceptor(ctx, command); - return processRetVal(retVal); - } - - @Override - public Object visitGetKeysCommand(InvocationContext ctx, GetKeysCommand command) throws Throwable - { - Object retVal = invokeNextInterceptor(ctx, command); - return processRetVal(retVal); - } - - @Override - public Object visitGetKeyValueCommand(InvocationContext ctx, GetKeyValueCommand command) throws Throwable - { - Set<MarshalledValue> marshalledValues = new HashSet<MarshalledValue>(); - if (!MarshalledValueHelper.isTypeExcluded(command.getKey().getClass())) - { - Object value = createAndAddMarshalledValue(command.getKey(), marshalledValues, ctx); - command.setKey(value); - } - Object retVal = invokeNextInterceptor(ctx, command); - return compactAndProcessRetVal(command, marshalledValues, retVal); - } - - private Object compactAndProcessRetVal(ReplicableCommand command, Set<MarshalledValue> marshalledValues, Object retVal) - throws IOException, ClassNotFoundException - { - if (trace) log.trace("Compacting MarshalledValues created"); - for (MarshalledValue mv : marshalledValues) mv.compact(false, false); - - return processRetVal(retVal); - } - - private Object processRetVal(Object retVal) - throws IOException, ClassNotFoundException - { - if (retVal instanceof MarshalledValue) - { - if (trace) log.trace("Return value is a MarshalledValue. Unwrapping."); - retVal = ((MarshalledValue) retVal).get(); - } - return retVal; - } - - @SuppressWarnings("unchecked") - protected Map wrapMap(Map<Object, Object> m, Set<MarshalledValue> marshalledValues, InvocationContext ctx) throws NotSerializableException - { - if (m == null) - { - if (trace) log.trace("Map is nul; returning an empty map."); - return Collections.emptyMap(); - } - if (trace) log.trace("Wrapping map contents of argument " + m); - Map copy = new HashMap(); - for (Map.Entry me : m.entrySet()) - { - Object key = me.getKey(); - Object value = me.getValue(); - copy.put((key == null || MarshalledValueHelper.isTypeExcluded(key.getClass())) ? key : createAndAddMarshalledValue(key, marshalledValues, ctx), - (value == null || MarshalledValueHelper.isTypeExcluded(value.getClass())) ? value : createAndAddMarshalledValue(value, marshalledValues, ctx)); - } - return copy; - } - - protected MarshalledValue createAndAddMarshalledValue(Object toWrap, Set<MarshalledValue> marshalledValues, InvocationContext ctx) throws NotSerializableException - { - MarshalledValue mv = new MarshalledValue(toWrap); - marshalledValues.add(mv); - if (!ctx.isOriginLocal()) mv.setEqualityPreferenceForInstance(false); - return mv; - } -} Deleted: core/branches/flat/src/main/java/org/jboss/cache/interceptors/NotificationInterceptor.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/cache/interceptors/NotificationInterceptor.java 2008-10-09 13:22:09 UTC (rev 6897) +++ core/branches/flat/src/main/java/org/jboss/cache/interceptors/NotificationInterceptor.java 2008-10-09 13:45:15 UTC (rev 6898) @@ -1,71 +0,0 @@ -/* - * 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.cache.interceptors; - -import org.jboss.cache.InvocationContext; -import org.jboss.cache.commands.tx.CommitCommand; -import org.jboss.cache.commands.tx.PrepareCommand; -import org.jboss.cache.commands.tx.RollbackCommand; -import org.jboss.cache.notifications.Notifier; -import org.jboss.starobrno.factories.annotations.Inject; - -/** - * The interceptor in charge of firing off notifications to cache listeners - * - * @author <a href="mailto:manik@jboss.org">Manik Surtani</a> - * @since 2.0.0 - */ -public class NotificationInterceptor extends BaseTransactionalContextInterceptor -{ - private Notifier notifier; - - @Inject - public void injectDependencies(Notifier notifier) - { - this.notifier = notifier; - } - - @Override - public Object visitPrepareCommand(InvocationContext ctx, PrepareCommand command) throws Throwable - { - Object retval = invokeNextInterceptor(ctx, command); - if (command.isOnePhaseCommit()) notifier.notifyTransactionCompleted(ctx.getTransaction(), true, ctx); - - return retval; - } - - @Override - public Object visitCommitCommand(InvocationContext ctx, CommitCommand command) throws Throwable - { - Object retval = invokeNextInterceptor(ctx, command); - notifier.notifyTransactionCompleted(ctx.getTransaction(), true, ctx); - return retval; - } - - @Override - public Object visitRollbackCommand(InvocationContext ctx, RollbackCommand command) throws Throwable - { - Object retval = invokeNextInterceptor(ctx, command); - notifier.notifyTransactionCompleted(ctx.getTransaction(), false, ctx); - return retval; - } -} Deleted: core/branches/flat/src/main/java/org/jboss/cache/interceptors/TxInterceptor.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/cache/interceptors/TxInterceptor.java 2008-10-09 13:22:09 UTC (rev 6897) +++ core/branches/flat/src/main/java/org/jboss/cache/interceptors/TxInterceptor.java 2008-10-09 13:45:15 UTC (rev 6898) @@ -1,1145 +0,0 @@ -/* - * 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.cache.interceptors; - -import org.jboss.cache.InvocationContext; -import org.jboss.cache.RPCManager; -import org.jboss.cache.ReplicationException; -import org.jboss.cache.commands.AbstractVisitor; -import org.jboss.cache.commands.CommandsFactory; -import org.jboss.cache.commands.ReplicableCommand; -import org.jboss.cache.commands.VisitableCommand; -import org.jboss.cache.commands.WriteCommand; -import org.jboss.cache.commands.tx.CommitCommand; -import org.jboss.cache.commands.tx.PrepareCommand; -import org.jboss.cache.commands.tx.RollbackCommand; -import org.jboss.cache.commands.write.ClearDataCommand; -import org.jboss.cache.commands.write.InvalidateCommand; -import org.jboss.cache.commands.write.PutDataMapCommand; -import org.jboss.cache.commands.write.PutKeyValueCommand; -import org.jboss.cache.commands.write.RemoveKeyCommand; -import org.jboss.cache.commands.write.RemoveNodeCommand; -import org.jboss.cache.invocation.InvocationContextContainer; -import org.jboss.cache.jmx.annotations.ManagedAttribute; -import org.jboss.cache.jmx.annotations.ManagedOperation; -import org.jboss.cache.lock.LockManager; -import org.jboss.cache.notifications.Notifier; -import org.jboss.cache.transaction.GlobalTransaction; -import org.jboss.cache.transaction.TransactionContext; -import org.jboss.cache.transaction.TransactionTable; -import org.jboss.cache.util.concurrent.ConcurrentHashSet; -import org.jboss.starobrno.CacheException; -import org.jboss.starobrno.config.Option; -import org.jboss.starobrno.factories.ComponentRegistry; -import org.jboss.starobrno.factories.annotations.Inject; -import org.jboss.starobrno.factories.context.ContextFactory; - -import javax.transaction.InvalidTransactionException; -import javax.transaction.Status; -import javax.transaction.Synchronization; -import javax.transaction.SystemException; -import javax.transaction.Transaction; -import java.util.Collections; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; - -/** - * This interceptor is the new default at the head of all interceptor chains, - * and makes transactional attributes available to all interceptors in the chain. - * This interceptor is also responsible for registering for synchronisation on - * transaction completion. - * - * @author <a href="mailto:manik@jboss.org">Manik Surtani (manik(a)jboss.org)&lt;/a&gt; - * @author <a href="mailto:stevew@jofti.com">Steve Woodcock (stevew(a)jofti.com)&lt;/a&gt; - */ -public class TxInterceptor extends BaseTransactionalContextInterceptor -{ - protected CommandsFactory commandsFactory; - protected RPCManager rpcManager; - private Notifier notifier; - private InvocationContextContainer invocationContextContainer; - private ComponentRegistry componentRegistry; - private ContextFactory contextFactory; - - /** - * List <Transaction>that we have registered for - */ - private final Set<Transaction> transactions = new ConcurrentHashSet<Transaction>(); - private final Map<Transaction, GlobalTransaction> rollbackTransactions = new ConcurrentHashMap<Transaction, GlobalTransaction>(16); - private long prepares = 0; - private long commits = 0; - private long rollbacks = 0; - protected boolean optimistic = false; - private LockManager lockManager; - private boolean statsEnabled; - - @Inject - public void intialize(RPCManager rpcManager, ContextFactory contextFactory, - Notifier notifier, InvocationContextContainer icc, - CommandsFactory factory, ComponentRegistry componentRegistry, LockManager lockManager) - { - this.contextFactory = contextFactory; - this.commandsFactory = factory; - this.rpcManager = rpcManager; - this.notifier = notifier; - this.invocationContextContainer = icc; - this.componentRegistry = componentRegistry; - this.lockManager = lockManager; - setStatisticsEnabled(configuration.getExposeManagementStatistics()); - } - - @Override - public Object visitPrepareCommand(InvocationContext ctx, PrepareCommand command) throws Throwable - { - Object result = null; - - // this is a prepare, commit, or rollback. - if (log.isDebugEnabled()) log.debug("Got gtx from invocation context " + ctx.getGlobalTransaction()); - try - { - if (ctx.getGlobalTransaction().isRemote()) - { - result = handleRemotePrepare(ctx, command); - if (getStatisticsEnabled()) prepares++; - } - else - { - if (trace) log.trace("received my own message (discarding it)"); - result = null; - } - } - catch (Throwable e) - { - ctx.throwIfNeeded(e); - } - - return result; - } - - @Override - public Object visitCommitCommand(InvocationContext ctx, CommitCommand command) throws Throwable - { - if (!ctx.getGlobalTransaction().isRemote()) - { - if (trace) log.trace("received my own message (discarding it)"); - return null; - } - try - { - if (trace) log.trace("(" + rpcManager.getLocalAddress() + ") call on command [" + command + "]"); - GlobalTransaction gtx = ctx.getGlobalTransaction(); - Transaction ltx = txTable.getLocalTransaction(gtx, true); - // disconnect if we have a current tx associated - Transaction currentTx = txManager.getTransaction(); - boolean resumeCurrentTxOnCompletion = false; - try - { - if (!ltx.equals(currentTx)) - { - currentTx = txManager.suspend(); - resumeCurrentTxOnCompletion = true; - txManager.resume(ltx); - // make sure we set this in the ctx - ctx.setTransaction(ltx); - } - if (log.isDebugEnabled()) log.debug(" executing commit() with local TX " + ltx + " under global tx " + gtx); - txManager.commit(); - if (getStatisticsEnabled()) commits++; - } - finally - { - //resume the old transaction if we suspended it - if (resumeCurrentTxOnCompletion) - { - resumeTransactionOnCompletion(ctx, currentTx); - } - // remove from local lists. - transactions.remove(ltx); - // this tx has completed. Clean up in the tx table. - txTable.remove(gtx, ltx); - } - if (log.isDebugEnabled()) log.debug("Finished remote rollback method for " + gtx); - } - catch (Throwable throwable) - { - ctx.throwIfNeeded(throwable); - } - - return null; - } - - @Override - public Object visitRollbackCommand(InvocationContext ctx, RollbackCommand command) throws Throwable - { - if (!ctx.getGlobalTransaction().isRemote()) - { - if (trace) log.trace("received my own message (discarding it)"); - return null; - } - try - { - if (trace) log.trace("(" + rpcManager.getLocalAddress() + ") call on command [" + command + "]"); - GlobalTransaction gtx = ctx.getGlobalTransaction(); - Transaction ltx = txTable.getLocalTransaction(gtx); - if (ltx == null) - { - log.warn("No local transaction for this remotely originating rollback. Possibly rolling back before a prepare call was broadcast?"); - txTable.remove(gtx); - return null; - } - // disconnect if we have a current tx associated - Transaction currentTx = txManager.getTransaction(); - boolean resumeCurrentTxOnCompletion = false; - try - { - if (!ltx.equals(currentTx)) - { - currentTx = txManager.suspend(); - resumeCurrentTxOnCompletion = true; - txManager.resume(ltx); - // make sure we set this in the ctx - ctx.setTransaction(ltx); - } - if (log.isDebugEnabled()) log.debug("executing with local TX " + ltx + " under global tx " + gtx); - txManager.rollback(); - if (getStatisticsEnabled()) rollbacks++; - } - finally - { - //resume the old transaction if we suspended it - if (resumeCurrentTxOnCompletion) - { - resumeTransactionOnCompletion(ctx, currentTx); - } - - // remove from local lists. - transactions.remove(ltx); - - // this tx has completed. Clean up in the tx table. - txTable.remove(gtx, ltx); - } - if (log.isDebugEnabled()) log.debug("Finished remote commit/rollback method for " + gtx); - } - catch (Throwable throwable) - { - ctx.throwIfNeeded(throwable); - } - - return null; - } - - @Override - public Object visitInvalidateCommand(InvocationContext ctx, InvalidateCommand command) throws Throwable - { - return invokeNextInterceptor(ctx, command); - } - - /** - * Tests if we already have a tx running. If so, register a sync handler for this method invocation. - * if not, create a local tx if we're using opt locking. - * - * @throws Throwable - */ - @Override - public Object handleDefault(InvocationContext ctx, VisitableCommand command) throws Throwable - { - try - { - return attachGtxAndPassUpChain(ctx, command); - } - catch (Throwable throwable) - { - ctx.throwIfNeeded(throwable); - return null; - } - } - - protected Object attachGtxAndPassUpChain(InvocationContext ctx, VisitableCommand command) throws Throwable - { - Transaction tx = ctx.getTransaction(); - if (tx != null) attachGlobalTransaction(ctx, tx, command); - return invokeNextInterceptor(ctx, command); - } - - // ------------------------------------------------------------------------ - // JMX statistics - // ------------------------------------------------------------------------ - - // -------------------------------------------------------------- - - /** - * Handles a remotely originating prepare call, by creating a local transaction for the remote global transaction - * and replaying modifications in this new local transaction. - * - * @param ctx invocation context - * @param command prepare command - * @return result of the prepare, typically a null. - * @throws Throwable in the event of problems. - */ - private Object handleRemotePrepare(InvocationContext ctx, PrepareCommand command) throws Throwable - { - // the InvocationContextInterceptor would have set this for us - GlobalTransaction gtx = ctx.getGlobalTransaction(); - - // Is there a local transaction associated with GTX? (not the current tx associated with the thread, which may be - // in the invocation context - Transaction ltx = txTable.getLocalTransaction(gtx); - Transaction currentTx = txManager.getTransaction(); - - Object retval = null; - boolean success = false; - try - { - if (ltx == null) - { - if (currentTx != null) txManager.suspend(); - // create a new local transaction - ltx = createLocalTx(); - // associate this with a global tx - txTable.put(ltx, gtx); - if (trace) log.trace("Created new tx for gtx " + gtx); - - if (log.isDebugEnabled()) - log.debug("Started new local tx as result of remote prepare: local tx=" + ltx + " (status=" + ltx.getStatus() + "), gtx=" + gtx); - } - else - { - //this should be valid - if (!TransactionTable.isValid(ltx)) - throw new CacheException("Transaction " + ltx + " not in correct state to be prepared"); - - //associate this thread with the local transaction associated with the global transaction, IF the localTx is NOT the current tx. - if (currentTx == null || !ltx.equals(currentTx)) - { - if (trace) log.trace("Suspending current tx " + currentTx); - txManager.suspend(); - txManager.resume(ltx); - } - } - if (trace) log.trace("Resuming existing tx " + ltx + ", global tx=" + gtx); - - // at this point we have a non-null ltx - - // Asssociate the local TX with the global TX. Create new - // transactionContext for TX in txTable, the modifications - // below will need this transactionContext to add their modifications - // under the GlobalTx key - TransactionContext transactionContext = txTable.get(gtx); - if (transactionContext == null) - { - // create a new transaction transactionContext - if (log.isDebugEnabled()) log.debug("creating new tx transactionContext"); -// transactionContext = contextFactory.createTransactionContext(ltx); - throw new RuntimeException("Delete this class!"); -// txTable.put(gtx, transactionContext); - } - - setTransactionalContext(ltx, gtx, transactionContext, ctx); - - // register a sync handler for this tx. - registerHandler(ltx, new RemoteSynchronizationHandler(gtx, ltx, transactionContext), ctx); - - success = false; - - // replay modifications - replayModifications(ctx, ltx, command); - - success = true; // no exceptions were thrown above!! - - // now pass the prepare command up the chain as well. - if (command.isOnePhaseCommit()) - { - if (trace) - log.trace("Using one-phase prepare. Not propagating the prepare call up the stack until called to do so by the sync handler."); - } - else - { - // now pass up the prepare method itself. - invokeNextInterceptor(ctx, command); - } - // JBCACHE-361 Confirm that the transaction is ACTIVE - assertTxIsStillValid(ltx); - } - finally - { - // if we are running a one-phase commit, perform a commit or rollback now. - if (trace) log.trace("Are we running a 1-phase commit? " + command.isOnePhaseCommit()); - - if (command.isOnePhaseCommit()) - { - try - { - if (success) - { - ltx.commit(); - } - else - { - ltx.rollback(); - } - } - catch (Throwable t) - { - log.error("Commit/rollback failed.", t); - if (success) - { - // try another rollback... - try - { - log.info("Attempting anotehr rollback"); - //invokeOnePhaseCommitMethod(globalTransaction, modifications.size() > 0, false); - ltx.rollback(); - } - catch (Throwable t2) - { - log.error("Unable to rollback", t2); - } - } - } - finally - { - transactions.remove(ltx);// JBAS-298 - } - } - - txManager.suspend();// suspends ltx - could be null - // resume whatever else we had going. - if (currentTx != null) txManager.resume(currentTx); - if (log.isDebugEnabled()) log.debug("Finished remote prepare " + gtx); - } - - return retval; - } - - private ReplicableCommand attachGlobalTransaction(InvocationContext ctx, Transaction tx, VisitableCommand command) throws Throwable - { - if (trace) - { - log.trace(" local transaction exists - registering global tx if not present for " + Thread.currentThread()); - } - if (trace) - { - GlobalTransaction tempGtx = txTable.get(tx); - log.trace("Associated gtx in txTable is " + tempGtx); - } - - // register a sync handler for this tx - only if the globalTransaction is not remotely initiated. - GlobalTransaction gtx = registerTransaction(tx, ctx); - if (gtx != null) - { - command = replaceGtx(command, gtx); - } - else - { - // get the current globalTransaction from the txTable. - gtx = txTable.get(tx); - } - - // make sure we attach this globalTransaction to the invocation context. - ctx.setGlobalTransaction(gtx); - - return command; - } - - /** - * Replays modifications - */ - protected void replayModifications(InvocationContext ctx, Transaction ltx, PrepareCommand command) throws Throwable - { - try - { - // replay modifications - for (WriteCommand modification : command.getModifications()) - { - invokeNextInterceptor(ctx, modification); - assertTxIsStillValid(ltx); - } - } - catch (Throwable th) - { - log.error("prepare failed!", th); - throw th; - } - } - - private void resumeTransactionOnCompletion(InvocationContext ctx, Transaction currentTx) - throws SystemException, InvalidTransactionException - { - if (trace) log.trace("Resuming suspended transaction " + currentTx); - txManager.suspend(); - if (currentTx != null) - { - txManager.resume(currentTx); - ctx.setTransaction(currentTx); - } - } - - /** - * Handles a commit or a rollback. Called by the synch handler. Simply tests that we are in the correct tx and - * passes the meth call up the interceptor chain. - * - * @throws Throwable - */ - @SuppressWarnings("deprecation") - private Object handleCommitRollback(InvocationContext ctx, VisitableCommand command) throws Throwable - { - GlobalTransaction gtx = ctx.getGlobalTransaction(); - Object result; - VisitableCommand originalCommand = ctx.getCommand(); - ctx.setCommand(command); - try - { - result = invokeNextInterceptor(ctx, command); - } - finally - { - ctx.setCommand(originalCommand); - ctx.setMethodCall(null); - } - if (log.isDebugEnabled()) log.debug("Finished local commit/rollback method for " + gtx); - return result; - } - - // -------------------------------------------------------------- - // Transaction phase runners - // -------------------------------------------------------------- - - protected PrepareCommand buildPrepareCommand(GlobalTransaction gtx, List modifications, boolean onePhaseCommit) - { - return commandsFactory.buildPrepareCommand(gtx, modifications, rpcManager.getLocalAddress(), onePhaseCommit); - } - - /** - * creates a commit() - */ - protected void runCommitPhase(InvocationContext ctx, GlobalTransaction gtx, List modifications, boolean onePhaseCommit) - { - try - { - VisitableCommand commitCommand = onePhaseCommit ? buildPrepareCommand(gtx, modifications, true) : commandsFactory.buildCommitCommand(gtx); - - if (trace) log.trace("Running commit for " + gtx); - - handleCommitRollback(ctx, commitCommand); - } - catch (Throwable e) - { - log.warn("Commit failed. Clearing stale locks."); - try - { - cleanupStaleLocks(ctx); - } - catch (RuntimeException re) - { - log.error("Unable to clear stale locks", re); - throw re; - } - catch (Throwable e2) - { - log.error("Unable to clear stale locks", e2); - throw new RuntimeException(e2); - } - if (e instanceof RuntimeException) - throw (RuntimeException) e; - else - throw new RuntimeException("Commit failed.", e); - } - } - - protected void cleanupStaleLocks(InvocationContext ctx) throws Throwable - { - TransactionContext transactionContext = ctx.getTransactionContext(); - if (transactionContext != null) lockManager.unlock(ctx); - } - - /** - * creates a rollback() - */ - protected void runRollbackPhase(InvocationContext ctx, GlobalTransaction gtx, Transaction tx) - { - try - { - // JBCACHE-457 - VisitableCommand rollbackCommand = commandsFactory.buildRollbackCommand(gtx); - if (trace) log.trace(" running rollback for " + gtx); - - //JBCACHE-359 Store a lookup for the globalTransaction so a listener - // callback can find it - rollbackTransactions.put(tx, gtx); - - handleCommitRollback(ctx, rollbackCommand); - } - catch (Throwable e) - { - log.warn("Rollback had a problem", e); - } - finally - { - rollbackTransactions.remove(tx); - } - } - - private boolean isOnePhaseCommit() - { - if (!configuration.getCacheMode().isSynchronous() && !optimistic) - { - // this is a REPL_ASYNC call - do 1-phase commit. break! - if (trace) log.trace("This is a REPL_ASYNC call (1 phase commit) - do nothing for beforeCompletion()"); - return true; - } - return false; - } - - /** - * Handles a local prepare - invoked by the sync handler. Tests if the current tx matches the gtx passed in to the - * method call and passes the prepare() call up the chain. - */ - @SuppressWarnings("deprecation") - public Object runPreparePhase(InvocationContext ctx, GlobalTransaction gtx, List<WriteCommand> modifications) throws Throwable - { - // running a 2-phase commit. - VisitableCommand prepareCommand = buildPrepareCommand(gtx, modifications, false); - - Object result; - - // Is there a local transaction associated with GTX ? - Transaction ltx = ctx.getTransaction(); - - //if ltx is not null and it is already running - Transaction currentTransaction = txManager.getTransaction(); - if (currentTransaction != null && ltx != null && currentTransaction.equals(ltx)) - { - VisitableCommand originalCommand = ctx.getCommand(); - ctx.setCommand(prepareCommand); - try - { - result = invokeNextInterceptor(ctx, prepareCommand); - } - finally - { - ctx.setCommand(originalCommand); - ctx.setMethodCall(null); - } - } - else - { - log.warn("Local transaction does not exist or does not match expected transaction " + gtx); - throw new CacheException(" local transaction " + ltx + " does not exist or does not match expected transaction " + gtx); - } - return result; - } - - // -------------------------------------------------------------- - // Private helper methods - // -------------------------------------------------------------- - - protected void assertTxIsStillValid(Transaction tx) - { - if (!TransactionTable.isActive(tx)) - { - try - { - throw new ReplicationException("prepare() failed -- local transaction status is not STATUS_ACTIVE; is " + tx.getStatus()); - } - catch (SystemException e) - { - throw new ReplicationException("prepare() failed -- local transaction status is not STATUS_ACTIVE; Unable to retrieve transaction status."); - } - } - } - - /** - * Creates a gtx (if one doesnt exist), a sync handler, and registers the tx. - */ - private GlobalTransaction registerTransaction(Transaction tx, InvocationContext ctx) throws Exception - { - GlobalTransaction gtx; - - if (TransactionTable.isValid(tx) && transactions.add(tx)) - { - gtx = txTable.getCurrentTransaction(tx, true); - TransactionContext transactionContext; - if (ctx.getGlobalTransaction() == null) - { - ctx.setGlobalTransaction(gtx); - transactionContext = txTable.get(gtx); - ctx.setTransactionContext(transactionContext); - } - else - { - transactionContext = ctx.getTransactionContext(); - } - if (gtx.isRemote()) - { - // should be no need to register a handler since this a remotely initiated globalTransaction - if (trace) log.trace("is a remotely initiated gtx so no need to register a tx for it"); - } - else - { - if (trace) log.trace("Registering sync handler for tx " + tx + ", gtx " + gtx); - - // see the comment in the LocalSyncHandler for the last isOriginLocal param. - LocalSynchronizationHandler myHandler = new LocalSynchronizationHandler(gtx, tx, transactionContext, !ctx.isOriginLocal()); - registerHandler(tx, myHandler, ctx); - } - } - else if ((gtx = rollbackTransactions.get(tx)) != null) - { - if (trace) log.trace("Transaction " + tx + " is already registered and is rolling back."); - } - else - { - if (trace) log.trace("Transaction " + tx + " is already registered."); - } - return gtx; - } - - /** - * Registers a sync hander against a tx. - */ - private void registerHandler(Transaction tx, Synchronization handler, InvocationContext ctx) throws Exception - { - OrderedSynchronizationHandler orderedHandler = ctx.getTransactionContext().getOrderedSynchronizationHandler(); //OrderedSynchronizationHandler.getInstance(tx); - - if (trace) log.trace("registering for TX completion: SynchronizationHandler(" + handler + ")"); - - orderedHandler.registerAtHead(handler);// needs to be invoked first on TX commit - - notifier.notifyTransactionRegistered(tx, ctx); - } - - /** - * Replaces the global transaction in a VisitableCommand with a new global transaction passed in. - */ - private VisitableCommand replaceGtx(VisitableCommand command, final GlobalTransaction gtx) throws Throwable - { - command.acceptVisitor(null, new AbstractVisitor() - { - @Override - public Object visitPutDataMapCommand(InvocationContext ctx, PutDataMapCommand command) throws Throwable - { - command.setGlobalTransaction(gtx); - return null; - } - - @Override - public Object visitPutKeyValueCommand(InvocationContext ctx, PutKeyValueCommand command) throws Throwable - { - command.setGlobalTransaction(gtx); - return null; - } - - @Override - public Object visitClearDataCommand(InvocationContext ctx, ClearDataCommand command) throws Throwable - { - command.setGlobalTransaction(gtx); - return null; - } - - @Override - public Object visitRemoveNodeCommand(InvocationContext ctx, RemoveNodeCommand command) throws Throwable - { - command.setGlobalTransaction(gtx); - return null; - } - - @Override - public Object visitRemoveKeyCommand(InvocationContext ctx, RemoveKeyCommand command) throws Throwable - { - command.setGlobalTransaction(gtx); - return null; - } - - @Override - public Object visitCommitCommand(InvocationContext ctx, CommitCommand command) throws Throwable - { - command.setGlobalTransaction(gtx); - return null; - } - - @Override - public Object visitPrepareCommand(InvocationContext ctx, PrepareCommand command) throws Throwable - { - command.setGlobalTransaction(gtx); - return null; - } - - @Override - public Object visitRollbackCommand(InvocationContext ctx, RollbackCommand command) throws Throwable - { - command.setGlobalTransaction(gtx); - return null; - } - - }); - return command; - } - - /** - * Creates and starts a local tx - * - * @throws Exception - */ - protected Transaction createLocalTx() throws Exception - { - if (trace) - { - log.trace("Creating transaction for thread " + Thread.currentThread()); - } - Transaction localTx; - if (txManager == null) throw new Exception("Failed to create local transaction; TransactionManager is null"); - txManager.begin(); - localTx = txManager.getTransaction(); - return localTx; - } - - // ------------------------------------------------------------------------ - // Synchronization classes - // ------------------------------------------------------------------------ - - // this controls the whole transaction - - private class RemoteSynchronizationHandler implements Synchronization - { - Transaction tx = null; - GlobalTransaction gtx = null; - List<WriteCommand> modifications = null; - TransactionContext transactionContext = null; - protected InvocationContext ctx; // the context for this call. - - RemoteSynchronizationHandler(GlobalTransaction gtx, Transaction tx, TransactionContext entry) - { - this.gtx = gtx; - this.tx = tx; - this.transactionContext = entry; - } - - public void beforeCompletion() - { - if (trace) log.trace("Running beforeCompletion on gtx " + gtx); - - if (transactionContext == null) - { - log.error("Transaction has a null transaction entry - beforeCompletion() will fail."); - throw new IllegalStateException("cannot find transaction entry for " + gtx); - } - - modifications = transactionContext.getModifications(); - ctx = invocationContextContainer.get(); - setTransactionalContext(tx, gtx, transactionContext, ctx); - - if (ctx.isOptionsUninitialised() && transactionContext.getOption() != null) - ctx.setOptionOverrides(transactionContext.getOption()); - - assertCanContinue(); - - ctx.setOriginLocal(false); - } - - // this should really not be done here - - // it is supposed to be post commit not actually run the commit - public void afterCompletion(int status) - { - // could happen if a rollback is called and beforeCompletion() doesn't get called. - if (ctx == null) - { - ctx = invocationContextContainer.get(); - setTransactionalContext(tx, gtx, transactionContext, ctx); - - if (ctx.isOptionsUninitialised() && transactionContext != null && transactionContext.getOption() != null) - { - // use the options from the transaction entry instead - ctx.setOptionOverrides(transactionContext.getOption()); - } - } - - try - { - assertCanContinue(); - - try - { - if (txManager.getTransaction() != null && !txManager.getTransaction().equals(tx)) txManager.resume(tx); - } - catch (Exception e) - { - log.error("afterCompletion error: " + status, e); - } - - if (trace) log.trace("calling aftercompletion for " + gtx); - - // set any transaction wide options as current for this thread. - if (transactionContext != null) - { - // this should ideally be set in beforeCompletion(), after compacting the list. - if (modifications == null) modifications = transactionContext.getModifications(); - ctx.setOptionOverrides(transactionContext.getOption()); - } - if (tx != null) transactions.remove(tx); - - switch (status) - { - case Status.STATUS_COMMITTED: - boolean onePhaseCommit = isOnePhaseCommit(); - if (log.isDebugEnabled()) log.debug("Running commit phase. One phase? " + onePhaseCommit); - runCommitPhase(ctx, gtx, modifications, onePhaseCommit); - log.debug("Finished commit phase"); - break; - case Status.STATUS_UNKNOWN: - log.warn("Received JTA STATUS_UNKNOWN in afterCompletion()! XA resources may not be in sync. The app should manually clean up resources at this point."); - case Status.STATUS_MARKED_ROLLBACK: - case Status.STATUS_ROLLEDBACK: - log.debug("Running rollback phase"); - runRollbackPhase(ctx, gtx, tx); - log.debug("Finished rollback phase"); - break; - - default: - throw new IllegalStateException("illegal status: " + status); - } - } - catch (Exception th) - { - log.trace("Caught exception ", th); - - } - finally - { - // clean up the tx table - txTable.remove(gtx); - txTable.remove(tx); - setTransactionalContext(null, null, null, ctx); - cleanupInternalState(); - } - } - - private void assertCanContinue() - { - if (!componentRegistry.invocationsAllowed(true) && (ctx.getOptionOverrides() == null || !ctx.getOptionOverrides().isSkipCacheStatusCheck())) - throw new IllegalStateException("Cache not in STARTED state!"); - } - - /** - * Cleans out (nullifies) member variables held by the sync object for easier gc. Could be (falsely) seen as a mem - * leak if the TM implementation hangs on to the synchronizations for an unnecessarily long time even after the tx - * completes. See JBCACHE-1007. - */ - private void cleanupInternalState() - { - tx = null; - gtx = null; - modifications = null; - if (transactionContext != null) transactionContext.reset(); - transactionContext = null; - } - - @Override - public String toString() - { - return "TxInterceptor.RemoteSynchronizationHandler(gtx=" + gtx + ", tx=" + getTxAsString() + ")"; - } - - protected String getTxAsString() - { - // JBCACHE-1114 -- don't call toString() on tx or it can lead to stack overflow - if (tx == null) - return null; - - return tx.getClass().getName() + "@" + System.identityHashCode(tx); - } - } - - private class LocalSynchronizationHandler extends RemoteSynchronizationHandler - { - private boolean localRollbackOnly = true; - // a VERY strange situation where a tx has remote origins, but since certain buddy group org methods perform local - // cleanups even when remotely triggered, and optimistic locking is used, you end up with an implicit local tx. - // This is STILL remotely originating though and this needs to be made explicit here. - // this can be checked by inspecting the InvocationContext.isOriginLocal() at the time of registering the sync. - private boolean remoteLocal = false; - private Option originalOptions, transactionalOptions; - - /** - * A Synchronization for locally originating txs. - * <p/> - * a VERY strange situation where a tx has remote origins, but since certain buddy group org methods perform local - * cleanups even when remotely triggered, and optimistic locking is used, you end up with an implicit local tx. - * This is STILL remotely originating though and this needs to be made explicit here. - * this can be checked by inspecting the InvocationContext.isOriginLocal() at the time of registering the sync. - * - * @param gtx - * @param tx - * @param remoteLocal - */ - LocalSynchronizationHandler(GlobalTransaction gtx, Transaction tx, TransactionContext transactionContext, boolean remoteLocal) - { - super(gtx, tx, transactionContext); - this.remoteLocal = remoteLocal; - } - - @Override - public void beforeCompletion() - { - super.beforeCompletion(); - ctx.setOriginLocal(!remoteLocal); // this is the LOCAL sync handler after all! - // fetch the modifications before the transaction is committed - // (and thus removed from the txTable) - setTransactionalContext(tx, gtx, transactionContext, ctx); - if (!transactionContext.hasModifications()) - { - if (trace) log.trace("No modifications in this tx. Skipping beforeCompletion()"); - modifications = Collections.emptyList(); - return; - } - else - { - modifications = transactionContext.getModifications(); - } - - // set any transaction wide options as current for this thread, caching original options that would then be reset - originalOptions = ctx.getOptionOverrides(); - transactionalOptions = transactionContext.getOption(); - ctx.setOptionOverrides(transactionalOptions); - - try - { - switch (tx.getStatus()) - { - // if we are active or preparing then we can go ahead - case Status.STATUS_ACTIVE: - case Status.STATUS_PREPARING: - // run a prepare call. - - Object result = isOnePhaseCommit() ? null : runPreparePhase(ctx, gtx, modifications); - - if (result instanceof Throwable) - { - if (log.isDebugEnabled()) - log.debug("Transaction needs to be rolled back - the cache returned an instance of Throwable for this prepare call (tx=" + tx + " and gtx=" + gtx + ")", (Throwable) result); - tx.setRollbackOnly(); - throw (Throwable) result; - } - break; - default: - throw new CacheException("transaction " + tx + " in status " + tx.getStatus() + " unable to start transaction"); - } - } - catch (Throwable t) - { - if (log.isWarnEnabled()) log.warn("Caught exception, will now set transaction to roll back", t); - try - { - tx.setRollbackOnly(); - } - catch (SystemException se) - { - throw new RuntimeException("setting tx rollback failed ", se); - } - if (t instanceof RuntimeException) - throw (RuntimeException) t; - else - throw new RuntimeException("", t); - } - finally - { - localRollbackOnly = false; - setTransactionalContext(null, null, null, ctx); - ctx.setOptionOverrides(originalOptions); - } - } - - @Override - public void afterCompletion(int status) - { - // could happen if a rollback is called and beforeCompletion() doesn't get called. - if (ctx == null) ctx = invocationContextContainer.get(); - ctx.setLocalRollbackOnly(localRollbackOnly); - setTransactionalContext(tx, gtx, transactionContext, ctx); - if (transactionalOptions != null) ctx.setOptionOverrides(transactionalOptions); - try - { - super.afterCompletion(status); - } - finally - { - ctx.setOptionOverrides(originalOptions); - } - } - - @Override - public String toString() - { - return "TxInterceptor.LocalSynchronizationHandler(gtx=" + gtx + ", tx=" + getTxAsString() + ")"; - } - } - - @ManagedOperation - public void resetStatistics() - { - prepares = 0; - commits = 0; - rollbacks = 0; - } - - @ManagedOperation - public Map<String, Object> dumpStatistics() - { - Map<String, Object> retval = new HashMap<String, Object>(3); - retval.put("Prepares", prepares); - retval.put("Commits", commits); - retval.put("Rollbacks", rollbacks); - return retval; - } - - @ManagedAttribute - public boolean getStatisticsEnabled() - { - return this.statsEnabled; - } - - @ManagedAttribute - public void setStatisticsEnabled(boolean enabled) - { - this.statsEnabled = enabled; - } - - @ManagedAttribute(description = "number of transaction prepares") - public long getPrepares() - { - return prepares; - } - - @ManagedAttribute(description = "number of transaction commits") - public long getCommits() - { - return commits; - } - - @ManagedAttribute(description = "number of transaction rollbacks") - public long getRollbacks() - { - return rollbacks; - } -} \ No newline at end of file Modified: core/branches/flat/src/main/java/org/jboss/cache/marshall/CacheMarshaller200.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/cache/marshall/CacheMarshaller200.java 2008-10-09 13:22:09 UTC (rev 6897) +++ core/branches/flat/src/main/java/org/jboss/cache/marshall/CacheMarshaller200.java 2008-10-09 13:45:15 UTC (rev 6898) @@ -31,6 +31,7 @@ import org.jboss.starobrno.commands.CommandsFactory; import org.jboss.starobrno.commands.ReplicableCommand; import org.jboss.starobrno.factories.annotations.Inject; +import org.jboss.starobrno.marshall.MarshalledValue; import org.jboss.starobrno.transaction.GlobalTransaction; import org.jgroups.Address; import org.jgroups.stack.IpAddress; Modified: core/branches/flat/src/main/java/org/jboss/cache/notifications/NotifierImpl.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/cache/notifications/NotifierImpl.java 2008-10-09 13:22:09 UTC (rev 6897) +++ core/branches/flat/src/main/java/org/jboss/cache/notifications/NotifierImpl.java 2008-10-09 13:45:15 UTC (rev 6898) @@ -26,7 +26,6 @@ import org.jboss.cache.Fqn; import org.jboss.cache.InvocationContext; import org.jboss.cache.buddyreplication.BuddyGroup; -import org.jboss.cache.marshall.MarshalledValueMap; import org.jboss.cache.notifications.annotation.*; import org.jboss.cache.notifications.event.*; import org.jboss.cache.util.Immutables; @@ -40,6 +39,7 @@ import org.jboss.starobrno.factories.annotations.NonVolatile; import org.jboss.starobrno.factories.annotations.Start; import org.jboss.starobrno.factories.annotations.Stop; +import org.jboss.starobrno.marshall.MarshalledValueMap; import org.jgroups.View; import javax.transaction.Transaction; Modified: core/branches/flat/src/main/java/org/jboss/starobrno/commands/write/PutKeyValueCommand.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/commands/write/PutKeyValueCommand.java 2008-10-09 13:22:09 UTC (rev 6897) +++ core/branches/flat/src/main/java/org/jboss/starobrno/commands/write/PutKeyValueCommand.java 2008-10-09 13:45:15 UTC (rev 6898) @@ -50,6 +50,16 @@ { } + public Object getValue() + { + return value; + } + + public void setValue(Object value) + { + this.value = value; + } + public Object acceptVisitor(InvocationContext ctx, Visitor visitor) throws Throwable { return visitor.visitPutKeyValueCommand(ctx, this); Modified: core/branches/flat/src/main/java/org/jboss/starobrno/commands/write/PutMapCommand.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/commands/write/PutMapCommand.java 2008-10-09 13:22:09 UTC (rev 6897) +++ core/branches/flat/src/main/java/org/jboss/starobrno/commands/write/PutMapCommand.java 2008-10-09 13:45:15 UTC (rev 6898) @@ -57,7 +57,7 @@ for (Entry<Object, Object> e : map.entrySet()) { MVCCEntry me = ctx.lookupEntry(e.getKey()); - if (e != null) e.setValue(e.getValue()); + if (e != null) e.setValue(me.getValue()); } return null; } @@ -67,6 +67,11 @@ return map; } + public void setMap(Map<Object, Object> map) + { + this.map = map; + } + public int getCommandId() { return METHOD_ID; Modified: core/branches/flat/src/main/java/org/jboss/starobrno/factories/InterceptorChainFactory.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/factories/InterceptorChainFactory.java 2008-10-09 13:22:09 UTC (rev 6897) +++ core/branches/flat/src/main/java/org/jboss/starobrno/factories/InterceptorChainFactory.java 2008-10-09 13:45:15 UTC (rev 6898) @@ -26,10 +26,14 @@ import org.jboss.starobrno.config.ConfigurationException; import org.jboss.starobrno.config.CustomInterceptorConfig; import org.jboss.starobrno.factories.annotations.DefaultFactoryFor; +import org.jboss.starobrno.interceptors.BatchingInterceptor; +import org.jboss.starobrno.interceptors.CacheMgmtInterceptor; import org.jboss.starobrno.interceptors.CallInterceptor; import org.jboss.starobrno.interceptors.InterceptorChain; import org.jboss.starobrno.interceptors.InvocationContextInterceptor; import org.jboss.starobrno.interceptors.LockingInterceptor; +import org.jboss.starobrno.interceptors.MarshalledValueInterceptor; +import org.jboss.starobrno.interceptors.NotificationInterceptor; import org.jboss.starobrno.interceptors.TxInterceptor; import org.jboss.starobrno.interceptors.base.CommandInterceptor; @@ -83,33 +87,30 @@ { boolean invocationBatching = configuration.isInvocationBatchingEnabled(); // load the icInterceptor first - // TODO: Uncomment once the BatchingInterceptor has been moved to Starobrno -// CommandInterceptor first = invocationBatching ? createInterceptor(BatchingInterceptor.class) : createInterceptor(InvocationContextInterceptor.class); - CommandInterceptor first = createInterceptor(InvocationContextInterceptor.class); + CommandInterceptor first = invocationBatching ? createInterceptor(BatchingInterceptor.class) : createInterceptor(InvocationContextInterceptor.class); + InterceptorChain interceptorChain = new InterceptorChain(first); // add the interceptor chain to the registry first, since some interceptors may ask for it. componentRegistry.registerComponent(interceptorChain, InterceptorChain.class); // NOW add the ICI if we are using batching! -// if (invocationBatching) -// interceptorChain.appendIntereceptor(createInterceptor(InvocationContextInterceptor.class)); + if (invocationBatching) + interceptorChain.appendIntereceptor(createInterceptor(InvocationContextInterceptor.class)); // load the cache management interceptor next - // TODO: Uncomment once the CacheMGMT has been moved to Starobrno + if (configuration.getExposeManagementStatistics()) + interceptorChain.appendIntereceptor(createInterceptor(CacheMgmtInterceptor.class)); -// if (configuration.getExposeManagementStatistics()) -// interceptorChain.appendIntereceptor(createInterceptor(CacheMgmtInterceptor.class)); - // load the tx interceptor interceptorChain.appendIntereceptor(createInterceptor(TxInterceptor.class)); - // TODO: Uncomment once the MarshalledValue, Notification has been moved to Starobrno -// if (configuration.isUseLazyDeserialization()) -// interceptorChain.appendIntereceptor(createInterceptor(MarshalledValueInterceptor.class)); -// interceptorChain.appendIntereceptor(createInterceptor(NotificationInterceptor.class)); + if (configuration.isUseLazyDeserialization()) + interceptorChain.appendIntereceptor(createInterceptor(MarshalledValueInterceptor.class)); + interceptorChain.appendIntereceptor(createInterceptor(NotificationInterceptor.class)); + // TODO: Uncomment once the Repl and Inval interceptors has been moved to Starobrno // switch (configuration.getCacheMode()) // { Modified: core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/BatchingInterceptor.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/BatchingInterceptor.java 2008-10-09 13:22:09 UTC (rev 6897) +++ core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/BatchingInterceptor.java 2008-10-09 13:45:15 UTC (rev 6898) @@ -21,11 +21,11 @@ */ package org.jboss.starobrno.interceptors; -import org.jboss.cache.InvocationContext; import org.jboss.starobrno.batch.BatchContainer; -import org.jboss.cache.commands.VisitableCommand; -import org.jboss.cache.interceptors.base.CommandInterceptor; +import org.jboss.starobrno.commands.VisitableCommand; +import org.jboss.starobrno.context.InvocationContext; import org.jboss.starobrno.factories.annotations.Inject; +import org.jboss.starobrno.interceptors.base.CommandInterceptor; import javax.transaction.Transaction; import javax.transaction.TransactionManager; Copied: core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/CacheMgmtInterceptor.java (from rev 6896, core/branches/flat/src/main/java/org/jboss/cache/interceptors/CacheMgmtInterceptor.java) =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/CacheMgmtInterceptor.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/CacheMgmtInterceptor.java 2008-10-09 13:45:15 UTC (rev 6898) @@ -0,0 +1,225 @@ +/* + * 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.interceptors; + +import org.jboss.starobrno.commands.read.GetKeyValueCommand; +import org.jboss.starobrno.commands.write.EvictCommand; +import org.jboss.starobrno.commands.write.PutKeyValueCommand; +import org.jboss.starobrno.commands.write.PutMapCommand; +import org.jboss.starobrno.container.DataContainer; +import org.jboss.starobrno.context.InvocationContext; +import org.jboss.starobrno.factories.annotations.Inject; +import org.jboss.starobrno.interceptors.base.JmxStatsCommandInterceptor; +import org.jboss.starobrno.jmx.annotations.ManagedAttribute; +import org.jboss.starobrno.jmx.annotations.ManagedOperation; + +import java.util.HashMap; +import java.util.Map; + +/** + * Captures cache management statistics + * + * @author Jerry Gauthier + * @version $Id$ + */ +public class CacheMgmtInterceptor extends JmxStatsCommandInterceptor +{ + private long hitTimes = 0; + private long missTimes = 0; + private long storeTimes = 0; + private long hits = 0; + private long misses = 0; + private long stores = 0; + private long evictions = 0; + private long start = System.currentTimeMillis(); + private long reset = start; + + private DataContainer dataContainer; + + @Inject + public void setDependencies(DataContainer dataContainer) + { + this.dataContainer = dataContainer; + } + + @Override + public Object visitEvictCommand(InvocationContext ctx, EvictCommand command) throws Throwable + { + Object returnValue = invokeNextInterceptor(ctx, command); + evictions++; + return returnValue; + } + + @Override + public Object visitGetKeyValueCommand(InvocationContext ctx, GetKeyValueCommand command) throws Throwable + { + long t1 = System.currentTimeMillis(); + Object retval = invokeNextInterceptor(ctx, command); + long t2 = System.currentTimeMillis(); + if (retval == null) + { + missTimes = missTimes + (t2 - t1); + misses++; + } + else + { + hitTimes = hitTimes + (t2 - t1); + hits++; + } + return retval; + } + + @Override + public Object visitPutMapCommand(InvocationContext ctx, PutMapCommand command) throws Throwable + { + Map data = command.getMap(); + long t1 = System.currentTimeMillis(); + Object retval = invokeNextInterceptor(ctx, command); + long t2 = System.currentTimeMillis(); + + if (data != null && data.size() > 0) + { + storeTimes = storeTimes + (t2 - t1); + stores = stores + data.size(); + } + return retval; + } + + @Override + public Object visitPutKeyValueCommand(InvocationContext ctx, PutKeyValueCommand command) throws Throwable + { + long t1 = System.currentTimeMillis(); + Object retval = invokeNextInterceptor(ctx, command); + long t2 = System.currentTimeMillis(); + storeTimes = storeTimes + (t2 - t1); + stores++; + return retval; + } + + @ManagedAttribute(description = "number of cache attribute hits") + public long getHits() + { + return hits; + } + + @ManagedAttribute(description = "number of cache attribute misses") + public long getMisses() + { + return misses; + } + + @ManagedAttribute(description = "number of cache attribute put operations") + public long getStores() + { + return stores; + } + + @ManagedAttribute(description = "number of cache eviction operations") + public long getEvictions() + { + return evictions; + } + + @ManagedAttribute(description = "hit/miss ratio for the cache") + public double getHitMissRatio() + { + double total = hits + misses; + if (total == 0) + return 0; + return (hits / total); + } + + @ManagedAttribute(description = "read/writes ratio for the cache") + public double getReadWriteRatio() + { + if (stores == 0) + return 0; + return (((double) (hits + misses) / (double) stores)); + } + + @ManagedAttribute(description = "average number of milliseconds for a read operation") + public long getAverageReadTime() + { + long total = hits + misses; + if (total == 0) + return 0; + return (hitTimes + missTimes) / total; + } + + @ManagedAttribute(description = "average number of milliseconds for a write operation") + public long getAverageWriteTime() + { + if (stores == 0) + return 0; + return (storeTimes) / stores; + } + + @ManagedAttribute(description = "number of entries in the cache") + public int getNumberOfEntries() + { + return dataContainer.size(); + } + + @ManagedAttribute(description = "seconds since cache started") + public long getElapsedTime() + { + return (System.currentTimeMillis() - start) / 1000; + } + + @ManagedAttribute(description = "number of seconds since the cache statistics were last reset") + public long getTimeSinceReset() + { + return (System.currentTimeMillis() - reset) / 1000; + } + + @ManagedOperation + public Map<String, Object> dumpStatistics() + { + Map<String, Object> retval = new HashMap<String, Object>(); + retval.put("Hits", hits); + retval.put("Misses", misses); + retval.put("Stores", stores); + retval.put("Evictions", evictions); + retval.put("NumberOfEntries", dataContainer.size()); + retval.put("ElapsedTime", getElapsedTime()); + retval.put("TimeSinceReset", getTimeSinceReset()); + retval.put("AverageReadTime", getAverageReadTime()); + retval.put("AverageWriteTime", getAverageWriteTime()); + retval.put("HitMissRatio", getHitMissRatio()); + retval.put("ReadWriteRatio", getReadWriteRatio()); + return retval; + } + + @ManagedOperation + public void resetStatistics() + { + hits = 0; + misses = 0; + stores = 0; + evictions = 0; + hitTimes = 0; + missTimes = 0; + storeTimes = 0; + reset = System.currentTimeMillis(); + } +} + Property changes on: core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/CacheMgmtInterceptor.java ___________________________________________________________________ Name: svn:keywords + Author Date Id Revision Name: svn:eol-style + native Copied: core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/MarshalledValueInterceptor.java (from rev 6896, core/branches/flat/src/main/java/org/jboss/cache/interceptors/MarshalledValueInterceptor.java) =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/MarshalledValueInterceptor.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/MarshalledValueInterceptor.java 2008-10-09 13:45:15 UTC (rev 6898) @@ -0,0 +1,155 @@ +/* + * 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.interceptors; + +import org.jboss.starobrno.commands.read.GetKeyValueCommand; +import org.jboss.starobrno.commands.write.PutKeyValueCommand; +import org.jboss.starobrno.commands.write.PutMapCommand; +import org.jboss.starobrno.commands.write.RemoveCommand; +import org.jboss.starobrno.context.InvocationContext; +import org.jboss.starobrno.interceptors.base.CommandInterceptor; +import org.jboss.starobrno.marshall.MarshalledValue; +import org.jboss.starobrno.marshall.MarshalledValueHelper; + +import java.io.IOException; +import java.io.NotSerializableException; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +/** + * Interceptor that handles the wrapping and unwrapping of cached data using {@link MarshalledValue}s. + * Known "excluded" types are not wrapped/unwrapped, which at this time include {@link String}, Java primitives + * and their Object wrappers, as well as arrays of excluded types. + * <p/> + * The {@link MarshalledValue} wrapper handles lazy deserialization from byte array representations. + * + * @author Manik Surtani (<a href="mailto:manik@jboss.org">manik@jboss.org</a>) + * @see MarshalledValue + * @since 2.1.0 + */ +public class MarshalledValueInterceptor extends CommandInterceptor +{ + @Override + public Object visitPutMapCommand(InvocationContext ctx, PutMapCommand command) throws Throwable + { + Set<MarshalledValue> marshalledValues = new HashSet<MarshalledValue>(); + + command.setMap(wrapMap(command.getMap(), marshalledValues, ctx)); + Object retVal = invokeNextInterceptor(ctx, command); + return compactAndProcessRetVal(marshalledValues, retVal); + } + + @Override + public Object visitPutKeyValueCommand(InvocationContext ctx, PutKeyValueCommand command) throws Throwable + { + Set<MarshalledValue> marshalledValues = new HashSet<MarshalledValue>(); + if (!MarshalledValueHelper.isTypeExcluded(command.getKey().getClass())) + { + Object newKey = createAndAddMarshalledValue(command.getKey(), marshalledValues, ctx); + command.setKey(newKey); + } + if (!MarshalledValueHelper.isTypeExcluded(command.getValue().getClass())) + { + Object value = createAndAddMarshalledValue(command.getValue(), marshalledValues, ctx); + command.setValue(value); + } + Object retVal = invokeNextInterceptor(ctx, command); + return compactAndProcessRetVal(marshalledValues, retVal); + } + + @Override + public Object visitRemoveCommand(InvocationContext ctx, RemoveCommand command) throws Throwable + { + Set<MarshalledValue> marshalledValues = new HashSet<MarshalledValue>(); + if (!MarshalledValueHelper.isTypeExcluded(command.getKey().getClass())) + { + Object value = createAndAddMarshalledValue(command.getKey(), marshalledValues, ctx); + command.setKey(value); + } + Object retVal = invokeNextInterceptor(ctx, command); + return compactAndProcessRetVal(marshalledValues, retVal); + } + + @Override + public Object visitGetKeyValueCommand(InvocationContext ctx, GetKeyValueCommand command) throws Throwable + { + Set<MarshalledValue> marshalledValues = new HashSet<MarshalledValue>(); + if (!MarshalledValueHelper.isTypeExcluded(command.getKey().getClass())) + { + Object value = createAndAddMarshalledValue(command.getKey(), marshalledValues, ctx); + command.setKey(value); + } + Object retVal = invokeNextInterceptor(ctx, command); + return compactAndProcessRetVal(marshalledValues, retVal); + } + + private Object compactAndProcessRetVal(Set<MarshalledValue> marshalledValues, Object retVal) + throws IOException, ClassNotFoundException + { + if (trace) log.trace("Compacting MarshalledValues created"); + for (MarshalledValue mv : marshalledValues) mv.compact(false, false); + + return processRetVal(retVal); + } + + private Object processRetVal(Object retVal) + throws IOException, ClassNotFoundException + { + if (retVal instanceof MarshalledValue) + { + if (trace) log.trace("Return value is a MarshalledValue. Unwrapping."); + retVal = ((MarshalledValue) retVal).get(); + } + return retVal; + } + + @SuppressWarnings("unchecked") + protected Map wrapMap(Map<Object, Object> m, Set<MarshalledValue> marshalledValues, InvocationContext ctx) throws NotSerializableException + { + if (m == null) + { + if (trace) log.trace("Map is nul; returning an empty map."); + return Collections.emptyMap(); + } + if (trace) log.trace("Wrapping map contents of argument " + m); + Map copy = new HashMap(); + for (Map.Entry me : m.entrySet()) + { + Object key = me.getKey(); + Object value = me.getValue(); + copy.put((key == null || MarshalledValueHelper.isTypeExcluded(key.getClass())) ? key : createAndAddMarshalledValue(key, marshalledValues, ctx), + (value == null || MarshalledValueHelper.isTypeExcluded(value.getClass())) ? value : createAndAddMarshalledValue(value, marshalledValues, ctx)); + } + return copy; + } + + protected MarshalledValue createAndAddMarshalledValue(Object toWrap, Set<MarshalledValue> marshalledValues, InvocationContext ctx) throws NotSerializableException + { + MarshalledValue mv = new MarshalledValue(toWrap); + marshalledValues.add(mv); + if (!ctx.isOriginLocal()) mv.setEqualityPreferenceForInstance(false); + return mv; + } +} Copied: core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/NotificationInterceptor.java (from rev 6896, core/branches/flat/src/main/java/org/jboss/cache/interceptors/NotificationInterceptor.java) =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/NotificationInterceptor.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/NotificationInterceptor.java 2008-10-09 13:45:15 UTC (rev 6898) @@ -0,0 +1,71 @@ +/* + * 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.interceptors; + +import org.jboss.starobrno.commands.tx.CommitCommand; +import org.jboss.starobrno.commands.tx.PrepareCommand; +import org.jboss.starobrno.commands.tx.RollbackCommand; +import org.jboss.starobrno.context.InvocationContext; +import org.jboss.starobrno.factories.annotations.Inject; +import org.jboss.starobrno.notifications.Notifier; + +/** + * The interceptor in charge of firing off notifications to cache listeners + * + * @author <a href="mailto:manik@jboss.org">Manik Surtani</a> + * @since 2.0.0 + */ +public class NotificationInterceptor extends BaseTransactionalContextInterceptor +{ + private Notifier notifier; + + @Inject + public void injectDependencies(Notifier notifier) + { + this.notifier = notifier; + } + + @Override + public Object visitPrepareCommand(InvocationContext ctx, PrepareCommand command) throws Throwable + { + Object retval = invokeNextInterceptor(ctx, command); + if (command.isOnePhaseCommit()) notifier.notifyTransactionCompleted(ctx.getTransaction(), true, ctx); + + return retval; + } + + @Override + public Object visitCommitCommand(InvocationContext ctx, CommitCommand command) throws Throwable + { + Object retval = invokeNextInterceptor(ctx, command); + notifier.notifyTransactionCompleted(ctx.getTransaction(), true, ctx); + return retval; + } + + @Override + public Object visitRollbackCommand(InvocationContext ctx, RollbackCommand command) throws Throwable + { + Object retval = invokeNextInterceptor(ctx, command); + notifier.notifyTransactionCompleted(ctx.getTransaction(), false, ctx); + return retval; + } +} Property changes on: core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/NotificationInterceptor.java ___________________________________________________________________ Name: svn:keywords + Author Date Id Revision Name: svn:eol-style + native Added: core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/base/JmxStatsCommandInterceptor.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/base/JmxStatsCommandInterceptor.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/interceptors/base/JmxStatsCommandInterceptor.java 2008-10-09 13:45:15 UTC (rev 6898) @@ -0,0 +1,87 @@ +/* + * 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.interceptors.base; + +import org.jboss.cache.jmx.JmxStatisticsExposer; +import org.jboss.starobrno.factories.annotations.Start; +import org.jboss.starobrno.jmx.annotations.ManagedAttribute; + +import java.util.Collections; +import java.util.Map; + +/** + * Base class for all the interceptors exposing management statistics. + * + * @author Mircea.Markus(a)jboss.com + * @since 3.0 + */ +public class JmxStatsCommandInterceptor extends CommandInterceptor implements JmxStatisticsExposer +{ + private boolean statsEnabled = false; + + @Start + public void checkStatisticsUsed() + { + setStatisticsEnabled(configuration.getExposeManagementStatistics()); + } + + /** + * Returns whether an interceptor's statistics are + * being captured. + * + * @return true if statistics are captured + */ + @ManagedAttribute + public boolean getStatisticsEnabled() + { + return statsEnabled; + } + + /** + * @param enabled whether gathering statistics for JMX are enabled. + */ + @ManagedAttribute + public void setStatisticsEnabled(boolean enabled) + { + statsEnabled = enabled; + } + + /** + * Returns a map of statistics. This is a default implementation which returns an empty map and should be overridden + * if it is to be meaningful. + * + * @return an empty map + */ + public Map<String, Object> dumpStatistics() + { + return Collections.emptyMap(); + } + + /** + * Resets statistics gathered. Is a no-op, and should be overridden if it is to be meaningful. + */ + public void resetStatistics() + { + } + + +} Copied: core/branches/flat/src/main/java/org/jboss/starobrno/marshall/MarshalledValue.java (from rev 6896, core/branches/flat/src/main/java/org/jboss/cache/marshall/MarshalledValue.java) =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/marshall/MarshalledValue.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/marshall/MarshalledValue.java 2008-10-09 13:45:15 UTC (rev 6898) @@ -0,0 +1,229 @@ +/* + * 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.marshall; + +import org.jboss.starobrno.CacheException; +import org.jboss.util.stream.MarshalledValueInputStream; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.Externalizable; +import java.io.IOException; +import java.io.NotSerializableException; +import java.io.ObjectInput; +import java.io.ObjectInputStream; +import java.io.ObjectOutput; +import java.io.ObjectOutputStream; +import java.io.Serializable; +import java.util.Arrays; + +/** + * Wrapper that wraps cached data, providing lazy deserialization using the calling thread's context class loader. + * <p/> + * The {@link org.jboss.starobrno.interceptors.MarshalledValueInterceptor} handles transparent + * wrapping/unwrapping of cached data. + * <p/> + * + * @author Manik Surtani (<a href="mailto:manik@jboss.org">manik@jboss.org</a>) + * @see org.jboss.starobrno.interceptors.MarshalledValueInterceptor + * @since 2.1.0 + */ +public class MarshalledValue implements Externalizable +{ + protected Object instance; + protected byte[] raw; + private int cachedHashCode = 0; + // by default equals() will test on the istance rather than the byte array if conversion is required. + private transient boolean equalityPreferenceForInstance = true; + + public MarshalledValue(Object instance) throws NotSerializableException + { + if (instance == null) throw new NullPointerException("Null values cannot be wrapped as MarshalledValues!"); + + if (instance instanceof Serializable) + this.instance = instance; + else + throw new NotSerializableException("Marshalled values can only wrap Objects that are serializable! Instance of " + instance.getClass() + " won't Serialize."); + } + + public MarshalledValue() + { + // empty ctor for serialization + } + + public void setEqualityPreferenceForInstance(boolean equalityPreferenceForInstance) + { + this.equalityPreferenceForInstance = equalityPreferenceForInstance; + } + + public synchronized void serialize() + { + if (raw == null) + { + try + { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(baos); + oos.writeObject(instance); + oos.close(); + baos.close(); + // Do NOT set instance to null over here, since it may be used elsewhere (e.g., in a cache listener). + // this will be compacted by the MarshalledValueInterceptor when the call returns. +// instance = null; + raw = baos.toByteArray(); + } + catch (Exception e) + { + throw new CacheException("Unable to marshall value " + instance, e); + } + } + } + + public synchronized void deserialize() + { + if (instance == null) + { + try + { + ByteArrayInputStream bais = new ByteArrayInputStream(raw); + // use a MarshalledValueInputStream since it needs to be aware of any context class loaders on the current thread. + ObjectInputStream ois = new MarshalledValueInputStream(bais); + instance = ois.readObject(); + ois.close(); + bais.close(); +// raw = null; + } + catch (Exception e) + { + throw new CacheException("Unable to unmarshall value", e); + } + } + } + + /** + * Compacts the references held by this class to a single reference. If only one representation exists this method + * is a no-op unless the 'force' parameter is used, in which case the reference held is forcefully switched to the + * 'preferred representation'. + * <p/> + * Either way, a call to compact() will ensure that only one representation is held. + * <p/> + * + * @param preferSerializedRepresentation if true and both representations exist, the serialized representation is favoured. If false, the deserialized representation is preferred. + * @param force ensures the preferred representation is maintained and the other released, even if this means serializing or deserializing. + */ + public void compact(boolean preferSerializedRepresentation, boolean force) + { + // reset the equalityPreference + equalityPreferenceForInstance = true; + if (force) + { + if (preferSerializedRepresentation && raw == null) serialize(); + else if (!preferSerializedRepresentation && instance == null) deserialize(); + } + + if (instance != null && raw != null) + { + // need to lose one representation! + + if (preferSerializedRepresentation) + { + instance = null; + } + else + { + raw = null; + } + } + } + + public void writeExternal(ObjectOutput out) throws IOException + { + if (raw == null) serialize(); + out.writeInt(raw.length); + out.write(raw); + out.writeInt(hashCode()); + } + + public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException + { + int size = in.readInt(); + raw = new byte[size]; + cachedHashCode = 0; + in.readFully(raw); + cachedHashCode = in.readInt(); + } + + public Object get() throws IOException, ClassNotFoundException + { + if (instance == null) deserialize(); + return instance; + } + + @Override + public boolean equals(Object o) + { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + MarshalledValue that = (MarshalledValue) o; + + // if both versions are serialized or deserialized, just compare the relevant representations. + if (raw != null && that.raw != null) return Arrays.equals(raw, that.raw); + if (instance != null && that.instance != null) return instance.equals(that.instance); + + // if conversion of one representation to the other is necessary, then see which we prefer converting. + if (equalityPreferenceForInstance) + { + if (instance == null) deserialize(); + if (that.instance == null) that.deserialize(); + return instance.equals(that.instance); + } + else + { + if (raw == null) serialize(); + if (that.raw == null) that.serialize(); + return Arrays.equals(raw, that.raw); + } + } + + @Override + public int hashCode() + { + if (cachedHashCode == 0) + { + // always calculate the hashcode based on the instance since this is where we're getting the equals() + if (instance == null) deserialize(); + cachedHashCode = instance.hashCode(); + if (cachedHashCode == 0) // degenerate case + { + cachedHashCode = 0xFEED; + } + } + return cachedHashCode; + } + + @Override + public String toString() + { + return "MarshalledValue(cachedHashCode=" + cachedHashCode + "; serialized=" + (raw != null) + ")"; + } +} Copied: core/branches/flat/src/main/java/org/jboss/starobrno/marshall/MarshalledValueHelper.java (from rev 6896, core/branches/flat/src/main/java/org/jboss/cache/marshall/MarshalledValueHelper.java) =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/marshall/MarshalledValueHelper.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/marshall/MarshalledValueHelper.java 2008-10-09 13:45:15 UTC (rev 6898) @@ -0,0 +1,54 @@ +/* + * 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.marshall; + +import org.jboss.starobrno.commands.ReplicableCommand; +import org.jboss.starobrno.transaction.GlobalTransaction; +import org.jgroups.Address; + +/** + * Common functionality used by the {@link org.jboss.starobrno.interceptors.MarshalledValueInterceptor} and the {@link MarshalledValueMap}. + * + * @author Manik Surtani (<a href="mailto:manik@jboss.org">manik@jboss.org</a>) + * @see MarshalledValue + * @see org.jboss.starobrno.interceptors.MarshalledValueInterceptor + * @see MarshalledValueMap + * @since 2.1.0 + */ +public class MarshalledValueHelper +{ + /** + * Tests whether the type should be excluded from MarshalledValue wrapping. + * + * @param type type to test. Should not be null. + * @return true if it should be excluded from MarshalledValue wrapping. + */ + public static boolean isTypeExcluded(Class type) + { + return type.equals(String.class) || type.isPrimitive() || + type.equals(Void.class) || type.equals(Boolean.class) || type.equals(Character.class) || + type.equals(Byte.class) || type.equals(Short.class) || type.equals(Integer.class) || + type.equals(Long.class) || type.equals(Float.class) || type.equals(Double.class) || + (type.isArray() && isTypeExcluded(type.getComponentType())) || type.equals(GlobalTransaction.class) || type.equals(Address.class) || + ReplicableCommand.class.isAssignableFrom(type) || type.equals(MarshalledValue.class); + } +} Copied: core/branches/flat/src/main/java/org/jboss/starobrno/marshall/MarshalledValueMap.java (from rev 6896, core/branches/flat/src/main/java/org/jboss/cache/marshall/MarshalledValueMap.java) =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/marshall/MarshalledValueMap.java (rev 0) +++ core/branches/flat/src/main/java/org/jboss/starobrno/marshall/MarshalledValueMap.java 2008-10-09 13:45:15 UTC (rev 6898) @@ -0,0 +1,181 @@ +/* + * 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.marshall; + +import net.jcip.annotations.Immutable; +import org.jboss.starobrno.CacheException; + +import java.io.Externalizable; +import java.io.IOException; +import java.io.ObjectInput; +import java.io.ObjectOutput; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; + +/** + * A Map that is able to wrap/unwrap MarshalledValues in keys or values. Note that calling keySet(), entrySet() or values() + * could be expensive if this map is large!! + * <p/> + * Also note that this is an immutable Map. + * <p/> + * + * @author Manik Surtani (<a href="mailto:manik@jboss.org">manik@jboss.org</a>) + * @see MarshalledValue + * @since 2.1.0 + */ +@Immutable +public class MarshalledValueMap implements Map, Externalizable +{ + Map delegate; + Map<Object, Object> unmarshalled; + + public MarshalledValueMap() + { + // for externalization + } + + public MarshalledValueMap(Map delegate) + { + this.delegate = delegate; + } + + @SuppressWarnings("unchecked") + protected synchronized Map getUnmarshalledMap() + { + if (unmarshalled == null) unmarshalled = unmarshalledMap(delegate.entrySet()); + return unmarshalled; + } + + public int size() + { + return delegate.size(); + } + + public boolean isEmpty() + { + return delegate.isEmpty(); + } + + public boolean containsKey(Object key) + { + return getUnmarshalledMap().containsKey(key); + } + + public boolean containsValue(Object value) + { + return getUnmarshalledMap().containsValue(value); + } + + public Object get(Object key) + { + return getUnmarshalledMap().get(key); + } + + public Object put(Object key, Object value) + { + throw new UnsupportedOperationException("This is an immutable map!"); + } + + public Object remove(Object key) + { + throw new UnsupportedOperationException("This is an immutable map!"); + } + + public void putAll(Map t) + { + throw new UnsupportedOperationException("This is an immutable map!"); + } + + public void clear() + { + throw new UnsupportedOperationException("This is an immutable map!"); + } + + public Set keySet() + { + return getUnmarshalledMap().keySet(); + } + + public Collection values() + { + return getUnmarshalledMap().values(); + } + + public Set entrySet() + { + return getUnmarshalledMap().entrySet(); + } + + @SuppressWarnings("unchecked") + protected Map unmarshalledMap(Set entries) + { + if (entries == null || entries.isEmpty()) return Collections.emptyMap(); + Map map = new HashMap(entries.size()); + for (Object e : entries) + { + Map.Entry entry = (Map.Entry) e; + map.put(getUnmarshalledValue(entry.getKey()), getUnmarshalledValue(entry.getValue())); + } + return map; + } + + private Object getUnmarshalledValue(Object o) + { + try + { + return o instanceof MarshalledValue ? ((MarshalledValue) o).get() : o; + } + catch (Exception e) + { + throw new CacheException("Unable to unmarshall value", e); + } + } + + @Override + public boolean equals(Object other) + { + if (other instanceof Map) + { + return getUnmarshalledMap().equals(other); + } + return false; + } + + @Override + public int hashCode() + { + return getUnmarshalledMap().hashCode(); + } + + public void writeExternal(ObjectOutput out) throws IOException + { + out.writeObject(delegate); + } + + public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException + { + delegate = (Map) in.readObject(); + } +} Modified: core/branches/flat/src/main/java/org/jboss/starobrno/notifications/NotifierImpl.java =================================================================== --- core/branches/flat/src/main/java/org/jboss/starobrno/notifications/NotifierImpl.java 2008-10-09 13:22:09 UTC (rev 6897) +++ core/branches/flat/src/main/java/org/jboss/starobrno/notifications/NotifierImpl.java 2008-10-09 13:45:15 UTC (rev 6898) @@ -25,7 +25,6 @@ import org.apache.commons.logging.LogFactory; import org.jboss.cache.Fqn; import org.jboss.cache.buddyreplication.BuddyGroup; -import org.jboss.cache.marshall.MarshalledValueMap; import org.jboss.cache.notifications.IncorrectCacheListenerException; import org.jboss.cache.notifications.annotation.*; import org.jboss.cache.notifications.event.*; @@ -43,6 +42,7 @@ import org.jboss.starobrno.factories.annotations.NonVolatile; import org.jboss.starobrno.factories.annotations.Start; import org.jboss.starobrno.factories.annotations.Stop; +import org.jboss.starobrno.marshall.MarshalledValueMap; import org.jboss.starobrno.notifications.event.NodeModifiedEvent; import org.jgroups.View;