[jbosscache-commits] JBoss Cache SVN: r7517 - in core/branches/flat/src: test/java/org/horizon/util and 1 other directories.

Author: manik.surtani(a)jboss.com Date: 2009-01-19 14:55:11 -0500 (Mon, 19 Jan 2009) New Revision: 7517 Added: core/branches/flat/src/test/java/org/horizon/util/internals/ core/branches/flat/src/test/java/org/horizon/util/internals/ReplicationListener.java Removed: core/branches/flat/src/main/java/org/horizon/util/internals/ core/branches/flat/src/test/java/org/horizon/util/internals/ReplicationListener.java Log: Moved test stuff Copied: core/branches/flat/src/test/java/org/horizon/util/internals (from rev 7514, core/branches/flat/src/main/java/org/horizon/util/internals) Deleted: core/branches/flat/src/test/java/org/horizon/util/internals/ReplicationListener.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/util/internals/ReplicationListener.java 2009-01-19 19:07:48 UTC (rev 7514) +++ core/branches/flat/src/test/java/org/horizon/util/internals/ReplicationListener.java 2009-01-19 19:55:11 UTC (rev 7517) @@ -1,254 +0,0 @@ -package org.horizon.util.internals; - -import org.horizon.Cache; -import org.horizon.commands.ReplicableCommand; -import org.horizon.commands.remote.ReplicateCommand; -import org.horizon.commands.tx.CommitCommand; -import org.horizon.commands.tx.PrepareCommand; -import org.horizon.config.Configuration; -import org.horizon.context.InvocationContext; -import org.horizon.factories.ComponentRegistry; -import org.horizon.io.ByteBuffer; -import org.horizon.logging.Log; -import org.horizon.logging.LogFactory; -import org.horizon.marshall.CacheMarshallerStarobrno; -import org.horizon.marshall.CommandAwareRpcDispatcher; -import org.horizon.marshall.Marshaller; -import org.horizon.remoting.RPCManager; -import org.horizon.util.TestingUtil; -import org.jgroups.blocks.RpcDispatcher; -import org.jgroups.util.Buffer; - -import java.io.InputStream; -import java.io.ObjectInputStream; -import java.io.ObjectOutputStream; -import java.util.Arrays; -import java.util.HashSet; -import java.util.Iterator; -import java.util.Set; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; - - -/* -* Utility class that notifies when certain commands were asynchronously replicated on secondary cache. - * Especially useful for avaoiding Thread.sleep() statements. - * <p/> - * Usage: - * <pre> - * Cache c1, c2; //these being two async caches - * AsyncReplicationListener listener2 = new AsyncReplicationListener(c2); - * listener2.expect(PutKeyValueCommand.class); - * c1.put(fqn, key, value); - * listener2.waitForReplicationToOccur(1000); // -this will block here untill c2 recieves the PutKeyValueCommand command - * </pre> - * Lifecycle - after being used (i.e. waitForReplicationToOccur returns sucessfully) the object returns to the - * non-initialized state and *can* be reused through expect-wait cycle. - * <b>Note</b>: this class might be used aswell for sync caches, e.g. a test could have subclasses which use sync and - * async replication - * - * @author Mircea.Markus(a)jboss.com - * @since 1.0 - */ -public class ReplicationListener { - private CountDownLatch latch = new CountDownLatch(1); - private Set<Class<? extends ReplicableCommand>> expectedCommands; - Configuration configuration; - private static final Log log = LogFactory.getLog(ReplicationListener.class); - - /** - * Builds a listener that will observe the given cache for recieving replication commands. - */ - public ReplicationListener(Cache cache) { - ComponentRegistry componentRegistry = TestingUtil.extractComponentRegistry(cache); - RPCManager rpcManager = componentRegistry.getComponent(RPCManager.class); - CommandAwareRpcDispatcher realDispatcher = (CommandAwareRpcDispatcher) TestingUtil.extractField(rpcManager, "rpcDispatcher"); - RpcDispatcher.Marshaller2 realMarshaller = (RpcDispatcher.Marshaller2) realDispatcher.getMarshaller(); - RpcDispatcher.Marshaller2 delegate = null; - delegate = new MarshallerDelegate(realMarshaller); - realDispatcher.setMarshaller(delegate); - realDispatcher.setRequestMarshaller(delegate); - realDispatcher.setResponseMarshaller(delegate); - configuration = cache.getConfiguration(); - } - - private class MarshallerDelegate implements RpcDispatcher.Marshaller2 { - RpcDispatcher.Marshaller2 marshaller; - - private MarshallerDelegate(RpcDispatcher.Marshaller2 marshaller) { - this.marshaller = marshaller; - } - - public byte[] objectToByteBuffer(Object obj) throws Exception { - return marshaller.objectToByteBuffer(obj); - } - - public Object objectFromByteBuffer(byte bytes[]) throws Exception { - Object result = marshaller.objectFromByteBuffer(bytes); - if (result instanceof ReplicateCommand && expectedCommands != null) { - ReplicateCommand replicateCommand = (ReplicateCommand) result; - return new ReplicateCommandDelegate(replicateCommand); - } - return result; - } - - public Buffer objectToBuffer(Object o) throws Exception { - return marshaller.objectToBuffer(o); - } - - public Object objectFromByteBuffer(byte[] bytes, int i, int i1) throws Exception { - Object result = marshaller.objectFromByteBuffer(bytes, i, i1); - if (result instanceof ReplicateCommand && expectedCommands != null) { - ReplicateCommand replicateCommand = (ReplicateCommand) result; - return new ReplicateCommandDelegate(replicateCommand); - } - return result; - } - } - - /** - * We want the notification to be performed only *after* the remote command is executed. - */ - private class ReplicateCommandDelegate extends ReplicateCommand { - ReplicateCommand realOne; - - private ReplicateCommandDelegate(ReplicateCommand realOne) { - this.realOne = realOne; - } - - @Override - public Object perform(InvocationContext ctx) throws Throwable { - try { - return realOne.perform(ctx); - } - finally { - log.trace("Processed command: " + realOne); - Iterator<Class<? extends ReplicableCommand>> it = expectedCommands.iterator(); - while (it.hasNext()) { - Class<? extends ReplicableCommand> replicableCommandClass = it.next(); - if (realOne.containsCommandType(replicableCommandClass)) { - it.remove(); - } else if (realOne.getSingleModification() instanceof PrepareCommand) //explicit transaction - { - PrepareCommand prepareCommand = (PrepareCommand) realOne.getSingleModification(); - if (prepareCommand.containsModificationType(replicableCommandClass)) { - it.remove(); - } - } - } - if (expectedCommands.isEmpty()) { - latch.countDown(); - } - } - } - } - - /** - * Needed for region based marshalling. - */ - private class RegionMarshallerDelegate extends CacheMarshallerStarobrno { - private Marshaller realOne; - - private RegionMarshallerDelegate(Marshaller realOne) { - this.realOne = realOne; - } - - @Override - public void objectToObjectStream(Object obj, ObjectOutputStream out) throws Exception { - realOne.objectToObjectStream(obj, out); - } - - @Override - public Object objectFromObjectStream(ObjectInputStream in) throws Exception { - return realOne.objectFromObjectStream(in); - } - - @Override - public Object objectFromStream(InputStream is) throws Exception { - return realOne.objectFromStream(is); - } - - public Object objectFromByteBuffer(byte[] bytes) throws Exception { - return this.objectFromByteBuffer(bytes, 0, bytes.length); - } - - - public ByteBuffer objectToBuffer(Object o) throws Exception { - return realOne.objectToBuffer(o); - } - - public Object objectFromByteBuffer(byte[] buffer, int i, int i1) throws Exception { - Object result = realOne.objectFromByteBuffer(buffer, i, i1); - if (result instanceof ReplicateCommand && expectedCommands != null) { - ReplicateCommand replicateCommand = (ReplicateCommand) result; - result = new ReplicateCommandDelegate(replicateCommand); - } - return result; - } - } - - /** - * Waits for 1 minute - */ - public void waitForReplicationToOccur() { - waitForReplicationToOccur(60000); - } - - /** - * Blocks for the elements specified through {@link #expect(Class[])} invocations to be replicated in this cache. if - * replication does not occur in the give timeout then an exception is being thrown. - */ - public void waitForReplicationToOccur(long timeoutMillis) { - log.trace("enter... ReplicationListener.waitForReplicationToOccur"); - waitForReplicationToOccur(timeoutMillis, TimeUnit.MILLISECONDS); - log.trace("exit... ReplicationListener.waitForReplicationToOccur"); - } - - /** - * Similar to {@link #waitForReplicationToOccur(long)} except that this method provides more flexibility in time - * units. - * - * @param timeout the maximum time to wait - * @param timeUnit the time unit of the <tt>timeout</tt> argument. - */ - public void waitForReplicationToOccur(long timeout, TimeUnit timeUnit) { - assert expectedCommands != null : "there are no replication expectations; please use AsyncReplicationListener.expect(...) before calling this method"; - try { - if (!latch.await(timeout, timeUnit)) { - assert false : "waiting for more than " + timeout + " " + timeUnit + " and following commands did not replicate: " + expectedCommands; - } - } - catch (InterruptedException e) { - throw new IllegalStateException("unexpected", e); - } - finally { - expectedCommands = null; - latch = new CountDownLatch(1); - } - } - - /** - * {@link #waitForReplicationToOccur(long)} will block untill all the commands specified here are being replicated to - * this cache. The method can be called several times with various arguments. - */ - public void expect(Class<? extends ReplicableCommand>... expectedCommands) { - if (this.expectedCommands == null) { - this.expectedCommands = new HashSet<Class<? extends ReplicableCommand>>(); - } - this.expectedCommands.addAll(Arrays.asList(expectedCommands)); - } - - /** - * Waits untill first command is replicated. - */ - public void expectAny() { - expect(); - } - - public void expectWithTx(Class<? extends ReplicableCommand>... writeCommands) { - expect(PrepareCommand.class); - //this is because for async replication we have an 1pc transaction - if (configuration.getCacheMode().isSynchronous()) expect(CommitCommand.class); - } - -} \ No newline at end of file Copied: core/branches/flat/src/test/java/org/horizon/util/internals/ReplicationListener.java (from rev 7515, core/branches/flat/src/main/java/org/horizon/util/internals/ReplicationListener.java) =================================================================== --- core/branches/flat/src/test/java/org/horizon/util/internals/ReplicationListener.java (rev 0) +++ core/branches/flat/src/test/java/org/horizon/util/internals/ReplicationListener.java 2009-01-19 19:55:11 UTC (rev 7517) @@ -0,0 +1,254 @@ +package org.horizon.util.internals; + +import org.horizon.Cache; +import org.horizon.commands.ReplicableCommand; +import org.horizon.commands.remote.ReplicateCommand; +import org.horizon.commands.tx.CommitCommand; +import org.horizon.commands.tx.PrepareCommand; +import org.horizon.config.Configuration; +import org.horizon.context.InvocationContext; +import org.horizon.factories.ComponentRegistry; +import org.horizon.io.ByteBuffer; +import org.horizon.logging.Log; +import org.horizon.logging.LogFactory; +import org.horizon.marshall.CacheMarshallerStarobrno; +import org.horizon.marshall.Marshaller; +import org.horizon.remoting.RPCManager; +import org.horizon.remoting.transport.jgroups.CommandAwareRpcDispatcher; +import org.horizon.util.TestingUtil; +import org.jgroups.blocks.RpcDispatcher; +import org.jgroups.util.Buffer; + +import java.io.InputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Iterator; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + + +/* +* Utility class that notifies when certain commands were asynchronously replicated on secondary cache. + * Especially useful for avaoiding Thread.sleep() statements. + * <p/> + * Usage: + * <pre> + * Cache c1, c2; //these being two async caches + * AsyncReplicationListener listener2 = new AsyncReplicationListener(c2); + * listener2.expect(PutKeyValueCommand.class); + * c1.put(fqn, key, value); + * listener2.waitForReplicationToOccur(1000); // -this will block here untill c2 recieves the PutKeyValueCommand command + * </pre> + * Lifecycle - after being used (i.e. waitForReplicationToOccur returns sucessfully) the object returns to the + * non-initialized state and *can* be reused through expect-wait cycle. + * <b>Note</b>: this class might be used aswell for sync caches, e.g. a test could have subclasses which use sync and + * async replication + * + * @author Mircea.Markus(a)jboss.com + * @since 1.0 + */ +public class ReplicationListener { + private CountDownLatch latch = new CountDownLatch(1); + private Set<Class<? extends ReplicableCommand>> expectedCommands; + Configuration configuration; + private static final Log log = LogFactory.getLog(ReplicationListener.class); + + /** + * Builds a listener that will observe the given cache for recieving replication commands. + */ + public ReplicationListener(Cache cache) { + ComponentRegistry componentRegistry = TestingUtil.extractComponentRegistry(cache); + RPCManager rpcManager = componentRegistry.getComponent(RPCManager.class); + CommandAwareRpcDispatcher realDispatcher = (CommandAwareRpcDispatcher) TestingUtil.extractField(rpcManager, "rpcDispatcher"); + RpcDispatcher.Marshaller2 realMarshaller = (RpcDispatcher.Marshaller2) realDispatcher.getMarshaller(); + RpcDispatcher.Marshaller2 delegate = null; + delegate = new MarshallerDelegate(realMarshaller); + realDispatcher.setMarshaller(delegate); + realDispatcher.setRequestMarshaller(delegate); + realDispatcher.setResponseMarshaller(delegate); + configuration = cache.getConfiguration(); + } + + private class MarshallerDelegate implements RpcDispatcher.Marshaller2 { + RpcDispatcher.Marshaller2 marshaller; + + private MarshallerDelegate(RpcDispatcher.Marshaller2 marshaller) { + this.marshaller = marshaller; + } + + public byte[] objectToByteBuffer(Object obj) throws Exception { + return marshaller.objectToByteBuffer(obj); + } + + public Object objectFromByteBuffer(byte bytes[]) throws Exception { + Object result = marshaller.objectFromByteBuffer(bytes); + if (result instanceof ReplicateCommand && expectedCommands != null) { + ReplicateCommand replicateCommand = (ReplicateCommand) result; + return new ReplicateCommandDelegate(replicateCommand); + } + return result; + } + + public Buffer objectToBuffer(Object o) throws Exception { + return marshaller.objectToBuffer(o); + } + + public Object objectFromByteBuffer(byte[] bytes, int i, int i1) throws Exception { + Object result = marshaller.objectFromByteBuffer(bytes, i, i1); + if (result instanceof ReplicateCommand && expectedCommands != null) { + ReplicateCommand replicateCommand = (ReplicateCommand) result; + return new ReplicateCommandDelegate(replicateCommand); + } + return result; + } + } + + /** + * We want the notification to be performed only *after* the remote command is executed. + */ + private class ReplicateCommandDelegate extends ReplicateCommand { + ReplicateCommand realOne; + + private ReplicateCommandDelegate(ReplicateCommand realOne) { + this.realOne = realOne; + } + + @Override + public Object perform(InvocationContext ctx) throws Throwable { + try { + return realOne.perform(ctx); + } + finally { + log.trace("Processed command: " + realOne); + Iterator<Class<? extends ReplicableCommand>> it = expectedCommands.iterator(); + while (it.hasNext()) { + Class<? extends ReplicableCommand> replicableCommandClass = it.next(); + if (realOne.containsCommandType(replicableCommandClass)) { + it.remove(); + } else if (realOne.getSingleModification() instanceof PrepareCommand) //explicit transaction + { + PrepareCommand prepareCommand = (PrepareCommand) realOne.getSingleModification(); + if (prepareCommand.containsModificationType(replicableCommandClass)) { + it.remove(); + } + } + } + if (expectedCommands.isEmpty()) { + latch.countDown(); + } + } + } + } + + /** + * Needed for region based marshalling. + */ + private class RegionMarshallerDelegate extends CacheMarshallerStarobrno { + private Marshaller realOne; + + private RegionMarshallerDelegate(Marshaller realOne) { + this.realOne = realOne; + } + + @Override + public void objectToObjectStream(Object obj, ObjectOutputStream out) throws Exception { + realOne.objectToObjectStream(obj, out); + } + + @Override + public Object objectFromObjectStream(ObjectInputStream in) throws Exception { + return realOne.objectFromObjectStream(in); + } + + @Override + public Object objectFromStream(InputStream is) throws Exception { + return realOne.objectFromStream(is); + } + + public Object objectFromByteBuffer(byte[] bytes) throws Exception { + return this.objectFromByteBuffer(bytes, 0, bytes.length); + } + + + public ByteBuffer objectToBuffer(Object o) throws Exception { + return realOne.objectToBuffer(o); + } + + public Object objectFromByteBuffer(byte[] buffer, int i, int i1) throws Exception { + Object result = realOne.objectFromByteBuffer(buffer, i, i1); + if (result instanceof ReplicateCommand && expectedCommands != null) { + ReplicateCommand replicateCommand = (ReplicateCommand) result; + result = new ReplicateCommandDelegate(replicateCommand); + } + return result; + } + } + + /** + * Waits for 1 minute + */ + public void waitForReplicationToOccur() { + waitForReplicationToOccur(60000); + } + + /** + * Blocks for the elements specified through {@link #expect(Class[])} invocations to be replicated in this cache. if + * replication does not occur in the give timeout then an exception is being thrown. + */ + public void waitForReplicationToOccur(long timeoutMillis) { + log.trace("enter... ReplicationListener.waitForReplicationToOccur"); + waitForReplicationToOccur(timeoutMillis, TimeUnit.MILLISECONDS); + log.trace("exit... ReplicationListener.waitForReplicationToOccur"); + } + + /** + * Similar to {@link #waitForReplicationToOccur(long)} except that this method provides more flexibility in time + * units. + * + * @param timeout the maximum time to wait + * @param timeUnit the time unit of the <tt>timeout</tt> argument. + */ + public void waitForReplicationToOccur(long timeout, TimeUnit timeUnit) { + assert expectedCommands != null : "there are no replication expectations; please use AsyncReplicationListener.expect(...) before calling this method"; + try { + if (!latch.await(timeout, timeUnit)) { + assert false : "waiting for more than " + timeout + " " + timeUnit + " and following commands did not replicate: " + expectedCommands; + } + } + catch (InterruptedException e) { + throw new IllegalStateException("unexpected", e); + } + finally { + expectedCommands = null; + latch = new CountDownLatch(1); + } + } + + /** + * {@link #waitForReplicationToOccur(long)} will block untill all the commands specified here are being replicated to + * this cache. The method can be called several times with various arguments. + */ + public void expect(Class<? extends ReplicableCommand>... expectedCommands) { + if (this.expectedCommands == null) { + this.expectedCommands = new HashSet<Class<? extends ReplicableCommand>>(); + } + this.expectedCommands.addAll(Arrays.asList(expectedCommands)); + } + + /** + * Waits untill first command is replicated. + */ + public void expectAny() { + expect(); + } + + public void expectWithTx(Class<? extends ReplicableCommand>... writeCommands) { + expect(PrepareCommand.class); + //this is because for async replication we have an 1pc transaction + if (configuration.getCacheMode().isSynchronous()) expect(CommitCommand.class); + } + +} \ No newline at end of file