diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AbstractRiakTemplate.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AbstractRiakTemplate.java
index 345e97745..aeced402e 100644
--- a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AbstractRiakTemplate.java
+++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AbstractRiakTemplate.java
@@ -104,10 +104,6 @@ public abstract class AbstractRiakTemplate extends RestGatewaySupport implements
* Whether or not to use the ETag-based cache.
*/
protected boolean useCache = true;
- /**
- * {@link java.util.concurrent.ExecutorService} to use for running asynchronous jobs.
- */
- protected ExecutorService executorService = Executors.newCachedThreadPool();
/**
* The URI to use inside the RestTemplate.
*/
@@ -124,6 +120,10 @@ public abstract class AbstractRiakTemplate extends RestGatewaySupport implements
* The default QosParameters to use for all operations through this template.
*/
protected QosParameters defaultQosParameters = null;
+ /**
+ * {@link java.util.concurrent.ExecutorService} to use for running asynchronous jobs.
+ */
+ protected ExecutorService workerPool = Executors.newCachedThreadPool();
protected Class defaultType = String.class;
protected ClassLoader classLoader = null;
@@ -192,6 +192,15 @@ public abstract class AbstractRiakTemplate extends RestGatewaySupport implements
this.defaultQosParameters = defaultQosParameters;
}
+ public ExecutorService getWorkerPool() {
+ return workerPool;
+ }
+
+ public void setWorkerPool(ExecutorService workerPool) {
+ this.workerPool = workerPool;
+ }
+
+
/**
* Get the default type to use if none can be inferred.
*
@@ -257,14 +266,6 @@ public abstract class AbstractRiakTemplate extends RestGatewaySupport implements
return "/riak";
}
- public ExecutorService getExecutorService() {
- return executorService;
- }
-
- public void setExecutorService(ExecutorService executorService) {
- this.executorService = executorService;
- }
-
public void afterPropertiesSet() throws Exception {
Assert.notNull(conversionService,
"Must specify a valid ConversionService.");
diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplate.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplate.java
index b099e5260..55fe982c6 100644
--- a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplate.java
+++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplate.java
@@ -38,16 +38,43 @@ import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.*;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
/**
+ * An implementation of {@link AsyncBucketKeyValueStoreOperations} and {@link
+ * AsyncMapReduceOperations} for the Riak datastore.
+ *
+ * To use the AsyncRiakTemplate, create a singleton in your Spring application-context.xml:
+ *
+ * <bean id="riak" class="org.springframework.data.keyvalue.riak.core.AsyncRiakTemplate"
+ * p:defaultUri="http://localhost:8098/riak/{bucket}/{key}"
+ * p:mapReduceUri="http://localhost:8098/mapred"/>
+ *
+ * def callback = [
+ * completed: { v, meta ->
+ * ... do something with results ...
+ * },
+ * failed: { err ->
+ * }
+ * ] as AsyncKeyValueStoreOperation
+ * def obj = new TestObject(name: "My Name", age: 40)
+ * def future = riak.set("mybucket", "mykey", obj, callback)
+ * ... this runs asynchronously, so do other work ...
+ * def name = future.get().name
+ * println "Hello $name!"
+ *
*/
public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBucketKeyValueStoreOperations, AsyncMapReduceOperations {
protected final Logger log = LoggerFactory.getLogger(getClass());
- protected ExecutorService workerPool = Executors.newCachedThreadPool();
protected AsyncKeyValueStoreOperation defaultErrorHandler = new LoggingErrorHandler();
public AsyncRiakTemplate() {
@@ -58,14 +85,6 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck
super(requestFactory);
}
- public ExecutorService getWorkerPool() {
- return workerPool;
- }
-
- public void setWorkerPool(ExecutorService workerPool) {
- this.workerPool = workerPool;
- }
-
public AsyncKeyValueStoreOperation getDefaultErrorHandler() {
return defaultErrorHandler;
}
diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/RiakTemplate.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/RiakTemplate.java
index 518bde723..d00abd718 100644
--- a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/RiakTemplate.java
+++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/RiakTemplate.java
@@ -129,7 +129,10 @@ public class RiakTemplate extends AbstractRiakTemplate implements BucketKeyValue
KeyValueStoreMetaData origMeta = getMetaData(bucket, keyName);
String vclock = null;
if (null != origMeta) {
- vclock = origMeta.getProperties().get(RIAK_VCLOCK).toString();
+ Object o = origMeta.getProperties().get(RIAK_VCLOCK);
+ if (null != o) {
+ vclock = o.toString();
+ }
}
RestTemplate restTemplate = getRestTemplate();
HttpHeaders headers = new HttpHeaders();
@@ -552,7 +555,7 @@ public class RiakTemplate extends AbstractRiakTemplate implements BucketKeyValue
@SuppressWarnings({"unchecked"})
public Future> submit(MapReduceJob job) {
// Run this job asynchronously.
- return executorService.submit(job);
+ return workerPool.submit(job);
}
/*----------------- Link Operations -----------------*/
diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakBuilder.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakBuilder.java
index 2c17e027b..5dced9278 100644
--- a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakBuilder.java
+++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakBuilder.java
@@ -37,6 +37,21 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
+ * A Groovy Builder that implements a powerful and syntactically succinct DSL for Riak datastore
+ * access using SDKV for Riak's {@link AsyncRiakTemplate}.
+ *
+ * The DSL responds to most of the important methods from the AsyncRiakTemplate:
+ * set setAsBytes put get getAsBytes
+ * getAsType containsKey delete foreach
+ * An example of DSL usage (to delete all entries in a bucket):
+ * riak.foreach(bucket: "test") {
+ * completed { v, meta ->
+ * delete(bucket: "test", key: meta.key)
+ * }
+ * }
+ *
*/
public class RiakBuilder extends BuilderSupport {
@@ -310,7 +325,9 @@ public class RiakBuilder extends BuilderSupport {
@SuppressWarnings({"unchecked"})
@Override
protected void nodeCompleted(Object parent, Object node) {
- log.debug("nodeCompleted: parent=" + parent + ", node=" + node);
+ if (log.isDebugEnabled()) {
+ log.debug("nodeCompleted: parent=" + parent + ", node=" + node);
+ }
if (parent instanceof RiakMapReduceOperation && node instanceof QueryPhase) {
QueryPhase p = (QueryPhase) node;
MapReduceOperation oper = null;
@@ -339,7 +356,9 @@ public class RiakBuilder extends BuilderSupport {
@SuppressWarnings({"unchecked"})
@Override
protected Object postNodeCompletion(Object parent, Object node) {
- log.debug("postNodeCompletion: " + parent + " " + node);
+ if (log.isDebugEnabled()) {
+ log.debug("postNodeCompletion: " + parent + " " + node);
+ }
if (node instanceof RiakOperation) {
RiakOperation op = (RiakOperation) node;
try {
diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakMapReduceOperation.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakMapReduceOperation.java
index 7d7ac3df8..e00fc69d1 100644
--- a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakMapReduceOperation.java
+++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakMapReduceOperation.java
@@ -84,9 +84,12 @@ public class RiakMapReduceOperation implements Callable {
public Object call() throws Exception {
Future f = riak.execute(job, new AsyncKeyValueStoreOperation, Object>() {
public Object completed(KeyValueStoreMetaData meta, List result) {
- Object arg = new Object[]{result, meta};
if (null != completed) {
- return completed.call(arg);
+ if (completed.getParameterTypes().length == 2) {
+ return completed.call(new Object[]{result, meta});
+ } else {
+ return completed.call(result);
+ }
} else {
return new Object[]{result, meta};
}
diff --git a/spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/RiakBuilderSpec.groovy b/spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/RiakBuilderSpec.groovy
index 1b8521aad..ae6e43393 100644
--- a/spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/RiakBuilderSpec.groovy
+++ b/spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/RiakBuilderSpec.groovy
@@ -264,10 +264,9 @@ class RiakBuilderSpec extends Specification {
given:
def riak = new RiakBuilder(riakTemplate)
- def deleted = false
when:
- riak {
+ def deleted = riak {
"test" {
foreach {
completed { v, meta ->
diff --git a/src/docbkx/reference/riak.xml b/src/docbkx/reference/riak.xml
index 9a50349c1..7d41aa66d 100644
--- a/src/docbkx/reference/riak.xml
+++ b/src/docbkx/reference/riak.xml
@@ -78,8 +78,24 @@
]]>
-
- It might also be necessary to replace the default ExecutorService (by default a cached ThreadPoolExecutor ) with an executor you've explicitly configured. Set your ExecutorService on the template's "executorService" property.
+
+ You can also set a specific ClassLoader to use when loading objects from Riak. Just set the classLoader property:
+
+
+
+
+ ]]>
+
+
@@ -191,7 +207,7 @@ riak.link("childbucket", "childkey", "sourcebucket", "sourcekey", "tagname");
Link Walking
- When entries are linked together in Riak, those relationships can be efficiently traversed on the server using a feature called Link Walking . Rather than requesting each object in a link's relationship individually, a link walk pulls all the related objects at once and sends that data back to the client as MIME-encoded multipart data. As such, it requires special processing to convert those multiple entries into a List of objects, just as if you had used a get method. If you don't specify a type to convert the objects to, the linkWalk java.util.Map .
+ When entries are linked together in Riak, those relationships can be efficiently traversed on the server using a feature called Link Walking . Rather than requesting each object in a link's relationship individually, a link walk pulls all the related objects at once and sends that data back to the client as MIME-encoded multipart data. As such, it requires special processing to convert those multiple entries into a List of objects, just as if you had used a get method. If you don't specify a type to convert the objects to, the linkWalk java.util.Map .
To link walk a relationship and return a list of custom POJOs, you would do something like this:
pair = new ArrayList() {{
add("mybucket");
add("mykey");
}};
-List keys = new ArrayList() {{
+List> keys = new ArrayList>() {{
add(pair);
}};
job.addInputs(keys); // Will M/R only specified keys
@@ -332,6 +348,156 @@ riak.updateBucketSchema("mybucket", props);
+
+ Asynchronous Access
+
+ SDKV for Riak also includes an asynchronous version of most of the methods available to the RiakTemplate , whose method calls are all synchronous. The asynchronous version of the template is called AsyncRiakTemplate .
+
+
+ Template Configuration
+
+ The AsyncRiakTemplate has the same basic configuration properties as the synchronous RiakTemplate . The only other property specific to the AsyncRiakTemplate you might want to configure is the thread pool the template uses to execute tasks asynchronously (by default a cached ThreadPoolExecutor ). Set your ExecutorService on the template's workerPool property.
+
+
+
+
+ Callbacks
+
+ Using the asynchronous Riak support in SDKV means you'll be relying on callbacks to execute your business logic when the requested operation is completed. All asynchronous operations follow a similar pattern:
+
+ They are named similarly to their synchronous counterparts.
+ They take a AsyncKeyValueStoreOperation<?, ?> as a final parameter.
+ They return a Future<?> .
+
+
+
+ To perform an asynchronous get on a JSON-serialized Map object which returns a custom object from the callback, you'd do something like:
+ future = riak.get("mybucket", "mykey", new AsyncKeyValueStoreOperation() {
+
+ MyObject obj = new MyObject();
+
+ MyObject completed(KeyValueStoreMetaData meta, Map result) {
+ obj.setName(result.get("name"));
+ return obj;
+ }
+
+ MyObject failed(Throwable error) {
+ obj.setError(error);
+ return obj;
+ }
+
+});
+
+// Maybe do other work while waiting...
+MyObject obj = future.get();
+ ]]>
+
+
+
+
+
+
+ Groovy Builder Support
+
+ If your application uses Groovy, either in a standalone context, or as part of a Grails application, then you could benefit from using the Groovy RiakBuilder AsyncRiakTemplate . To use the RiakBuilder , pass the constructor a configured AsyncRiakTemplate .
+
+ Instances of RiakBuilder are NOT thread-safe and should not be shared across threads. The RiakBuilder implements an easy-to-use DSL for interacting with Riak. It doesn't implement the full set of methods available on the underlying AsyncRiakTemplate but a subset. The methods that the RiakBuilder responds to are:
+
+ set
+ setAsBytes
+ put
+ get
+ getAsBytes
+ getAsType
+ containsKey
+ delete
+ foreach
+
+
+
+
+ Riak DSL Usage
+
+ The following example illustrates the different uses of the Riak DSL, including batching requests together into a logical group, using a default bucket name (the node directly beneath riak will be considered the default bucket to use for the contained operations unless a different one is specified on the operation itself):
+ meta.key }}
+ put(value: [test: "value"]) { completed { v, meta -> meta.key }}
+ put(value: [test: "value"]) { completed { v, meta -> meta.key }}
+ put(value: [test: "value"]) { completed { v, meta -> meta.key }}
+
+ mapreduce {
+ query {
+ map(arg: [test: "arg", alist: [1, 2, 3, 4]]) {
+ source "function(v, keyInfo, arg){ return [1]; }"
+ }
+ reduce {
+ source "function(v){ return Riak.reduceSum(v); }"
+ }
+ }
+ failed { it.printStackTrace() }
+ }
+ }
+}
+def results = riak.results
+
+riak.foreach(bucket: "test") {
+ completed { v, meta ->
+ riak.delete(bucket: "test", key: meta.key)
+ }
+}
+ ]]>
+
+
+ Some important things to note from this example:
+
+ Each operation in the Riak DSL has two callbacks: completed and failed .
+ The completed closure is passed either the result object, or, if your closure is defined with two parameters, the result object and the metadata associated with that entry.
+ Operations can be enclosed in an arbitrarily-named closure which the builder interprets as a default bucket name (in this case, the node "test" tells the builder to use the bucket name "test" for a default, unless one is specified on one of the enclosed operations).
+ Each operation within a builder's execution will be accumulated inside the special results property. Code that needs to know the output of individual operations within the batch can get access to that object through this property. Note that this means that RiakBuilder instances are NOT thread-safe.
+
+
+
+ Even though the Riak DSL uses an asynchronous template underneath, all operations performed through the DSL will, by default, block until complete. To get a truly asynchronous operation, pass the parameter wait: 0 (or give a meaningful timeout in milliseconds to wait for the operation to complete) on the operation.
+ QosParameters on Riak DSL Operations
+
+ You can pass QosParameters to Riak DSL operations by simply defining them as parameters to the operation:
+
+
+
+
+ Working with Riak DSL Output
+
+ The output of DSL operations will either be passed to the configured completed callback, or be returned to the caller if no callback is specified. In the example above, the mapreduce operation has no completed closure. Therefore, the return of the reduce phase is simply passed back to the builder, which makes that output available on the special results property.
+
+ To gain access to the operation's results immediately, simply assign it to a variable:
+ If you add a non-zero wait value to the operation, "myobj" will contain a Future<?> rather than the result object itself.
+
+
+
+
+
+