[infinispan-commits] Infinispan SVN: r2061 - in trunk: cachestore/remote/src/test/java/org/infinispan/loaders/remote and 8 other directories.
infinispan-commits at lists.jboss.org
infinispan-commits at lists.jboss.org
Mon Jul 19 09:44:39 EDT 2010
Author: mircea.markus
Date: 2010-07-19 09:44:38 -0400 (Mon, 19 Jul 2010)
New Revision: 2061
Added:
trunk/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/impl/operations/BulkGetOperation.java
trunk/client/hotrod-client/src/test/java/org/infinispan/client/hotrod/BulkGetSimpleTest.java
Modified:
trunk/cachestore/remote/src/main/java/org/infinispan/loaders/remote/RemoteCacheStore.java
trunk/cachestore/remote/src/main/java/org/infinispan/loaders/remote/RemoteCacheStoreConfig.java
trunk/cachestore/remote/src/test/java/org/infinispan/loaders/remote/RemoteCacheStoreTest.java
trunk/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/RemoteCache.java
trunk/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/impl/RemoteCacheImpl.java
trunk/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/impl/operations/OperationsFactory.java
trunk/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/impl/protocol/HotRodConstants.java
trunk/client/hotrod-client/src/test/java/org/infinispan/client/hotrod/retry/ReplicationRetryTest.java
trunk/core/src/test/java/org/infinispan/loaders/BaseCacheStoreTest.java
trunk/server/hotrod/src/main/scala/org/infinispan/server/hotrod/Decoder10.scala
trunk/server/hotrod/src/main/scala/org/infinispan/server/hotrod/HotRodDecoder.scala
trunk/server/hotrod/src/main/scala/org/infinispan/server/hotrod/HotRodEncoder.scala
trunk/server/hotrod/src/main/scala/org/infinispan/server/hotrod/HotRodOperation.scala
trunk/server/hotrod/src/main/scala/org/infinispan/server/hotrod/Response.scala
Log:
merged 2051 - 2060 from trunk
Modified: trunk/cachestore/remote/src/main/java/org/infinispan/loaders/remote/RemoteCacheStore.java
===================================================================
--- trunk/cachestore/remote/src/main/java/org/infinispan/loaders/remote/RemoteCacheStore.java 2010-07-19 13:15:06 UTC (rev 2060)
+++ trunk/cachestore/remote/src/main/java/org/infinispan/loaders/remote/RemoteCacheStore.java 2010-07-19 13:44:38 UTC (rev 2061)
@@ -2,8 +2,10 @@
import net.jcip.annotations.ThreadSafe;
import org.infinispan.Cache;
+import org.infinispan.client.hotrod.RemoteCache;
import org.infinispan.client.hotrod.RemoteCacheManager;
import org.infinispan.container.entries.InternalCacheEntry;
+import org.infinispan.container.entries.InternalEntryFactory;
import org.infinispan.loaders.AbstractCacheStore;
import org.infinispan.loaders.CacheLoaderConfig;
import org.infinispan.loaders.CacheLoaderException;
@@ -13,8 +15,11 @@
import org.infinispan.util.logging.Log;
import org.infinispan.util.logging.LogFactory;
+import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
+import java.util.HashSet;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
@@ -30,10 +35,9 @@
* very costly operation as well). Purging takes place at the remote end (infinispan cluster).
* <p/>
*
+ * @author Mircea.Markus at jboss.com
* @see org.infinispan.loaders.remote.RemoteCacheStoreConfig
* @see <a href="http://community.jboss.org/wiki/JavaHotRodclient">Hotrod Java Client</a>
- *
- * @author Mircea.Markus at jboss.com
* @since 4.1
*/
@ThreadSafe
@@ -44,13 +48,13 @@
private volatile RemoteCacheStoreConfig config;
private volatile RemoteCacheManager remoteCacheManager;
- private volatile Cache<Object, InternalCacheEntry> remoteCache;
+ private volatile RemoteCache remoteCache;
private static final String LIFESPAN = "lifespan";
private static final String MAXIDLE = "maxidle";
@Override
public InternalCacheEntry load(Object key) throws CacheLoaderException {
- return remoteCache.get(key);
+ return (InternalCacheEntry) remoteCache.get(key);
}
@Override
@@ -61,6 +65,11 @@
}
@Override
+ public boolean containsKey(Object key) throws CacheLoaderException {
+ return remoteCache.containsKey(key);
+ }
+
+ @Override
public void store(InternalCacheEntry entry) throws CacheLoaderException {
if (log.isTraceEnabled()) {
log.trace("Adding entry: " + entry);
@@ -68,25 +77,25 @@
remoteCache.put(entry.getKey(), entry, toSeconds(entry.getLifespan(), entry, LIFESPAN), TimeUnit.SECONDS, toSeconds(entry.getMaxIdle(), entry, MAXIDLE), TimeUnit.SECONDS);
}
- private long toSeconds(long millis, InternalCacheEntry entry, String desc) {
- if (millis > 0 && millis < 1000) {
- if (log.isTraceEnabled()) {
- log.trace("Adjusting " + desc + " time for (k,v): (" + entry.getKey() + ", " + entry.getValue() + ") from "
- + millis + " millis to 1 sec, as milliseconds are not supported by HotRod");
- }
- return 1;
- }
- return TimeUnit.MILLISECONDS.toSeconds(millis);
- }
-
@Override
public void fromStream(ObjectInput inputStream) throws CacheLoaderException {
- fail();
+ Map result;
+ try {
+ result = (Map<Object, InternalCacheEntry>) marshaller.objectFromObjectStream(inputStream);
+ remoteCache.putAll(result);
+ } catch (Exception e) {
+ throw new CacheLoaderException("Exception while reading data", e);
+ }
}
@Override
public void toStream(ObjectOutput outputStream) throws CacheLoaderException {
- fail();
+ Map map = remoteCache.getBulk();
+ try {
+ marshaller.objectToObjectStream(map, outputStream);
+ } catch (IOException e) {
+ throw new CacheLoaderException("Exception while serializing remote data to stream", e);
+ }
}
@Override
@@ -101,24 +110,18 @@
@Override
public Set<InternalCacheEntry> loadAll() throws CacheLoaderException {
- fail();
- return null;
+ Map map = remoteCache.getBulk();
+ return convertToInternalCacheEntries(map);
}
@Override
public Set<InternalCacheEntry> load(int numEntries) throws CacheLoaderException {
- fail();
- return null;
+ return convertToInternalCacheEntries(remoteCache.getBulk(numEntries));
}
@Override
public Set<Object> loadAllKeys(Set<Object> keysToExclude) throws CacheLoaderException {
- fail();
- return null;
- }
-
- private void fail() throws CacheLoaderException {
- String message = "RemoteCacheStore can only run in shared mode and it doesn't support preload!";
+ String message = "RemoteCacheStore can only run in shared mode! This method shouldn't be called in shared mode";
log.error(message);
throw new CacheLoaderException(message);
}
@@ -151,4 +154,24 @@
public Class<? extends CacheLoaderConfig> getConfigurationClass() {
return RemoteCacheStoreConfig.class;
}
+
+ private long toSeconds(long millis, InternalCacheEntry entry, String desc) {
+ if (millis > 0 && millis < 1000) {
+ if (log.isTraceEnabled()) {
+ log.trace("Adjusting " + desc + " time for (k,v): (" + entry.getKey() + ", " + entry.getValue() + ") from "
+ + millis + " millis to 1 sec, as milliseconds are not supported by HotRod");
+ }
+ return 1;
+ }
+ return TimeUnit.MILLISECONDS.toSeconds(millis);
+ }
+
+ private Set<InternalCacheEntry> convertToInternalCacheEntries(Map map) {
+ Set<InternalCacheEntry> result = new HashSet<InternalCacheEntry>(map.size());
+ Set<Map.Entry> set = map.entrySet();
+ for (Map.Entry e : set) {
+ result.add((InternalCacheEntry) e.getValue());
+ }
+ return result;
+ }
}
Modified: trunk/cachestore/remote/src/main/java/org/infinispan/loaders/remote/RemoteCacheStoreConfig.java
===================================================================
--- trunk/cachestore/remote/src/main/java/org/infinispan/loaders/remote/RemoteCacheStoreConfig.java 2010-07-19 13:15:06 UTC (rev 2060)
+++ trunk/cachestore/remote/src/main/java/org/infinispan/loaders/remote/RemoteCacheStoreConfig.java 2010-07-19 13:44:38 UTC (rev 2061)
@@ -51,27 +51,6 @@
return CacheContainer.DEFAULT_CACHE_NAME.equals(getRemoteCacheName());
}
- @Override
- public void setPurgeOnStartup(Boolean purgeOnStartup) {
- super.setPurgeOnStartup(purgeOnStartup);
- if (purgeOnStartup) {
- log.info("Purge on start-up will be ignored; remote cache store cannot be purged.");
- }
- }
-
- @Override
- public void setFetchPersistentState(Boolean fetchPersistentState) {
- if (fetchPersistentState) {
- String message = "fetchPersistentState cannot be set for remote cache store. This is because " +
- "persistent state cannot be generated by RemoteCacheStore, and state generation is proprietary. I.e. this " +
- "store would not be able to integrate state generated by other store implementations.";
- log.error(message);
- throw new IllegalStateException(message);
- } else {
- super.setFetchPersistentState(fetchPersistentState);
- }
- }
-
public Properties getHotRodClientProperties() {
return hotRodClientProperties;
}
Modified: trunk/cachestore/remote/src/test/java/org/infinispan/loaders/remote/RemoteCacheStoreTest.java
===================================================================
--- trunk/cachestore/remote/src/test/java/org/infinispan/loaders/remote/RemoteCacheStoreTest.java 2010-07-19 13:15:06 UTC (rev 2060)
+++ trunk/cachestore/remote/src/test/java/org/infinispan/loaders/remote/RemoteCacheStoreTest.java 2010-07-19 13:44:38 UTC (rev 2061)
@@ -30,7 +30,7 @@
RemoteCacheStoreConfig remoteCacheStoreConfig = new RemoteCacheStoreConfig();
remoteCacheStoreConfig.setUseDefaultRemoteCache(true);
assert remoteCacheStoreConfig.isUseDefaultRemoteCache();
-
+
localCacheManager = TestCacheManagerFactory.createLocalCacheManager();
Configuration configuration = localCacheManager.getDefaultConfiguration();
configuration.setEvictionWakeUpInterval(10);
@@ -43,7 +43,6 @@
RemoteCacheStore remoteCacheStore = new RemoteCacheStore();
remoteCacheStore.init(remoteCacheStoreConfig, getCache(), getMarshaller());
remoteCacheStore.start();
- super.supportsLoadAll = false;
return remoteCacheStore;
}
@@ -54,42 +53,28 @@
}
@Override
- public void testLoadKeys() throws CacheLoaderException {
- //not applicable as relies on loadAll
+ protected void assertEventuallyExpires(String key) throws Exception {
+ for (int i = 0; i < 10; i++) {
+ if (cs.load("k") == null) break;
+ Thread.sleep(1000);
+ }
+ assert cs.load("k") == null;
}
@Override
- protected void purgeExpired() throws CacheLoaderException {
- localCacheManager.getCache().clear();
- }
-
- @Override
- public void testPreload() throws CacheLoaderException {
- //not applicable as relies on loadAll
- }
-
- @Override
- public void testPreloadWithMaxSize() throws CacheLoaderException {
- //not applicable as relies on loadAll
- }
-
protected void sleepForStopStartTest() throws InterruptedException {
Thread.sleep(3000);
}
-
@Override
- public void testStoreAndRemoveAll() throws CacheLoaderException {
- //not applicable as relies on loadAll
+ protected void purgeExpired() throws CacheLoaderException {
+ localCacheManager.getCache().getAdvancedCache().getEvictionManager().processEviction();
}
+ /**
+ * This is not supported, see assertion in {@link RemoteCacheStore#loadAllKeys(java.util.Set)}
+ */
@Override
- public void testStreamingAPI() throws IOException, ClassNotFoundException, CacheLoaderException {
- //not applicable as relies on loadAll
+ public void testLoadKeys() throws CacheLoaderException {
}
-
- @Override
- public void testStreamingAPIReusingStreams() throws IOException, ClassNotFoundException, CacheLoaderException {
- //not applicable as relies on loadAll
- }
}
Modified: trunk/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/RemoteCache.java
===================================================================
--- trunk/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/RemoteCache.java 2010-07-19 13:15:06 UTC (rev 2060)
+++ trunk/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/RemoteCache.java 2010-07-19 13:44:38 UTC (rev 2061)
@@ -14,42 +14,41 @@
* Provides remote reference to a Hot Rod server/cluster. It implements {@link org.infinispan.Cache}, but given its
* nature (remote) some operations are not supported. All these unsupported operations are being overridden within this
* interface and documented as such.
- * <p>
+ * <p/>
* <b>New operations</b>: besides the operations inherited from {@link org.infinispan.Cache}, RemoteCache also adds new
* operations to optimize/reduce network traffic: e.g. versioned put operation.
- * <p>
+ * <p/>
* <b>Concurrency</b>: implementors of this interface will support multi-threaded access, similar to the way {@link
* org.infinispan.Cache} supports it.
- * <p>
- * <b>Return values</b>: previously existing values for certain {@link java.util.Map} operations are not returned,
- * null is returned instead. E.g. {@link java.util.Map#put(Object, Object)} returns the previous value
- * associated to the supplied key. In case of RemoteCache, this returns null.
- * <p>
- * <b>Synthetic operations</b>: aggregate operations are being implemented based on other Hot Rod operations.
- * E.g. all the {@link java.util.Map#putAll(java.util.Map)} is implemented through multiple individual puts. This means
- * that the these operations are not atomic and that they are costly, e.g. as the number of network round-trips is not
- * one, but the size of the added map. All these synthetic operations are documented as such.
- * <p>
- * <b>changing default behavior through {@link org.infinispan.client.hotrod.Flag}s</b>: it is possible to change de default cache behaviour by using
- * flags on an per invocation basis.
- * E.g.
+ * <p/>
+ * <b>Return values</b>: previously existing values for certain {@link java.util.Map} operations are not returned, null
+ * is returned instead. E.g. {@link java.util.Map#put(Object, Object)} returns the previous value associated to the
+ * supplied key. In case of RemoteCache, this returns null.
+ * <p/>
+ * <b>Synthetic operations</b>: aggregate operations are being implemented based on other Hot Rod operations. E.g. all
+ * the {@link java.util.Map#putAll(java.util.Map)} is implemented through multiple individual puts. This means that the
+ * these operations are not atomic and that they are costly, e.g. as the number of network round-trips is not one, but
+ * the size of the added map. All these synthetic operations are documented as such.
+ * <p/>
+ * <b>changing default behavior through {@link org.infinispan.client.hotrod.Flag}s</b>: it is possible to change de
+ * default cache behaviour by using flags on an per invocation basis. E.g.
* <pre>
* RemoteCache cache = getRemoteCache();
* Object value = cache.withFlags(Flag.FORCE_RETURN_VALUE).get(aKey);
* </pre>
- * In the previous example, using {@link org.infinispan.client.hotrod.Flag#FORCE_RETURN_VALUE} will make the client to also return previously
- * existing value associated with <tt>aKey</tt>. If this flag would not be present, Infinispan would return (by default)
- * <tt>null</tt>. This is in order to avoid fetching a possibly large object from the remote server, which might not be
- * needed. The flags as set by the {@link org.infinispan.client.hotrod.RemoteCache#withFlags(Flag...)} operation only apply for the very next
- * operation executed <b>by the same thread</b> on the RemoteCache.
- * <p>
- * <b><a href="http://community.jboss.org/wiki/Eviction">Eviction and expiration</a></b>:
- * Unlike local {@link org.infinispan.Cache} cache, which allows specifying time values with any granularity (as defined by {@link TimeUnit}),
- * HotRod only supports seconds as time units. If a different time unit is used instead, HotRod will transparently convert it to
- * seconds, using {@link java.util.concurrent.TimeUnit#toSeconds(long)} method. This might result in loss of precision for
- * values specified as nanos or milliseconds. <br/>
- * Another fundamental difference is in the case of lifespan (naturally does NOT apply for max idle): If number of seconds is bigger than 30 days,
- * this number of seconds is treated as UNIX time and so, represents the number of seconds since 1/1/1970. <br/>
+ * In the previous example, using {@link org.infinispan.client.hotrod.Flag#FORCE_RETURN_VALUE} will make the client to
+ * also return previously existing value associated with <tt>aKey</tt>. If this flag would not be present, Infinispan
+ * would return (by default) <tt>null</tt>. This is in order to avoid fetching a possibly large object from the remote
+ * server, which might not be needed. The flags as set by the {@link org.infinispan.client.hotrod.RemoteCache#withFlags(Flag...)}
+ * operation only apply for the very next operation executed <b>by the same thread</b> on the RemoteCache.
+ * <p/>
+ * <b><a href="http://community.jboss.org/wiki/Eviction">Eviction and expiration</a></b>: Unlike local {@link
+ * org.infinispan.Cache} cache, which allows specifying time values with any granularity (as defined by {@link
+ * TimeUnit}), HotRod only supports seconds as time units. If a different time unit is used instead, HotRod will
+ * transparently convert it to seconds, using {@link java.util.concurrent.TimeUnit#toSeconds(long)} method. This might
+ * result in loss of precision for values specified as nanos or milliseconds. <br/> Another fundamental difference is in
+ * the case of lifespan (naturally does NOT apply for max idle): If number of seconds is bigger than 30 days, this
+ * number of seconds is treated as UNIX time and so, represents the number of seconds since 1/1/1970. <br/>
*
* @author Mircea.Markus at jboss.com
* @since 4.1
@@ -63,7 +62,8 @@
* //some processing
* remoteCache.removeWithVersion(key, ve.getVersion();
* </pre>
- * Lat call (removeWithVersion) will make sure that the entry will only be removed if it hasn't been changed in between.
+ * Lat call (removeWithVersion) will make sure that the entry will only be removed if it hasn't been changed in
+ * between.
*
* @return true if the entry has been removed
* @see VersionedValue
@@ -77,8 +77,8 @@
NotifyingFuture<Boolean> removeWithVersionAsync(K key, long version);
/**
- * Removes the given value only if its version matches the supplied version. See {@link #removeWithVersion(Object, long)} for a
- * sample usage.
+ * Removes the given value only if its version matches the supplied version. See {@link #removeWithVersion(Object,
+ * long)} for a sample usage.
*
* @return true if the method has been replaced
* @see #getVersioned(Object)
@@ -249,8 +249,7 @@
NotifyingFuture<Boolean> replaceAsync(K key, V oldValue, V newValue);
/**
- * This operation is not supported. Consider using {@link #replaceAsync(K,V,long,int)}
- * instead.
+ * This operation is not supported. Consider using {@link #replaceAsync(K,V,long,int)} instead.
*
* @throws UnsupportedOperationException
*/
@@ -339,10 +338,25 @@
public ServerStatistics stats();
- RemoteCache<K,V> withFlags(Flag... flags);
+ RemoteCache<K, V> withFlags(Flag... flags);
/**
* Returns the {@link org.infinispan.client.hotrod.RemoteCacheManager} that created this cache.
*/
public RemoteCacheManager getRemoteCacheManager();
+
+ /**
+ * Bulk get operations, returns all the entries within the remote cache.
+ *
+ * @return the returned values depend on the configuration of the back-end infinispan servers. Read <a
+ * href="http://community.jboss.org/wiki/HotRodBulkGet-Design#Server_side">this</a> for more details. The
+ * returned Map is unmodifiable.
+ */
+ public Map<K, V> getBulk();
+
+ /**
+ * Same as {@link #getBulk()}, but limits the returned set of values to the specified size. No ordering is guaranteed, and there is no
+ * guarantee that "size" elements are returned( e.g. if the number of elements in the back-end server is smaller that "size")
+ */
+ public Map<K, V> getBulk(int size);
}
Modified: trunk/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/impl/RemoteCacheImpl.java
===================================================================
--- trunk/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/impl/RemoteCacheImpl.java 2010-07-19 13:15:06 UTC (rev 2060)
+++ trunk/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/impl/RemoteCacheImpl.java 2010-07-19 13:44:38 UTC (rev 2061)
@@ -16,6 +16,8 @@
import org.infinispan.util.logging.LogFactory;
import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
@@ -279,10 +281,33 @@
byte[] keyBytes = obj2bytes(key, true);
GetOperation gco = operationsFactory.newGetKeyOperation(keyBytes);
byte[] bytes = (byte[]) gco.execute();
- return (V) bytes2obj(bytes);
+ V result = (V) bytes2obj(bytes);
+ if (log.isTraceEnabled()) {
+ log.trace("For key(" + key + ") returning " + result);
+ }
+ return result;
}
@Override
+ public Map<K, V> getBulk() {
+ return getBulk(0);
+ }
+
+ @Override
+ public Map<K, V> getBulk(int size) {
+ assertRemoteCacheManagerIsStarted();
+ BulkGetOperation op = operationsFactory.newBulkGetOperation(size);
+ Map<byte[], byte[]> result = (Map) op.execute();
+ Map<K,V> toReturn = new HashMap<K,V>();
+ for (Map.Entry<byte[], byte[]> entry : result.entrySet()) {
+ V value = (V) bytes2obj(entry.getValue());
+ K key = (K) bytes2obj(entry.getKey());
+ toReturn.put(key, value);
+ }
+ return Collections.unmodifiableMap(toReturn);
+ }
+
+ @Override
public V remove(Object key) {
assertRemoteCacheManagerIsStarted();
RemoveOperation removeOperation = operationsFactory.newRemoveOperation(obj2bytes(key, true));
Copied: trunk/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/impl/operations/BulkGetOperation.java (from rev 2059, branches/4.1.x/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/impl/operations/BulkGetOperation.java)
===================================================================
--- trunk/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/impl/operations/BulkGetOperation.java (rev 0)
+++ trunk/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/impl/operations/BulkGetOperation.java 2010-07-19 13:44:38 UTC (rev 2061)
@@ -0,0 +1,42 @@
+package org.infinispan.client.hotrod.impl.operations;
+
+import org.infinispan.client.hotrod.Flag;
+import org.infinispan.client.hotrod.impl.transport.Transport;
+import org.infinispan.client.hotrod.impl.transport.TransportFactory;
+
+import java.util.HashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Reads more keys at a time. Specified <a href="http://community.jboss.org/wiki/HotRodBulkGet-Design">here</a>.
+ *
+ * @author Mircea.Markus at jboss.com
+ * @since 4.1
+ */
+public class BulkGetOperation extends RetryOnFailureOperation {
+
+ private final int entryCount;
+
+ public BulkGetOperation(TransportFactory transportFactory, byte[] cacheName, AtomicInteger topologyId, Flag[] flags, int entryCount) {
+ super(transportFactory, cacheName, topologyId, flags);
+ this.entryCount = entryCount;
+ }
+
+ @Override
+ protected Transport getTransport(int retryCount) {
+ return transportFactory.getTransport();
+ }
+
+ @Override
+ protected Object executeOperation(Transport transport) {
+ long messageId = writeHeader(transport, BULK_GET_REQUEST);
+ transport.writeVInt(entryCount);
+ transport.flush();
+ readHeaderAndValidate(transport, messageId, BULK_GET_RESPONSE);
+ HashMap result = new HashMap();
+ while ( transport.readByte() == 1) { //there's more!
+ result.put(transport.readArray(), transport.readArray());
+ }
+ return result;
+ }
+}
Modified: trunk/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/impl/operations/OperationsFactory.java
===================================================================
--- trunk/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/impl/operations/OperationsFactory.java 2010-07-19 13:15:06 UTC (rev 2060)
+++ trunk/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/impl/operations/OperationsFactory.java 2010-07-19 13:44:38 UTC (rev 2061)
@@ -80,6 +80,10 @@
return new ClearOperation(transportFactory, cacheNameBytes, topologyId, flags());
}
+ public BulkGetOperation newBulkGetOperation(int size) {
+ return new BulkGetOperation(transportFactory, cacheNameBytes, topologyId, flags(), size);
+ }
+
private Flag[] flags() {
Flag[] flags = this.flagsMap.get();
this.flagsMap.remove();
Modified: trunk/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/impl/protocol/HotRodConstants.java
===================================================================
--- trunk/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/impl/protocol/HotRodConstants.java 2010-07-19 13:15:06 UTC (rev 2060)
+++ trunk/client/hotrod-client/src/main/java/org/infinispan/client/hotrod/impl/protocol/HotRodConstants.java 2010-07-19 13:44:38 UTC (rev 2061)
@@ -28,6 +28,7 @@
static final byte CLEAR_REQUEST = 0x13;
static final byte STATS_REQUEST = 0x15;
static final byte PING_REQUEST = 0x17;
+ static final byte BULK_GET_REQUEST = 0x19;
//responses
@@ -43,6 +44,7 @@
static final byte CLEAR_RESPONSE = 0x14;
static final byte STATS_RESPONSE = 0x16;
static final byte PING_RESPONSE = 0x18;
+ static final byte BULK_GET_RESPONSE = 0x20;
static final byte ERROR_RESPONSE = 0x50;
//response status
Copied: trunk/client/hotrod-client/src/test/java/org/infinispan/client/hotrod/BulkGetSimpleTest.java (from rev 2059, branches/4.1.x/client/hotrod-client/src/test/java/org/infinispan/client/hotrod/BulkGetSimpleTest.java)
===================================================================
--- trunk/client/hotrod-client/src/test/java/org/infinispan/client/hotrod/BulkGetSimpleTest.java (rev 0)
+++ trunk/client/hotrod-client/src/test/java/org/infinispan/client/hotrod/BulkGetSimpleTest.java 2010-07-19 13:44:38 UTC (rev 2061)
@@ -0,0 +1,70 @@
+package org.infinispan.client.hotrod;
+
+import org.infinispan.manager.EmbeddedCacheManager;
+import org.infinispan.server.hotrod.HotRodServer;
+import org.infinispan.test.SingleCacheManagerTest;
+import org.infinispan.test.fwk.TestCacheManagerFactory;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.Test;
+
+import java.util.Map;
+import java.util.Properties;
+
+import static org.testng.AssertJUnit.assertEquals;
+
+/**
+ * @author Mircea.Markus at jboss.com
+ * @since 4.1
+ */
+ at Test(testName = "client.hotrod.BulkGetSimpleTest", groups = "functional")
+public class BulkGetSimpleTest extends SingleCacheManagerTest {
+ private HotRodServer hotRodServer;
+ private RemoteCacheManager remoteCacheManager;
+ private RemoteCache<Object, Object> remoteCache;
+
+ @Override
+ protected EmbeddedCacheManager createCacheManager() throws Exception {
+ cacheManager = TestCacheManagerFactory.createLocalCacheManager();
+ cache = cacheManager.getCache();
+
+ hotRodServer = TestHelper.startHotRodServer(cacheManager);
+
+ Properties hotrodClientConf = new Properties();
+ hotrodClientConf.put("infinispan.client.hotrod.server_list", "localhost:" + hotRodServer.getPort());
+ remoteCacheManager = new RemoteCacheManager(hotrodClientConf);
+ remoteCache = remoteCacheManager.getCache();
+ populateCacheManager();
+ return cacheManager;
+ }
+
+ @AfterMethod
+ @Override
+ protected void clearContent() {
+
+ }
+
+ private void populateCacheManager() {
+ for (int i = 0; i < 100; i++) {
+ remoteCache.put(i, i);
+ }
+ }
+
+ public void testBulkGet() {
+ Map<Object,Object> map = remoteCache.getBulk();
+ assert map.size() == 100;
+ for (int i = 0; i < 100; i++) {
+ assert map.get(i).equals(i);
+ }
+ }
+
+ public void testBulkGetWithSize() {
+ Map<Object,Object> map = remoteCache.getBulk(50);
+ assertEquals(50, map.size());
+ for (int i = 0; i < 100; i++) {
+ if (map.containsKey(i)) {
+ Integer value = (Integer) map.get(i);
+ assertEquals((Integer)i, value);
+ }
+ }
+ }
+}
Modified: trunk/client/hotrod-client/src/test/java/org/infinispan/client/hotrod/retry/ReplicationRetryTest.java
===================================================================
--- trunk/client/hotrod-client/src/test/java/org/infinispan/client/hotrod/retry/ReplicationRetryTest.java 2010-07-19 13:15:06 UTC (rev 2060)
+++ trunk/client/hotrod-client/src/test/java/org/infinispan/client/hotrod/retry/ReplicationRetryTest.java 2010-07-19 13:44:38 UTC (rev 2061)
@@ -16,6 +16,7 @@
import org.testng.annotations.Test;
import java.net.InetSocketAddress;
+import java.util.Map;
import java.util.Properties;
import static org.testng.Assert.assertEquals;
@@ -99,6 +100,13 @@
assertEquals(false, remoteCache.containsKey("k"));
}
+ public void testBulkGet() {
+ validateSequenceAndStopServer();
+ resetStats();
+ Map map = remoteCache.getBulk();
+ assertEquals(3, map.size());
+ }
+
private void validateSequenceAndStopServer() {
resetStats();
assertNoHits();
Modified: trunk/core/src/test/java/org/infinispan/loaders/BaseCacheStoreTest.java
===================================================================
--- trunk/core/src/test/java/org/infinispan/loaders/BaseCacheStoreTest.java 2010-07-19 13:15:06 UTC (rev 2060)
+++ trunk/core/src/test/java/org/infinispan/loaders/BaseCacheStoreTest.java 2010-07-19 13:44:38 UTC (rev 2061)
@@ -36,9 +36,9 @@
import static java.util.Collections.emptySet;
/**
- * This is a base class containing various unit tests for each and every different CacheStore implementations.
- * If you need to add Cache/CacheManager tests that need to be run for each cache store/loader implementation,
- * then use BaseCacheStoreFunctionalTest.
+ * This is a base class containing various unit tests for each and every different CacheStore implementations. If you
+ * need to add Cache/CacheManager tests that need to be run for each cache store/loader implementation, then use
+ * BaseCacheStoreFunctionalTest.
*/
@SuppressWarnings("unchecked")
// this needs to be here for the test to run in an IDE
@@ -51,8 +51,6 @@
protected GlobalTransactionFactory gtf = new GlobalTransactionFactory();
- protected volatile boolean supportsLoadAll = true;
-
@BeforeMethod
public void setUp() throws Exception {
try {
@@ -76,7 +74,7 @@
}
}
- @AfterMethod (alwaysRun = false)
+ @AfterMethod(alwaysRun = false)
public void assertNoLocksHeld() {
//doesn't really make sense to add a subclass for this check only
if (cs instanceof LockSupportCacheStore) {
@@ -113,7 +111,7 @@
assert cs.containsKey("k");
}
- public void testLoadAndStoreWithLifespan() throws InterruptedException, CacheLoaderException {
+ public void testLoadAndStoreWithLifespan() throws Exception {
assert !cs.containsKey("k");
long lifespan = 120000;
@@ -123,10 +121,8 @@
assert cs.containsKey("k");
InternalCacheEntry ice = cs.load("k");
assertCorrectExpiry(ice, "v", lifespan, -1, false);
- if (supportsLoadAll) {
- ice = cs.loadAll().iterator().next();
- assertCorrectExpiry(ice, "v", lifespan, -1, false);
- }
+ ice = cs.loadAll().iterator().next();
+ assertCorrectExpiry(ice, "v", lifespan, -1, false);
lifespan = 1;
se = InternalEntryFactory.create("k", "v", lifespan);
@@ -134,11 +130,9 @@
Thread.sleep(100);
purgeExpired();
assert se.isExpired();
- assert cs.load("k") == null;
+ assertEventuallyExpires("k");
assert !cs.containsKey("k");
- if (supportsLoadAll) {
- assert cs.loadAll().isEmpty();
- }
+ assert cs.loadAll().isEmpty();
}
private void assertCorrectExpiry(InternalCacheEntry ice, String value, long lifespan, long maxIdle, boolean expired) {
@@ -152,7 +146,7 @@
}
- public void testLoadAndStoreWithIdle() throws InterruptedException, CacheLoaderException {
+ public void testLoadAndStoreWithIdle() throws Exception {
assert !cs.containsKey("k");
long idle = 120000;
@@ -162,10 +156,8 @@
assert cs.containsKey("k");
InternalCacheEntry ice = cs.load("k");
assertCorrectExpiry(ice, "v", -1, idle, false);
- if (supportsLoadAll) {
- ice = cs.loadAll().iterator().next();
- assertCorrectExpiry(ice, "v", -1, idle, false);
- }
+ ice = cs.loadAll().iterator().next();
+ assertCorrectExpiry(ice, "v", -1, idle, false);
idle = 1;
se = InternalEntryFactory.create("k", "v", -1, idle);
@@ -173,18 +165,20 @@
Thread.sleep(100);
purgeExpired();
assert se.isExpired();
- assert cs.load("k") == null;
+ assertEventuallyExpires("k");
assert !cs.containsKey("k");
- if (supportsLoadAll) {
- assert cs.loadAll().isEmpty();
- }
+ assert cs.loadAll().isEmpty();
}
+ protected void assertEventuallyExpires(String key) throws Exception {
+ assert cs.load(key) == null;
+ }
+
protected void purgeExpired() throws CacheLoaderException {
cs.purgeExpired();
}
- public void testLoadAndStoreWithLifespanAndIdle() throws InterruptedException, CacheLoaderException {
+ public void testLoadAndStoreWithLifespanAndIdle() throws Exception {
assert !cs.containsKey("k");
long lifespan = 200000;
@@ -195,10 +189,8 @@
assert cs.containsKey("k");
InternalCacheEntry ice = cs.load("k");
assertCorrectExpiry(ice, "v", lifespan, idle, false);
- if (supportsLoadAll) {
- ice = cs.loadAll().iterator().next();
- assertCorrectExpiry(ice, "v", lifespan, idle, false);
- }
+ ice = cs.loadAll().iterator().next();
+ assertCorrectExpiry(ice, "v", lifespan, idle, false);
idle = 1;
se = InternalEntryFactory.create("k", "v", lifespan, idle);
@@ -206,11 +198,9 @@
Thread.sleep(100);
purgeExpired();
assert se.isExpired();
- assert cs.load("k") == null;
+ assertEventuallyExpires("k");
assert !cs.containsKey("k");
- if (supportsLoadAll) {
- assert cs.loadAll().isEmpty();
- }
+ assert cs.loadAll().isEmpty();
}
public void testStopStartDoesNotNukeValues() throws InterruptedException, CacheLoaderException {
@@ -522,13 +512,13 @@
} finally {
marshaller.finishObjectOutput(oo);
out.close();
- cs.clear();
+ cs.clear();
}
ByteArrayInputStream in = new ByteArrayInputStream(out.toByteArray());
ObjectInput oi = marshaller.startObjectInput(in, false);
try {
- cs.fromStream(new UnclosableObjectInputStream(oi));
+ cs.fromStream(new UnclosableObjectInputStream(oi));
} finally {
marshaller.finishObjectInput(oi);
in.close();
@@ -562,7 +552,7 @@
} finally {
marshaller.finishObjectOutput(oo);
out.close();
- cs.clear();
+ cs.clear();
}
// first pop the start bytes
@@ -572,17 +562,17 @@
try {
int bytesRead = oi.read(dummy, 0, 8);
assert bytesRead == 8;
- for (int i = 1; i < 9; i++) assert dummy[i - 1] == i : "Start byte stream corrupted!";
+ for (int i = 1; i < 9; i++) assert dummy[i - 1] == i : "Start byte stream corrupted!";
cs.fromStream(new UnclosableObjectInputStream(oi));
bytesRead = oi.read(dummy, 0, 8);
assert bytesRead == 8;
- for (int i = 8; i > 0; i--) assert dummy[8 - i] == i : "Start byte stream corrupted!";
+ for (int i = 8; i > 0; i--) assert dummy[8 - i] == i : "Start byte stream corrupted!";
} finally {
marshaller.finishObjectInput(oi);
in.close();
}
-
- Set<InternalCacheEntry> set = cs.loadAll();
+
+ Set<InternalCacheEntry> set = cs.loadAll();
assert set.size() == 3;
Set expected = new HashSet();
expected.add("k1");
@@ -637,9 +627,7 @@
int randomInt = r.nextInt(10);
InternalCacheEntry se = cs.load(keys[randomInt]);
assert se == null || se.getValue().equals(values[randomInt]);
- if (supportsLoadAll) {
- cs.loadAll();
- }
+ cs.loadAll();
} catch (Exception e) {
exceptions.add(e);
}
Modified: trunk/server/hotrod/src/main/scala/org/infinispan/server/hotrod/Decoder10.scala
===================================================================
--- trunk/server/hotrod/src/main/scala/org/infinispan/server/hotrod/Decoder10.scala 2010-07-19 13:15:06 UTC (rev 2060)
+++ trunk/server/hotrod/src/main/scala/org/infinispan/server/hotrod/Decoder10.scala 2010-07-19 13:44:38 UTC (rev 2061)
@@ -40,6 +40,7 @@
case 0x13 => ClearRequest
case 0x15 => StatsRequest
case 0x17 => PingRequest
+ case 0x19 => BulkGetRequest
case _ => throw new UnknownOperationException("Unknown operation: " + streamOp)
}
if (isTraceEnabled) trace("Operation code: {0} has been matched to {1}", streamOp, op)
@@ -145,6 +146,11 @@
new Response(h.messageId, h.cacheName, h.clientIntel, ClearResponse, Success, h.topologyId)
}
case PingRequest => new Response(h.messageId, h.cacheName, h.clientIntel, PingResponse, Success, h.topologyId)
+ case BulkGetRequest => {
+ val count = buffer.readUnsignedInt
+ if (isTraceEnabled) trace("About to create bulk response, count = " + count)
+ new BulkGetResponse(h.messageId, h.cacheName, h.clientIntel, BulkGetResponse, Success, h.topologyId, cache, count)
+ }
}
}
@@ -195,6 +201,7 @@
case ClearRequest => ClearResponse
case StatsRequest => StatsResponse
case PingRequest => PingResponse
+ case BulkGetRequest => BulkGetResponse
}
}
@@ -214,6 +221,7 @@
val ClearResponse = Value(0x14)
val StatsResponse = Value(0x16)
val PingResponse = Value(0x18)
+ val BulkGetResponse = Value(0x20)
val ErrorResponse = Value(0x50)
}
Modified: trunk/server/hotrod/src/main/scala/org/infinispan/server/hotrod/HotRodDecoder.scala
===================================================================
--- trunk/server/hotrod/src/main/scala/org/infinispan/server/hotrod/HotRodDecoder.scala 2010-07-19 13:15:06 UTC (rev 2060)
+++ trunk/server/hotrod/src/main/scala/org/infinispan/server/hotrod/HotRodDecoder.scala 2010-07-19 13:44:38 UTC (rev 2061)
@@ -102,9 +102,13 @@
override def createMultiGetResponse(h: HotRodHeader, pairs: Map[ByteArrayKey, CacheValue]): AnyRef =
null // Unsupported
- override def handleCustomRequest(h: HotRodHeader, b: ChannelBuffer, cache: Cache[ByteArrayKey, CacheValue]): AnyRef =
- h.decoder.handleCustomRequest(h, b, cache)
+ override def handleCustomRequest(h: HotRodHeader, b: ChannelBuffer, cache: Cache[ByteArrayKey, CacheValue]): AnyRef = {
+ val result = h.decoder.handleCustomRequest(h, b, cache)
+ if (isTrace) trace("About to return: " + result)
+ result
+ }
+
override def createStatsResponse(h: HotRodHeader, stats: Stats): AnyRef =
h.decoder.createStatsResponse(h, stats)
Modified: trunk/server/hotrod/src/main/scala/org/infinispan/server/hotrod/HotRodEncoder.scala
===================================================================
--- trunk/server/hotrod/src/main/scala/org/infinispan/server/hotrod/HotRodEncoder.scala 2010-07-19 13:15:06 UTC (rev 2060)
+++ trunk/server/hotrod/src/main/scala/org/infinispan/server/hotrod/HotRodEncoder.scala 2010-07-19 13:44:38 UTC (rev 2061)
@@ -8,6 +8,8 @@
import org.infinispan.Cache
import org.infinispan.server.core.{CacheValue, Logging}
import org.infinispan.util.ByteArrayKey
+import java.util.Iterator
+import org.infinispan.container.entries.{InternalCacheValue, InternalCacheEntry}
/**
* // TODO: Document this
@@ -50,6 +52,27 @@
buffer.writeRangedBytes(g.data.get)
}
}
+ case g: BulkGetResponse => {
+ if (isTrace) trace("About to repond to bulk get request: ")
+ if (g.status == Success) {
+ val dataContainer = g.cache.getAdvancedCache().getDataContainer()
+ var iterator: Iterator[InternalCacheEntry] = dataContainer.iterator()
+ val count = g.count
+ var written:Int = 0;
+ if (isTrace) trace("About to write (max) " + count + " messages to the client. Is written <= count ?" + (written <= count))
+ while (iterator.hasNext() && ((written < count) || (count == 0)) ) {
+ if (isTrace) trace("About to write message number " + written)
+ buffer.writeByte(1) //not done
+ written = written + 1
+ var ice: InternalCacheEntry = iterator.next()
+ val key:ByteArrayKey = ice.getKey().asInstanceOf[ByteArrayKey]
+ buffer.writeRangedBytes(key.getData)
+ val cacheValue : CacheValue = ice.getValue().asInstanceOf[CacheValue]
+ buffer.writeRangedBytes(cacheValue.data)
+ }
+ buffer.writeByte(0)
+ }
+ }
case g: GetResponse => if (g.status == Success) buffer.writeRangedBytes(g.data.get)
case e: ErrorResponse => buffer.writeString(e.msg)
case _ => if (buffer == null) throw new IllegalArgumentException("Response received is unknown: " + msg);
Modified: trunk/server/hotrod/src/main/scala/org/infinispan/server/hotrod/HotRodOperation.scala
===================================================================
--- trunk/server/hotrod/src/main/scala/org/infinispan/server/hotrod/HotRodOperation.scala 2010-07-19 13:15:06 UTC (rev 2060)
+++ trunk/server/hotrod/src/main/scala/org/infinispan/server/hotrod/HotRodOperation.scala 2010-07-19 13:44:38 UTC (rev 2061)
@@ -14,5 +14,6 @@
val ClearRequest = Value
val QuitRequest = Value
val PingRequest = Value
+ val BulkGetRequest = Value
}
\ No newline at end of file
Modified: trunk/server/hotrod/src/main/scala/org/infinispan/server/hotrod/Response.scala
===================================================================
--- trunk/server/hotrod/src/main/scala/org/infinispan/server/hotrod/Response.scala 2010-07-19 13:15:06 UTC (rev 2060)
+++ trunk/server/hotrod/src/main/scala/org/infinispan/server/hotrod/Response.scala 2010-07-19 13:44:38 UTC (rev 2061)
@@ -2,7 +2,9 @@
import OperationStatus._
import OperationResponse._
-import org.infinispan.util.Util
+import org.infinispan.Cache
+import org.infinispan.server.core.CacheValue
+import org.infinispan.util.{ByteArrayKey, Util}
/**
* // TODO: Document this
@@ -51,6 +53,19 @@
.append("}").toString
}
}
+class BulkGetResponse(override val messageId: Long, override val cacheName: String, override val clientIntel: Short,
+ override val operation: OperationResponse, override val status: OperationStatus,
+ override val topologyId: Int,
+ val cache: Cache[ByteArrayKey, CacheValue], val count: Int)
+ extends Response(messageId, cacheName, clientIntel, operation, status, topologyId) {
+ override def toString = {
+ new StringBuilder().append("BulkGetResponse").append("{")
+ .append("messageId=").append(messageId)
+ .append(", operation=").append(operation)
+ .append(", status=").append(status)
+ .append(", data=").append("}").toString
+ }
+}
class GetWithVersionResponse(override val messageId: Long, override val cacheName: String,
override val clientIntel: Short, override val operation: OperationResponse,
More information about the infinispan-commits
mailing list