Added asynchronous version of RiakTemplate, Groovy DSL for data access.

spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AbstractAsyncOperation.java

@@ -1,52 +0,0 @@
-/*
- * Copyright (c) 2010 by J. Brisbin <jon@jbrisbin.com>
- *     Portions (c) 2010 by NPC International, Inc. or the
- *     original author(s).
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.data.keyvalue.riak.core;
-
-import org.springframework.beans.factory.InitializingBean;
-import org.springframework.util.Assert;
-
-import java.util.concurrent.Callable;
-
-/**
- * @author J. Brisbin <jon@jbrisbin.com>
- */
-public abstract class AbstractAsyncOperation<T> implements Callable<T>, InitializingBean {
-
-  protected RiakTemplate riakTemplate;
-
-  protected AbstractAsyncOperation() {
-  }
-
-  protected AbstractAsyncOperation(RiakTemplate riakTemplate) {
-    this.riakTemplate = riakTemplate;
-  }
-
-  public RiakTemplate getRiakTemplate() {
-    return riakTemplate;
-  }
-
-  public void setRiakTemplate(RiakTemplate riakTemplate) {
-    this.riakTemplate = riakTemplate;
-  }
-
-  public void afterPropertiesSet() throws Exception {
-    Assert.notNull(riakTemplate, "Must provide a configured RiakTemplate for this operation.");
-  }
-
-}

spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AbstractRiakTemplate.java

@@ -59,10 +59,8 @@ import java.util.regex.Pattern;
  */
 public abstract class AbstractRiakTemplate extends RestGatewaySupport implements InitializingBean {
 
-  /**
-   * Client ID used by Riak to correlate updates.
-   */
-  protected static final String RIAK_CLIENT_ID = "org.springframework.data.keyvalue.riak.core.RiakTemplate/1.0";
+  protected static final String RIAK_META_CLASSNAME = "X-Riak-Meta-ClassName";
+
   /**
    * Regex used to extract host, port, and prefix from the given URI.
    */
@@ -81,6 +79,12 @@ public abstract class AbstractRiakTemplate extends RestGatewaySupport implements
       "EEE, d MMM yyyy HH:mm:ss z");
 
   protected final Logger log = LoggerFactory.getLogger(getClass());
+
+  /**
+   * Client ID used by Riak to correlate updates.
+   */
+  protected final String RIAK_CLIENT_ID = getClass().getName() + "/1.0";
+
   /**
    * For converting objects to/from other kinds of objects.
    */
@@ -380,4 +384,16 @@ public abstract class AbstractRiakTemplate extends RestGatewaySupport implements
         "&") : "");
   }
 
+  protected HttpHeaders defaultHeaders(Map<String, ?> metadata) {
+    HttpHeaders headers = new HttpHeaders();
+    headers.set("X-Riak-ClientId", RIAK_CLIENT_ID);
+    if (null != metadata) {
+      for (Map.Entry<String, ?> entry : metadata.entrySet()) {
+        Object o = entry.getValue();
+        headers.set(entry.getKey(), (null != o ? o.toString() : null));
+      }
+    }
+    return headers;
+  }
+
 }

spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncBucketKeyValueStoreOperations.java

@@ -0,0 +1,166 @@
+/*
+ * Copyright (c) 2010 by J. Brisbin <jon@jbrisbin.com>
+ * Portions (c) 2010 by NPC International, Inc. or the
+ * original author(s).
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.data.keyvalue.riak.core;
+
+import java.util.Map;
+import java.util.concurrent.Future;
+
+/**
+ * An asynchronous version of {@link BucketKeyValueStoreOperations}.
+ *
+ * @author J. Brisbin <jon@jbrisbin.com>
+ */
+public interface AsyncBucketKeyValueStoreOperations {
+
+  /**
+   * Put an object in Riak at a specific bucket and key and invoke callback with the value
+   * pulled back out of Riak after the update, which contains full headers and metadata.
+   *
+   * @param bucket
+   * @param key
+   * @param value
+   * @param callback Called with the update value pulled from Riak
+   */
+  <B, K, V> Future<?> set(B bucket, K key, V value, AsyncKeyValueStoreOperation<V> callback);
+
+  /**
+   * @param bucket
+   * @param key
+   * @param value
+   * @param qosParams
+   * @return
+   */
+  <B, K, V> Future<?> set(B bucket, K key, V value, QosParameters qosParams, AsyncKeyValueStoreOperation<V> callback);
+
+  /**
+   * @param bucket
+   * @param key
+   * @param value
+   * @return
+   */
+  <B, K> Future<?> setAsBytes(B bucket, K key, byte[] value, AsyncKeyValueStoreOperation<byte[]> callback);
+
+  /**
+   * @param bucket
+   * @param key
+   * @param value
+   * @param qosParams
+   * @return
+   */
+  <B, K> Future<?> setAsBytes(B bucket, K key, byte[] value, QosParameters qosParams, AsyncKeyValueStoreOperation<byte[]> callback);
+
+  /**
+   * @param bucket
+   * @param key
+   * @param value
+   * @param metaData
+   * @return
+   */
+  <B, K, V> Future<?> setWithMetaData(B bucket, K key, V value, Map<String, String> metaData, AsyncKeyValueStoreOperation<V> callback);
+
+  /**
+   * @param bucket
+   * @param key
+   * @param value
+   * @param metaData
+   * @param qosParams
+   * @return
+   */
+  <B, K, V> Future<?> setWithMetaData(B bucket, K key, V value, Map<String, String> metaData, QosParameters qosParams, AsyncKeyValueStoreOperation<V> callback);
+
+  /**
+   * @param bucket
+   * @param key
+   * @return
+   */
+  <B, K, V> Future<?> get(B bucket, K key, AsyncKeyValueStoreOperation<V> callback);
+
+  /**
+   * @param bucket
+   * @param key
+   * @return
+   */
+  <B, K> Future<?> getAsBytes(B bucket, K key, AsyncKeyValueStoreOperation<byte[]> callback);
+
+  /**
+   * @param bucket
+   * @param key
+   * @param requiredType
+   * @return
+   */
+  <B, K, T> Future<?> getAsType(B bucket, K key, Class<T> requiredType, AsyncKeyValueStoreOperation<T> callback);
+
+  /**
+   * @param bucket
+   * @param key
+   * @param value
+   * @return
+   */
+  <B, K, V> Future<?> getAndSet(B bucket, K key, V value, AsyncKeyValueStoreOperation<V> callback);
+
+  /**
+   * @param bucket
+   * @param key
+   * @param value
+   * @return
+   */
+  <B, K> Future<?> getAndSetAsBytes(B bucket, K key, byte[] value, AsyncKeyValueStoreOperation<byte[]> callback);
+
+  /**
+   * @param bucket
+   * @param key
+   * @param value
+   * @param requiredType
+   * @return
+   */
+  <B, K, V, T> Future<?> getAndSetAsType(B bucket, K key, V value, Class<T> requiredType, AsyncKeyValueStoreOperation<V> callback);
+
+  /**
+   * @param bucket
+   * @param key
+   * @param value
+   * @return
+   */
+  <B, K, V> Future<?> setIfKeyNonExistent(B bucket, K key, V value, AsyncKeyValueStoreOperation<V> callback);
+
+  /**
+   * @param bucket
+   * @param key
+   * @param value
+   * @return
+   */
+  <B, K> Future<?> setIfKeyNonExistentAsBytes(B bucket, K key, byte[] value, AsyncKeyValueStoreOperation<byte[]> callback);
+
+  /**
+   * @param bucket
+   * @param key
+   * @return
+   */
+  <B, K> Future<?> containsKey(B bucket, K key, AsyncKeyValueStoreOperation<Boolean> callback);
+
+  /**
+   * Delete a specific entry from this data store.
+   *
+   * @param bucket
+   * @param key
+   * @return
+   */
+  <B, K> Future<?> delete(B bucket, K key, AsyncKeyValueStoreOperation<Boolean> callback);
+
+}

spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncKeyValueStoreOperation.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright (c) 2010 by J. Brisbin <jon@jbrisbin.com>
+ * Portions (c) 2010 by NPC International, Inc. or the
+ * original author(s).
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.data.keyvalue.riak.core;
+
+/**
+ * @author J. Brisbin <jon@jbrisbin.com>
+ */
+public interface AsyncKeyValueStoreOperation<V> {
+
+  void completed(KeyValueStoreMetaData meta, V result);
+
+  void failed(Throwable error);
+}

spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplate.java

@@ -0,0 +1,417 @@
+/*
+ * Copyright (c) 2010 by J. Brisbin <jon@jbrisbin.com>
+ * Portions (c) 2010 by NPC International, Inc. or the
+ * original author(s).
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.data.keyvalue.riak.core;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.keyvalue.riak.DataStoreOperationException;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.http.client.ClientHttpRequestFactory;
+import org.springframework.util.Assert;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+/**
+ * @author J. Brisbin <jon@jbrisbin.com>
+ */
+public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBucketKeyValueStoreOperations {
+
+  protected final Logger log = LoggerFactory.getLogger(getClass());
+
+  protected ExecutorService workerPool = Executors.newCachedThreadPool();
+  protected AsyncKeyValueStoreOperation<Throwable> defaultErrorHandler = new LoggingErrorHandler();
+
+  public AsyncRiakTemplate() {
+    super();
+  }
+
+  public AsyncRiakTemplate(ClientHttpRequestFactory requestFactory) {
+    super(requestFactory);
+  }
+
+  public ExecutorService getWorkerPool() {
+    return workerPool;
+  }
+
+  public void setWorkerPool(ExecutorService workerPool) {
+    this.workerPool = workerPool;
+  }
+
+  public AsyncKeyValueStoreOperation<Throwable> getDefaultErrorHandler() {
+    return defaultErrorHandler;
+  }
+
+  public void setDefaultErrorHandler(AsyncKeyValueStoreOperation<Throwable> defaultErrorHandler) {
+    this.defaultErrorHandler = defaultErrorHandler;
+  }
+
+  public <B, K, V> Future<?> set(B bucket, K key, V value, AsyncKeyValueStoreOperation<V> callback) {
+    return setWithMetaData(bucket, key, value, null, null, callback);
+  }
+
+  public <B, K, V> Future<?> set(B bucket, K key, V value, QosParameters qosParams, AsyncKeyValueStoreOperation<V> callback) {
+    return setWithMetaData(bucket, key, value, null, qosParams, callback);
+  }
+
+  public <B, K> Future<?> setAsBytes(B bucket, K key, byte[] value, AsyncKeyValueStoreOperation<byte[]> callback) {
+    return setWithMetaData(bucket, key, value, null, null, callback);
+  }
+
+  @SuppressWarnings({"unchecked"})
+  public <B, K, V> Future<V> setWithMetaData(B bucket, K key, V value, Map<String, String> metaData, QosParameters qosParams, AsyncKeyValueStoreOperation<V> callback) {
+    String bucketName = (null != bucket ? bucket.toString() : value.getClass().getName());
+    // Get a key name that may or may not include the QOS parameters.
+    Assert.notNull(key, "Cannot use a <NULL> key.");
+    String keyName = (null != qosParams ? key.toString() + extractQosParameters(qosParams) : key
+        .toString());
+    HttpHeaders headers = defaultHeaders(metaData);
+    headers.setContentType(extractMediaType(value));
+    headers.set(RIAK_META_CLASSNAME, value.getClass().getName());
+    HttpEntity<V> entity = new HttpEntity<V>(value, headers);
+    return (Future<V>) workerPool.submit(new AsyncPost<V>(bucketName,
+        keyName,
+        entity,
+        callback));
+  }
+
+  public <B, K, V> Future<?> get(B bucket, K key, AsyncKeyValueStoreOperation<V> callback) {
+    return getWithMetaData(bucket, key, null, callback);
+  }
+
+  @SuppressWarnings({"unchecked"})
+  public <B, K, T> Future<?> getWithMetaData(B bucket, K key, Class<T> requiredType, AsyncKeyValueStoreOperation<T> callback) {
+    String bucketName = (null != bucket ? bucket.toString() : requiredType.getName());
+    // Get a key name that may or may not include the QOS parameters.
+    Assert.notNull(key, "Cannot use a <NULL> key.");
+    if (null == requiredType) {
+      try {
+        requiredType = (Class<T>) getType(bucketName, key.toString());
+      } catch (ClassNotFoundException e) {
+        throw new DataStoreOperationException(e.getMessage(), e);
+      }
+    }
+    return workerPool.submit(new AsyncGet<T>(bucketName,
+        key.toString(),
+        requiredType,
+        callback));
+  }
+
+  public <B, K> Future<?> getAsBytes(B bucket, K key, AsyncKeyValueStoreOperation<byte[]> callback) {
+    return getWithMetaData(bucket, key, byte[].class, callback);
+  }
+
+  public <B, K, T> Future<?> getAsType(B bucket, K key, Class<T> requiredType, AsyncKeyValueStoreOperation<T> callback) {
+    return getWithMetaData(bucket, key, requiredType, callback);
+  }
+
+  public <B, K, V> Future<?> getAndSet(final B bucket, final K key, final V value, final AsyncKeyValueStoreOperation<V> callback) {
+    final List<Future<?>> futures = new ArrayList<Future<?>>();
+    try {
+      getWithMetaData(bucket, key, null, new AsyncKeyValueStoreOperation<Object>() {
+        @SuppressWarnings({"unchecked"})
+        public void completed(KeyValueStoreMetaData meta, Object result) {
+          futures.add(setWithMetaData(bucket, key, value, null, null, null));
+          callback.completed(meta, (V) result);
+        }
+
+        public void failed(Throwable error) {
+          callback.failed(error);
+        }
+      }).get();
+    } catch (InterruptedException e) {
+      log.error(e.getMessage(), e);
+    } catch (ExecutionException e) {
+      log.error(e.getMessage(), e);
+    }
+    return futures.size() > 0 ? futures.get(0) : null;
+  }
+
+  public <B, K> Future<?> getAndSetAsBytes(B bucket, K key, byte[] value, AsyncKeyValueStoreOperation<byte[]> callback) {
+    return getAndSet(bucket, key, value, callback);
+  }
+
+  public <B, K, V, T> Future<?> getAndSetAsType(final B bucket, final K key, final V value, final Class<T> requiredType, final AsyncKeyValueStoreOperation<V> callback) {
+    final List<Future<?>> futures = new ArrayList<Future<?>>();
+    getWithMetaData(bucket, key, requiredType, new AsyncKeyValueStoreOperation<T>() {
+      @SuppressWarnings({"unchecked"})
+      public void completed(KeyValueStoreMetaData meta, T result) {
+        futures.add(setWithMetaData(bucket, key, value, null, null, null));
+        callback.completed(meta, (V) result);
+      }
+
+      public void failed(Throwable error) {
+        callback.failed(error);
+      }
+    });
+    return futures.size() > 0 ? futures.get(0) : null;
+  }
+
+  public <B, K, V> Future<?> setIfKeyNonExistent(final B bucket, final K key, final V value, final AsyncKeyValueStoreOperation<V> callback) {
+    return containsKey(bucket, key, new AsyncKeyValueStoreOperation<Boolean>() {
+      public void completed(KeyValueStoreMetaData meta, Boolean result) {
+        if (!result) {
+          setWithMetaData(bucket, key, value, null, null, callback);
+        }
+      }
+
+      public void failed(Throwable error) {
+        callback.failed(error);
+      }
+    });
+  }
+
+  public <B, K> Future<?> setIfKeyNonExistentAsBytes(final B bucket, final K key, final byte[] value, final AsyncKeyValueStoreOperation<byte[]> callback) {
+    return containsKey(bucket, key, new AsyncKeyValueStoreOperation<Boolean>() {
+      public void completed(KeyValueStoreMetaData meta, Boolean result) {
+        if (!result) {
+          setWithMetaData(bucket, key, value, null, null, callback);
+        }
+      }
+
+      public void failed(Throwable error) {
+        callback.failed(error);
+      }
+    });
+  }
+
+  public <B, K> Future<?> containsKey(B bucket, K key, final AsyncKeyValueStoreOperation<Boolean> callback) {
+    Assert.notNull(bucket, "Bucket cannot be null when checking for existence.");
+    Assert.notNull(key, "Key cannot be null when checking for existence");
+    return workerPool.submit(new AsyncHead(bucket.toString(),
+        key.toString(),
+        new AsyncKeyValueStoreOperation<HttpHeaders>() {
+          public void completed(KeyValueStoreMetaData meta, HttpHeaders result) {
+            callback.completed(null, (null != result));
+          }
+
+          public void failed(Throwable error) {
+            callback.failed(error);
+          }
+        }));
+  }
+
+  public <B, K> Future<?> delete(B bucket, K key, AsyncKeyValueStoreOperation<Boolean> callback) {
+    Assert.notNull(bucket, "Bucket cannot be null when deleting.");
+    Assert.notNull(key, "Key cannot be null when deleting.");
+    return workerPool.submit(new AsyncDelete(bucket.toString(), key.toString(), callback));
+  }
+
+  public <B, K> Future<?> setAsBytes(B bucket, K key, byte[] value, QosParameters qosParams, AsyncKeyValueStoreOperation<byte[]> callback) {
+    return setWithMetaData(bucket, key, value, null, qosParams, callback);
+  }
+
+  public <B, K, V> Future<?> setWithMetaData(B bucket, K key, V value, Map<String, String> metaData, AsyncKeyValueStoreOperation<V> callback) {
+    return setWithMetaData(bucket, key, value, metaData, null, callback);
+  }
+
+  protected Class<?> getType(String bucket, String key) throws ClassNotFoundException {
+    HttpHeaders headers = getRestTemplate().headForHeaders(defaultUri, bucket, key);
+    Class<?> clazz = null;
+    if (null != headers) {
+      String s = headers.getFirst(RIAK_META_CLASSNAME);
+      if (null != s) {
+        try {
+          clazz = Class.forName(s);
+        } catch (ClassNotFoundException ignored) {
+          if (headers.getContentType().equals(MediaType.APPLICATION_JSON)) {
+            clazz = Map.class;
+          } else if (headers.getContentType().equals(MediaType.TEXT_PLAIN)) {
+            clazz = String.class;
+          } else {
+            // handle as bytes
+            log.error("Need to handle bytes!");
+          }
+        }
+      }
+    }
+    if (null == clazz) {
+      clazz = byte[].class;
+    }
+    return clazz;
+  }
+
+  protected class AsyncPost<V> implements Runnable {
+
+    private String bucket;
+    private String key;
+    private HttpEntity<V> entity = null;
+    private AsyncKeyValueStoreOperation<V> callback = null;
+
+    public AsyncPost(String bucket, String key, HttpEntity<V> entity, AsyncKeyValueStoreOperation<V> callback) {
+      this.bucket = bucket;
+      this.key = key;
+      this.entity = entity;
+      this.callback = callback;
+    }
+
+    @SuppressWarnings({"unchecked"})
+    public void run() {
+      try {
+        HttpEntity<?> result = getRestTemplate().postForEntity(defaultUri,
+            entity,
+            (entity.getBody() instanceof byte[] ? byte[].class : entity.getBody().getClass()),
+            bucket,
+            key + "?returnbody=true");
+        if (log.isDebugEnabled()) {
+          log.debug(String.format("PUT object: bucket=%s, key=%s, value=%s",
+              bucket,
+              key,
+              entity));
+        }
+        if (null != callback) {
+          callback.completed(extractMetaData(result.getHeaders()), (V) result.getBody());
+        }
+      } catch (Throwable t) {
+        DataStoreOperationException dsoe = new DataStoreOperationException(t.getMessage(), t);
+        if (null != callback) {
+          callback.failed(dsoe);
+        } else {
+          defaultErrorHandler.failed(dsoe);
+        }
+      }
+    }
+
+  }
+
+  protected class AsyncGet<T> implements Runnable {
+
+    private String bucket;
+    private String key;
+    private Class<T> requiredType;
+    private AsyncKeyValueStoreOperation<T> callback = null;
+
+    public AsyncGet(String bucket, String key, Class<T> requiredType, AsyncKeyValueStoreOperation<T> callback) {
+      this.bucket = bucket;
+      this.key = key;
+      this.requiredType = requiredType;
+      this.callback = callback;
+    }
+
+    public void run() {
+      try {
+        ResponseEntity<T> result = getRestTemplate().getForEntity(defaultUri,
+            requiredType,
+            bucket,
+            key);
+        if (result.hasBody()) {
+          RiakMetaData meta = extractMetaData(result.getHeaders());
+          RiakValue<T> val = new RiakValue<T>(result.getBody(), meta);
+          if (useCache) {
+            cache.put(new SimpleBucketKeyPair<Object, Object>(bucket, key), val);
+          }
+          if (null != callback) {
+            callback.completed(meta, val.get());
+          }
+          if (log.isDebugEnabled()) {
+            log.debug(String.format("GET object: bucket=%s, key=%s, type=%s",
+                bucket,
+                key,
+                requiredType.getName()));
+          }
+        }
+      } catch (Throwable t) {
+        DataStoreOperationException dsoe = new DataStoreOperationException(t.getMessage(), t);
+        if (null != callback) {
+          callback.failed(dsoe);
+        } else {
+          defaultErrorHandler.failed(dsoe);
+        }
+      }
+    }
+  }
+
+  protected class AsyncHead implements Runnable {
+
+    private String bucket;
+    private String key;
+    private AsyncKeyValueStoreOperation<HttpHeaders> callback = null;
+
+    public AsyncHead(String bucket, String key, AsyncKeyValueStoreOperation<HttpHeaders> callback) {
+      this.bucket = bucket;
+      this.key = key;
+      this.callback = callback;
+    }
+
+    public void run() {
+      try {
+        HttpHeaders headers = getRestTemplate().headForHeaders(defaultUri, bucket, key);
+        if (null != headers) {
+          if (null != callback) {
+            callback.completed(null, headers);
+          }
+        }
+      } catch (Throwable t) {
+        DataStoreOperationException dsoe = new DataStoreOperationException(t.getMessage(), t);
+        if (null != callback) {
+          callback.failed(dsoe);
+        } else {
+          defaultErrorHandler.failed(dsoe);
+        }
+      }
+    }
+  }
+
+  protected class AsyncDelete implements Runnable {
+
+    private String bucket;
+    private String key;
+    private AsyncKeyValueStoreOperation<Boolean> callback = null;
+
+    public AsyncDelete(String bucket, String key, AsyncKeyValueStoreOperation<Boolean> callback) {
+      this.bucket = bucket;
+      this.key = key;
+      this.callback = callback;
+    }
+
+    public void run() {
+      try {
+        getRestTemplate().delete(defaultUri, bucket, key);
+        if (null != callback) {
+          callback.completed(null, true);
+        }
+      } catch (Throwable t) {
+        DataStoreOperationException dsoe = new DataStoreOperationException(t.getMessage(), t);
+        if (null != callback) {
+          callback.failed(dsoe);
+        } else {
+          defaultErrorHandler.failed(dsoe);
+        }
+      }
+    }
+  }
+
+  protected class LoggingErrorHandler implements AsyncKeyValueStoreOperation<Throwable> {
+    public void completed(KeyValueStoreMetaData meta, Throwable result) {
+    }
+
+    public void failed(Throwable error) {
+      log.error(error.getMessage(), error);
+    }
+  }
+
+}

spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/BucketKeyValueStoreOperations.java

@@ -201,4 +201,5 @@ public interface BucketKeyValueStoreOperations {
    * @return
    */
   <B, K, V> BucketKeyValueStoreOperations setWithMetaData(B bucket, K key, V value, Map<String, String> metaData);
+
 }

spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakBuilder.java

@@ -0,0 +1,184 @@
+/*
+ * Copyright (c) 2010 by J. Brisbin <jon@jbrisbin.com>
+ * Portions (c) 2010 by NPC International, Inc. or the
+ * original author(s).
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.data.keyvalue.riak.groovy;
+
+import groovy.lang.Closure;
+import groovy.util.BuilderSupport;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.keyvalue.riak.core.AsyncRiakTemplate;
+import org.springframework.data.keyvalue.riak.core.RiakQosParameters;
+
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+/**
+ * @author J. Brisbin <jon@jbrisbin.com>
+ */
+public class RiakBuilder extends BuilderSupport {
+
+  protected final Logger log = LoggerFactory.getLogger(getClass());
+  protected AsyncRiakTemplate riak;
+  protected ExecutorService workerPool = Executors.newCachedThreadPool();
+
+  public RiakBuilder(AsyncRiakTemplate riak) {
+    this.riak = riak;
+  }
+
+  public RiakBuilder(AsyncRiakTemplate riak, ExecutorService workerPool) {
+    this.riak = riak;
+    this.workerPool = workerPool;
+  }
+
+  public RiakBuilder(BuilderSupport proxyBuilder, AsyncRiakTemplate riak) {
+    super(proxyBuilder);
+    this.riak = riak;
+  }
+
+  public RiakBuilder(Closure nameMappingClosure, BuilderSupport proxyBuilder, AsyncRiakTemplate riak) {
+    super(nameMappingClosure, proxyBuilder);
+    this.riak = riak;
+  }
+
+  public ExecutorService getWorkerPool() {
+    return workerPool;
+  }
+
+  public void setWorkerPool(ExecutorService workerPool) {
+    this.workerPool = workerPool;
+  }
+
+  @Override
+  protected void setParent(Object parent, Object child) {
+    log.debug("setParent/2 " + parent + " " + child);
+  }
+
+  @Override
+  protected Object createNode(Object name) {
+    log.debug("createNode/1 " + name);
+    return this;
+  }
+
+  @Override
+  protected Object createNode(Object name, Object value) {
+    log.debug("createNode/2 " + name + " " + value);
+    return null;  //To change body of implemented methods use File | Settings | File Templates.
+  }
+
+  @SuppressWarnings({"unchecked"})
+  @Override
+  protected Object createNode(Object name, Map attributes) {
+    log.debug("createNode/2 (Map) " + name + " " + attributes);
+    RiakOperation.Type type = RiakOperation.Type.valueOf(name.toString().toUpperCase());
+    if (null != type) {
+      RiakOperation op = new RiakOperation<Object>(riak, type);
+      Object o = attributes.get("bucket");
+      op.setBucket((null != o ? o.toString() : null));
+      o = attributes.get("key");
+      op.setKey((null != o ? o.toString() : null));
+      o = attributes.get("value");
+      op.setValue(o);
+      o = attributes.get("qos");
+      if (null != o) {
+        RiakQosParameters qos = new RiakQosParameters();
+        Map<String, Object> qosParams = (Map<String, Object>) o;
+        if (qosParams.containsKey("dw")) {
+          qos.setDurableWriteThreshold(qosParams.get("dw"));
+        }
+        if (qosParams.containsKey("w")) {
+          qos.setWriteThreshold(qosParams.get("w"));
+        }
+        if (qosParams.containsKey("r")) {
+          qos.setReadThreshold(qosParams.get("r"));
+        }
+        op.setQosParameters(qos);
+      }
+
+      o = attributes.get("wait");
+      if (null != o && o instanceof Long) {
+        op.setTimeout((Long) o);
+      }
+      return op;
+    }
+    return null;
+  }
+
+  @Override
+  protected Object createNode(Object name, Map attributes, Object value) {
+    log.debug("createNode/3");
+    return null;  //To change body of implemented methods use File | Settings | File Templates.
+  }
+
+  @Override
+  public Object invokeMethod(String methodName) {
+    log.debug("invokeMethod/1 " + methodName);
+    return super.invokeMethod(methodName);    //To change body of overridden methods use File | Settings | File Templates.
+  }
+
+  @SuppressWarnings({"unchecked"})
+  @Override
+  public Object invokeMethod(String methodName, Object arg) {
+    if (log.isDebugEnabled()) {
+      log.debug("invokeMethod: " + methodName + " " + arg);
+    }
+    if ("completed".equals(methodName) || "failed".equals(methodName)) {
+      RiakOperation<Object> op = (RiakOperation<Object>) getCurrent();
+      Object[] args = (Object[]) arg;
+      Map<String, Object> params;
+      Closure handler = null;
+      Closure guard = null;
+      for (Object o : args) {
+        if (o instanceof Map) {
+          params = (Map<String, Object>) o;
+          if (params.containsKey("when")) {
+            guard = (Closure) params.get("when");
+          }
+        } else if (o instanceof Closure) {
+          handler = (Closure) o;
+        }
+      }
+      op.addHandler(methodName, handler, guard);
+      return op;
+    }
+    return super.invokeMethod(methodName, arg);
+  }
+
+  @SuppressWarnings({"unchecked"})
+  @Override
+  protected void nodeCompleted(Object parent, Object node) {
+    log.debug("nodeCompleted: " + parent + " " + node);
+    if (null == parent && node instanceof RiakOperation) {
+      RiakOperation<Object> op = (RiakOperation<Object>) node;
+      try {
+        op.call();
+      } catch (Exception e) {
+        log.error(e.getMessage(), e);
+      }
+    } else {
+      super.nodeCompleted(parent, node);
+    }
+  }
+
+  @Override
+  protected Object postNodeCompletion(Object parent, Object node) {
+    log.debug("postNodeCompletion: " + parent + " " + node);
+    return super.postNodeCompletion(parent, node);
+  }
+}

spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakOperation.java

@@ -0,0 +1,238 @@
+/*
+ * Copyright (c) 2010 by J. Brisbin <jon@jbrisbin.com>
+ * Portions (c) 2010 by NPC International, Inc. or the
+ * original author(s).
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.data.keyvalue.riak.groovy;
+
+import groovy.lang.Closure;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.keyvalue.riak.core.AsyncKeyValueStoreOperation;
+import org.springframework.data.keyvalue.riak.core.AsyncRiakTemplate;
+import org.springframework.data.keyvalue.riak.core.KeyValueStoreMetaData;
+import org.springframework.data.keyvalue.riak.core.QosParameters;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author J. Brisbin <jon@jbrisbin.com>
+ */
+public class RiakOperation<T> implements Callable {
+
+  static enum Type {
+    SET, SETASBYTES, PUT, GET, GETASBYTES, CONTAINSKEY, DELETE
+  }
+
+  static String COMPLETED = "completed";
+  static String FAILED = "failed";
+
+  protected final Logger log = LoggerFactory.getLogger(getClass());
+
+  protected AsyncRiakTemplate riak;
+  protected Type type;
+  protected String bucket;
+  protected String key;
+  protected T value;
+  protected long timeout = -1L;
+  protected QosParameters qosParameters;
+  protected Map<String, List<GuardedClosure>> callbacks = new LinkedHashMap<String, List<GuardedClosure>>();
+  protected ClosureInvokingCallback callbackInvoker = new ClosureInvokingCallback();
+
+  public RiakOperation(AsyncRiakTemplate riak, Type type) {
+    this.riak = riak;
+    this.type = type;
+  }
+
+  public Type getType() {
+    return type;
+  }
+
+  public Map<String, List<GuardedClosure>> getCallbacks() {
+    return callbacks;
+  }
+
+  public QosParameters getQosParameters() {
+    return qosParameters;
+  }
+
+  public void setQosParameters(QosParameters qosParameters) {
+    this.qosParameters = qosParameters;
+  }
+
+  public String getBucket() {
+    return bucket;
+  }
+
+  public void setBucket(String bucket) {
+    this.bucket = bucket;
+  }
+
+  public String getKey() {
+    return key;
+  }
+
+  public void setKey(String key) {
+    this.key = key;
+  }
+
+  public T getValue() {
+    return value;
+  }
+
+  public void setValue(T value) {
+    this.value = value;
+  }
+
+  public long getTimeout() {
+    return timeout;
+  }
+
+  public void setTimeout(long timeout) {
+    this.timeout = timeout;
+  }
+
+  public void addHandler(String type, Closure handler, Closure guard) {
+    List<GuardedClosure> guardedClosures = callbacks.get(type);
+    if (null == guardedClosures) {
+      guardedClosures = new ArrayList<GuardedClosure>();
+      callbacks.put(type, guardedClosures);
+    }
+    guardedClosures.add(new GuardedClosure(handler, guard));
+  }
+
+  @SuppressWarnings({"unchecked"})
+  public T call() throws Exception {
+    Future<?> f = null;
+    switch (type) {
+      case GET:
+        f = riak.get(bucket, key, callbackInvoker);
+        break;
+      case GETASBYTES:
+        f = riak.getAsBytes(bucket, key, callbackInvoker);
+        break;
+      case PUT:
+        throw new IllegalStateException("PUT not yet implemented in AsyncRiakTemplate");
+      case SET:
+        f = riak.set(bucket, key, value, callbackInvoker);
+        break;
+      case SETASBYTES:
+        if (value instanceof byte[]) {
+          f = riak.setAsBytes(bucket, key, (byte[]) value, callbackInvoker);
+        } else {
+          log.error("Need to convert obj to byte array first!");
+        }
+        break;
+      case CONTAINSKEY:
+        f = riak.containsKey(bucket, key, callbackInvoker);
+        break;
+      case DELETE:
+        f = riak.delete(bucket, key, callbackInvoker);
+        break;
+    }
+    return null != f && timeout > 0 ? (T) f.get(timeout, TimeUnit.MILLISECONDS) : null;
+  }
+
+  class GuardedClosure {
+
+    private Closure delegate;
+    private Closure guard;
+
+    GuardedClosure(Closure delegate, Closure guard) {
+      this.delegate = delegate;
+      this.guard = guard;
+    }
+
+    public Closure getDelegate() {
+      return delegate;
+    }
+
+    public Closure getGuard() {
+      return guard;
+    }
+
+  }
+
+  class ClosureInvokingCallback implements AsyncKeyValueStoreOperation {
+
+    public void completed(KeyValueStoreMetaData meta, Object result) {
+      for (GuardedClosure cl : callbacks.get(COMPLETED)) {
+        boolean execute = true;
+
+        Closure guardExpr = cl.getGuard();
+        if (null != guardExpr) {
+          int noOfParams = guardExpr.getParameterTypes().length;
+          Object guardResult;
+          if (noOfParams == 2) {
+            guardResult = guardExpr.call(new Object[]{result, meta});
+          } else {
+            guardResult = guardExpr.call(result);
+          }
+          if (null != guardResult) {
+            if (guardResult instanceof Boolean) {
+              execute = (Boolean) guardResult;
+            } else {
+              execute = true;
+            }
+          }
+        }
+
+        if (execute) {
+          Closure callback = cl.getDelegate();
+          if (callback.getParameterTypes().length == 2) {
+            // Pass value and metadata
+            callback.call(new Object[]{result, meta});
+          } else {
+            callback.call(result);
+          }
+          break;
+        }
+      }
+    }
+
+    public void failed(Throwable error) {
+      for (GuardedClosure cl : callbacks.get(FAILED)) {
+        boolean execute = true;
+        Object param;
+
+        Closure guardExpr = cl.getGuard();
+        if (null != guardExpr) {
+          Object guardResult = guardExpr.call(error);
+          if (null != guardResult) {
+            if (guardResult instanceof Boolean) {
+              execute = (Boolean) guardResult;
+            } else {
+              execute = true;
+            }
+          }
+        }
+
+        if (execute) {
+          Closure callback = cl.getDelegate();
+          callback.call(error);
+        }
+      }
+    }
+
+  }
+
+}

spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/MapReduceJob.java

@@ -22,8 +22,8 @@ import java.util.List;
 import java.util.concurrent.Callable;
 
 /**
- * A generic interface to representing a Map/Reduce job to a data store that
- * supports that operation.
+ * A generic interface to representing a Map/Reduce job to a data store that supports that
+ * operation.
  *
  * @author J. Brisbin <jon@jbrisbin.com>
  */
@@ -53,6 +53,13 @@ public interface MapReduceJob<T> extends Callable {
    */
   MapReduceJob addPhase(MapReducePhase phase);
 
+  /**
+   * Get the list of phases for this job.
+   *
+   * @return
+   */
+  List<MapReducePhase> getPhases();
+
   /**
    * Set the static argument for this job.
    *

spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/RiakMapReduceJob.java

@@ -73,6 +73,10 @@ public class RiakMapReduceJob implements MapReduceJob {
     return this;
   }
 
+  public List<MapReducePhase> getPhases() {
+    return this.phases;
+  }
+
   public void setArg(Object arg) {
     this.arg = arg;
   }

spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplateSpec.groovy

@@ -0,0 +1,87 @@
+/*
+ * Copyright (c) 2010 by J. Brisbin <jon@jbrisbin.com>
+ * Portions (c) 2010 by NPC International, Inc. or the
+ * original author(s).
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.data.keyvalue.riak.core
+
+import java.util.concurrent.Future
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.context.ApplicationContext
+import org.springframework.test.context.ContextConfiguration
+import spock.lang.Specification
+
+/**
+ * @author J. Brisbin <jon@jbrisbin.com>
+ */
+@ContextConfiguration(locations = "/org/springframework/data/AsyncRiakTemplateTests.xml")
+class AsyncRiakTemplateSpec extends Specification {
+
+  @Autowired
+  ApplicationContext appCtx
+  @Autowired
+  AsyncRiakTemplate riak
+
+  def "Test async setWithMetaData"() {
+
+    given:
+    def obj = [test: "value", integer: 12]
+    def success = false
+    def failure = false
+    def testValue = "bad value"
+    def callback = [
+        completed: { v ->
+          success = true
+          testValue = v.get().test
+        },
+        failed: { e ->
+          failure = true
+        }
+    ] as AsyncKeyValueStoreOperation
+
+    when:
+    Future future = riak.setWithMetaData("test", "test", obj, null, null, callback)
+    println "Waiting for result: ${future.get()}"
+
+    then:
+    success && !failure
+    "value" == testValue
+
+  }
+
+  def "Test async getWithMetaData"() {
+
+    given:
+    def result = null
+    def callback = [
+        completed: { meta, v ->
+          println "got value: $meta $v"
+          result = v
+        },
+        failed: { e ->
+          println "got error: $e"
+        }
+    ] as AsyncKeyValueStoreOperation
+
+    when:
+    riak.getWithMetaData("test", "test", Map, callback).get()
+
+    then:
+    null != result
+
+  }
+
+}

spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/RiakBuilderSpec.groovy

@@ -0,0 +1,140 @@
+/*
+ * Copyright (c) 2010 by J. Brisbin <jon@jbrisbin.com>
+ * Portions (c) 2010 by NPC International, Inc. or the
+ * original author(s).
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.data.keyvalue.riak.core
+
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.context.ApplicationContext
+import org.springframework.data.keyvalue.riak.groovy.RiakBuilder
+import org.springframework.test.context.ContextConfiguration
+import spock.lang.Specification
+
+/**
+ * @author J. Brisbin <jon@jbrisbin.com>
+ */
+@ContextConfiguration(locations = "/org/springframework/data/AsyncRiakTemplateTests.xml")
+class RiakBuilderSpec extends Specification {
+
+  @Autowired
+  ApplicationContext appCtx
+  @Autowired
+  AsyncRiakTemplate riakTemplate
+
+  def "Test builder set"() {
+
+    given:
+    def obj = [test: "value", integer: 12]
+    def riak = new RiakBuilder(riakTemplate)
+    def result = null
+
+    when:
+    riak.set(bucket: "test", key: "test", qos: [dw: "all"], value: obj, wait: 3000L) {
+
+      completed(when: { v -> v.integer == 12 }) { v, meta ->
+        result = v.test
+      }
+      completed { v -> result = "otherwise" }
+
+      failed { e -> println "failure: $e" }
+
+    }
+
+    then:
+    "value" == result
+
+  }
+
+  def "Test builder get"() {
+
+    given:
+    def riak = new RiakBuilder(riakTemplate)
+    def result = null
+
+    when:
+    riak.get(bucket: "test", key: "test", wait: 3000L) {
+
+      completed(when: { v -> v.integer == 12 }) { v, meta ->
+        result = v.test
+      }
+      completed { v -> result = "otherwise" }
+
+      failed { e -> println "failure: $e" }
+
+    }
+
+    then:
+    "value" == result
+
+  }
+
+  def "Test builder setAsBytes"() {
+
+    given:
+    def obj = "test bytes".bytes
+    def riak = new RiakBuilder(riakTemplate)
+    def result = null
+
+    when:
+    riak.setAsBytes(bucket: "test", key: "test", value: obj, qos: [dw: "all"], wait: 3000L) {
+      completed { v -> result = "success" }
+      failed { e -> result = "failure" }
+    }
+
+    then:
+    null != result
+    "success" == result
+
+  }
+
+  def "Test builder get with bytes"() {
+
+    given:
+    def riak = new RiakBuilder(riakTemplate)
+    def result = null
+
+    when:
+    riak.get(bucket: "test", key: "test", wait: 3000L) {
+      completed { v -> result = v }
+      failed { e -> println "failure: $e" }
+    }
+
+    then:
+    null != result
+    "test bytes".bytes == result
+
+  }
+
+  def "Test builder delete"() {
+
+    given:
+    def riak = new RiakBuilder(riakTemplate)
+    def result = null
+
+    when:
+    riak.delete(bucket: "test", key: "test", wait: 3000L) {
+      completed { v -> result = v }
+      failed { e -> println "failure: $e" }
+    }
+
+    then:
+    null != result
+    result
+
+  }
+
+}

spring-data-riak/src/test/resources/org/springframework/data/AsyncRiakTemplateTests.xml

@@ -0,0 +1,34 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright (c) 2010 by J. Brisbin <jon@jbrisbin.com>
+  ~ Portions (c) 2010 by NPC International, Inc. or the
+  ~ original author(s).
+  ~
+  ~ Licensed under the Apache License, Version 2.0 (the "License");
+  ~ you may not use this file except in compliance with the License.
+  ~ You may obtain a copy of the License at
+  ~
+  ~ http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing, software
+  ~ distributed under the License is distributed on an "AS IS" BASIS,
+  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  ~ See the License for the specific language governing permissions and
+  ~ limitations under the License.
+  -->
+
+<beans xmlns="http://www.springframework.org/schema/beans"
+       xmlns:p="http://www.springframework.org/schema/p"
+       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+
+  <import resource="classpath:/META-INF/spring/app-context.xml"/>
+
+  <bean id="qos" class="org.springframework.data.keyvalue.riak.core.RiakQosParameters"
+        p:durableWriteThreshold="all"
+        p:writeThreshold="all"/>
+  <bean id="riakTemplate"
+        class="org.springframework.data.keyvalue.riak.core.AsyncRiakTemplate"
+        p:defaultQosParameters-ref="qos"/>
+
+</beans>