There are a couple additional properties on the RiakTemplate that can be changed from their defaults. If you want to specify your own ConversionService to use when converting objects for storage inside Riak, then set it on the "conversionService" property:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:p="http://www.springframework.org/schema/p"
       xsi:schemaLocation="http://www.springframework.org/schema/beans 
       http://www.springframework.org/schema/beans/spring-beans.xsd">

    <bean id="conversionService" class="com.mycompany.convert.MyConversionService"/> 
    <bean id="riakTemplate" class="org.springframework.data.keyvalue.riak.core.RiakTemplate" 
          p:defaultUri="http://localhost:8098/riak/{bucket}/{key}"
          p:mapReduceUri="http://localhost:8098/mapred"
          p:conversionService-ref="conversionService"/>

</beans>

Depending on the application, it might be useful to set default Quality-of-Service parameters. In Riak paralance, these are the "dw", "w", and "r" parameters. They can be set to an integer representing the number of vnodes that need to report having received the data before declaring the operation a success, or the string "one", "all", or (the default) "quorum". These values can be overridden by passing a different set of QosParameters to the set/get operation you're performing.

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:p="http://www.springframework.org/schema/p"
       xsi:schemaLocation="http://www.springframework.org/schema/beans 
       http://www.springframework.org/schema/beans/spring-beans.xsd">

    <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.RiakTemplate" 
          p:defaultUri="http://localhost:8098/riak/{bucket}/{key}"
          p:mapReduceUri="http://localhost:8098/mapred"
          p:defaultQosParameters-ref="qos"/>

</beans>

You can also set a specific ClassLoader to use when loading objects from Riak. Just set the classLoader property:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:p="http://www.springframework.org/schema/p"
       xsi:schemaLocation="http://www.springframework.org/schema/beans 
       http://www.springframework.org/schema/beans/spring-beans.xsd">

    <bean id="riakTemplate" class="org.springframework.data.keyvalue.riak.core.RiakTemplate" 
          p:defaultUri="http://localhost:8098/riak/{bucket}/{key}"
          p:mapReduceUri="http://localhost:8098/mapred"
          p:classLoader-ref="customClassLoader"/>

</beans>

To store data in Riak, use one of the six different set methods:

import org.springframework.data.keyvalue.riak.core.RiakTemplate;

public class Example {

    @Autowired
    RiakTemplate riak;

    public void setData(String bucket, String key, String data) throws Exception {
        riak.set(bucket, key, data); // Set as Content-Type: text/plain
        riak.setAsBytes(bucket, key, data.getBytes()); // Set as Content-Type: application/octet-stream
    }
  
    public void setData(String bucket, String key, MyPojo data) throws Exception {
        riak.set(bucket, key, data); // Converted to JSON automatically, Content-Type: application/json
    }

}
        

Additionally, there is a setWithMetaData method that takes a Map of metadata that will be set as the outgoing HTTP headers. To set custom metadata, your key should be prefixed with X-Riak-Meta- e.g. X-Riak-Meta-Custom-Header.

import org.springframework.data.keyvalue.riak.core.RiakTemplate;

public class Example {

    @Autowired
    RiakTemplate riak;

    public String setData(String bucket, String data) throws Exception {
        String id = riak.put(bucket, data); // Returns the generated ID
        return id;
    }

}
          

Retrieving data from Riak is just as easy. There are actually 13 different get methods on RiakTemplate that give the developer a wide range options for accessing and converting your data.

Assuming you've stored a POJO using an appropriate set method, you can retrieve that object from Riak using a get:

import org.springframework.data.keyvalue.riak.core.RiakTemplate;

  public class Example {

      @Autowired
      RiakTemplate riak;

      public void getData(String bucket, String key) throws Exception {
          // What you get depends on Content-Type. 
          // application/json=Map, text/plain=String, etc...
          Object o = riak.get(bucket, key); 
        
          // If your entry is Content-Type: application/json...
          // It will automatically be converted when retrieved.
          MyPojo s = riak.getAsType(bucket, key, MyPojo.class);
        
          // If your entry is Content-Type: application/octet-stream,
          // you can access the raw bytes.
          byte[] b = riak.getAsBytes(bucket, key); 
      }

  }
        

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 method will try to infer it from the bucket name. If the bucket name is not a valid class name, it will default to using a java.util.Map.

To link walk a relationship and return a list of custom POJOs, you would do something like this:

@Autowired
RiakTemplate riak;

List<MyPojo> result = riak.linkWalk("sourcebucket", "sourcekey", "tagname", MyPojo.class);
        

Riak will accept either a string denoting the bucket in which to get the list of keys to operate on, or a List of Lists denoting the bucket/key pairs to operate on while executing this Map/Reduce job. If you call the addInputs method on the job passing a List with a single string entry, the job will assume you want to operate on an entire bucket. Otherwise, you'll need to pass a multi-dimensional List of bucket/key pairs.

To operate on an entire bucket:

@Autowired
RiakTemplate riak;

RiakMapReduceJob job = riak.createMapReduceJob();
List<String> bucket = new ArrayList<String>() {{
  add("mybucket");
}};
job.addInputs(bucket); // Will M/R entire bucket
        

To operate on a set of keys:

import org.springframework.data.keyvalue.riak.mapreduce.*;
          
@Autowired
RiakTemplate riak;

RiakMapReduceJob job = riak.createMapReduceJob();

List<String> pair = new ArrayList<String>() {{
  add("mybucket");
  add("mykey");
}};
List<List<String>> keys = new ArrayList<List<String>>() {{
  add(pair);
}};
job.addInputs(keys); // Will M/R only specified keys
        

Map/Reduce operations in Riak are broken up into phases. Phases contain a MapReduceOperation. There are currently two implementations to handle Javascript or Erlang M/R operations: JavascriptMapReduceOperation and ErlangMapReduceOperation.

An example Map/Reduce job defining a single "map" phase defined in anonymous Javascript might look like this:

import org.springframework.data.keyvalue.riak.mapreduce.*;

@Autowired
RiakTemplate riak;

RiakMapReduceJob job = riak.createMapReduceJob();
List<String> bucket = new ArrayList<String>() {{
  add("mybucket");
}};

job.addInputs(bucket); // M/R the entire bucket

MapReduceOperation mapOper = new JavascriptMapReduceOperation("function(v){ ...M/R function body... }");
MapReducePhase mapPhase = new RiakMapReducePhase("map", "javascript", mapOper);

job.addPhase(mapPhase);
        

To execute a configured job on your Riak server, use either the synchronous execute or asynchronous submit methods of your configured RiakTemplate:

Object o = riak.execute(job); // Results of last Map or Reduce phase. Should be a List<?>

...or...

List<MyPojo> o = riak.execute(job, MyPojo.class); // Coerce to given type

...or...

Future<List<?>> f = riak.submit(job); // Job runs in a separate thread
        

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.

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:

To perform an asynchronous get on a JSON-serialized Map object which returns a custom object from the callback, you'd do something like:

@Autowired
AsyncRiakTemplate riak;

Future<MyObject> future = riak.get("mybucket", "mykey", new AsyncKeyValueStoreOperation<Map, MyObject>() {

  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();
        

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):

def riak = new RiakBuilder(asyncRiakTemplate)
riak {
  test {
    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 }}
    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:

	Important
	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.

def riak = new RiakBuilder(asyncRiakTemplate)

def myobj = riak.set(bucket: "mybucket", key: "mykey", qos: ["dw": "all"])
          

def riak = new RiakBuilder(asyncRiakTemplate)

def myobj = riak.get(bucket: "mybucket", key: "mykey")