diff --git a/README.md b/README.md
index 68e873af6..408c26a59 100644
--- a/README.md
+++ b/README.md
@@ -94,15 +94,15 @@ For those in a hurry:
       -----
       
       MyObject obj = new MyObject("value1", "value2");
-      riakTemplate.set("mybucket:mykey", obj);
+      riakTemplate.set("mybucket", "mykey", obj);
       
-      Map returnObj = riakTemplate.getAsType("mybucket:mykey", Map.class);
+      Map returnObj = riakTemplate.getAsType("mybucket", "mykey", Map.class);
       
       Groovy:
       -----
       
       def obj = [first: "value1", second: "value2"]
-      riakTemplate.set([bucket: "mybucket", key: "mykey"], obj)
+      riakTemplate.set("mybucket", "mykey", obj)
       
 
 Contributing to Spring Data
diff --git a/spring-data-riak/pom.xml b/spring-data-riak/pom.xml
index bed2c1486..7936d8731 100644
--- a/spring-data-riak/pom.xml
+++ b/spring-data-riak/pom.xml
@@ -98,6 +98,10 @@
       <groupId>javax.mail</groupId>
       <artifactId>mail</artifactId>
     </dependency>
+    <dependency>
+      <groupId>javax.activation</groupId>
+      <artifactId>activation</artifactId>
+    </dependency>
 
     <dependency>
       <groupId>org.mockito</groupId>
@@ -129,7 +133,7 @@
         <artifactId>com.springsource.bundlor.maven</artifactId>
       </plugin>
 
-      <!-- For running Groovy/Spock tests 
+      <!-- For running Groovy/Spock tests
       <plugin>
         <groupId>org.spockframework</groupId>
         <artifactId>spock-maven</artifactId>
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 3ed61af18..aa7156779 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
@@ -176,6 +176,22 @@ public abstract class AbstractRiakTemplate extends RestGatewaySupport implements
     this.defaultQosParameters = defaultQosParameters;
   }
 
+  public String getHost() {
+    Matcher m = prefix.matcher(defaultUri);
+    if (m.matches()) {
+      return m.group(1);
+    }
+    return "localhost";
+  }
+
+  public Integer getPort() {
+    Matcher m = prefix.matcher(defaultUri);
+    if (m.matches()) {
+      return new Integer(m.group(2));
+    }
+    return 8098;
+  }
+
   /**
    * Extract the prefix from the URI for use in creating links.
    *
diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/io/RiakFile.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/io/RiakFile.java
new file mode 100644
index 000000000..eb697a8ca
--- /dev/null
+++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/io/RiakFile.java
@@ -0,0 +1,392 @@
+/*
+ * 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.io;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.keyvalue.riak.DataStoreOperationException;
+import org.springframework.data.keyvalue.riak.core.KeyValueStoreMetaData;
+import org.springframework.data.keyvalue.riak.core.RiakTemplate;
+import org.springframework.data.keyvalue.riak.core.RiakValue;
+
+import java.io.File;
+import java.io.FileFilter;
+import java.io.FilenameFilter;
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @author J. Brisbin <jon@jbrisbin.com>
+ */
+public class RiakFile<B, K> extends File {
+
+  private static final long serialVersionUID = 1L;
+  private static final Logger log = LoggerFactory.getLogger(RiakFile.class);
+
+  private RiakTemplate riak;
+  private B bucket;
+  private K key;
+
+  public RiakFile(RiakTemplate riak, B bucket, K key) throws URISyntaxException {
+    super(riak.getDefaultUri());
+    this.riak = riak;
+    this.bucket = bucket;
+    this.key = key;
+  }
+
+  public String getUriAsString(boolean includeKey) {
+    String protocol = riak.getDefaultUri().substring(0, riak.getDefaultUri().indexOf(":"));
+    String uri = String.format("%s://%s:%s%s/%s/%s",
+        protocol,
+        riak.getHost(),
+        riak.getPort(),
+        riak.getPrefix(),
+        bucket,
+        (includeKey ? key : ""));
+    return uri;
+  }
+
+  public RiakTemplate getRiak() {
+    return riak;
+  }
+
+  public void setRiak(RiakTemplate riak) {
+    this.riak = riak;
+  }
+
+  public B getBucket() {
+    return bucket;
+  }
+
+  public void setBucket(B bucket) {
+    this.bucket = bucket;
+  }
+
+  public K getKey() {
+    return key;
+  }
+
+  public void setKey(K key) {
+    this.key = key;
+  }
+
+  @Override
+  public String getName() {
+    return getUriAsString(true);
+  }
+
+  @Override
+  public String getParent() {
+    return getUriAsString(false);
+  }
+
+  @SuppressWarnings({"unchecked"})
+  @Override
+  public File getParentFile() {
+    try {
+      return new RiakFile(riak, bucket, "");
+    } catch (URISyntaxException e) {
+      log.error(e.getMessage(), e);
+    }
+    return null;
+  }
+
+  @Override
+  public String getPath() {
+    return getUriAsString(true);
+  }
+
+  @Override
+  public boolean isAbsolute() {
+    return true;
+  }
+
+  @Override
+  public String getAbsolutePath() {
+    return getUriAsString(true);
+  }
+
+  @Override
+  public File getAbsoluteFile() {
+    return this;
+  }
+
+  @Override
+  public String getCanonicalPath() throws IOException {
+    return getUriAsString(true);
+  }
+
+  @Override
+  public File getCanonicalFile() throws IOException {
+    return this;
+  }
+
+  @SuppressWarnings({"deprecation"})
+  @Override
+  public URL toURL() throws MalformedURLException {
+    return new URL(getUriAsString(true));
+  }
+
+  @Override
+  public URI toURI() {
+    try {
+      return new URI(getUriAsString(true));
+    } catch (URISyntaxException e) {
+      log.error(e.getMessage(), e);
+    }
+    return null;
+  }
+
+  @Override
+  public boolean canRead() {
+    return true;
+  }
+
+  @Override
+  public boolean canWrite() {
+    return true;
+  }
+
+  @Override
+  public boolean exists() {
+    return riak.containsKey(bucket, key);
+  }
+
+  @Override
+  public boolean isDirectory() {
+    return (key != null && "".equals(key));
+  }
+
+  @Override
+  public boolean isFile() {
+    return (key != null && !("".equals(key)));
+  }
+
+  @Override
+  public boolean isHidden() {
+    return false;
+  }
+
+  @Override
+  public long lastModified() {
+    KeyValueStoreMetaData meta = riak.getMetaData(bucket, key);
+    return (null != meta ? meta.getLastModified() : null);
+  }
+
+  @Override
+  public long length() {
+    return riak.getAsBytes(bucket, key).length;
+  }
+
+  @Override
+  public boolean createNewFile() throws IOException {
+    return true;
+  }
+
+  @Override
+  public boolean delete() {
+    return riak.delete(bucket, key);
+  }
+
+  @Override
+  public void deleteOnExit() {
+    // NO-OP
+  }
+
+  @Override
+  public String[] list() {
+    if (isDirectory()) {
+      Map<String, Object> schema = riak.getBucketSchema(bucket, true);
+      String baseUri = getUriAsString(false);
+      List<String> uris = new LinkedList<String>();
+      for (Object key : (List) ((Map) schema.get("props")).get("keys")) {
+        uris.add(baseUri + key);
+      }
+      return (String[]) uris.toArray();
+    }
+    return new String[]{};
+  }
+
+  @Override
+  public String[] list(FilenameFilter filenameFilter) {
+    List<String> uris = new LinkedList<String>();
+    for (String s : list()) {
+      if (filenameFilter.accept(this, s)) {
+        uris.add(s);
+      }
+    }
+    return (String[]) uris.toArray();
+  }
+
+  @SuppressWarnings({"unchecked"})
+  @Override
+  public File[] listFiles() {
+    List<RiakFile> uris = new LinkedList<RiakFile>();
+    for (String s : list()) {
+      try {
+        uris.add(new RiakFile(riak, bucket, s.substring(s.lastIndexOf("/"))));
+      } catch (URISyntaxException e) {
+        log.error(e.getMessage(), e);
+      }
+    }
+    return (File[]) uris.toArray();
+  }
+
+  @SuppressWarnings({"unchecked"})
+  @Override
+  public File[] listFiles(FilenameFilter filenameFilter) {
+    List<RiakFile> uris = new LinkedList<RiakFile>();
+    for (String s : list(filenameFilter)) {
+      try {
+        uris.add(new RiakFile(riak, bucket, s.substring(s.lastIndexOf("/"))));
+      } catch (URISyntaxException e) {
+        log.error(e.getMessage(), e);
+      }
+    }
+    return (File[]) uris.toArray();
+  }
+
+  @Override
+  public File[] listFiles(FileFilter fileFilter) {
+    return super.listFiles(fileFilter);    //To change body of overridden methods use File | Settings | File Templates.
+  }
+
+  @Override
+  public boolean mkdir() {
+    return true;
+  }
+
+  @Override
+  public boolean mkdirs() {
+    return true;
+  }
+
+  @SuppressWarnings({"unchecked"})
+  @Override
+  public boolean renameTo(File file) {
+    if (file instanceof RiakFile) {
+      RiakFile f = (RiakFile) file;
+      RiakValue<byte[]> v = riak.getAsBytesWithMetaData(bucket, key);
+      try {
+        riak.setWithMetaData(f.getBucket(), f.getKey(), v.get(),
+            (Map<String, String>) v.getMetaData());
+      } catch (DataStoreOperationException e) {
+        log.error(e.getMessage(), e);
+        return false;
+      }
+    } else {
+      throw new IllegalArgumentException("Renaming to a non-Riak file is not yet supported.");
+    }
+    return true;
+  }
+
+  @Override
+  public boolean setLastModified(long l) {
+    return false;
+  }
+
+  @Override
+  public boolean setReadOnly() {
+    return false;
+  }
+
+  @Override
+  public boolean setWritable(boolean b, boolean b1) {
+    return true;
+  }
+
+  @Override
+  public boolean setWritable(boolean b) {
+    return true;
+  }
+
+  @Override
+  public boolean setReadable(boolean b, boolean b1) {
+    return true;
+  }
+
+  @Override
+  public boolean setReadable(boolean b) {
+    return true;
+  }
+
+  @Override
+  public boolean setExecutable(boolean b, boolean b1) {
+    return false;
+  }
+
+  @Override
+  public boolean setExecutable(boolean b) {
+    return false;
+  }
+
+  @Override
+  public boolean canExecute() {
+    return false;
+  }
+
+  @Override
+  public long getTotalSpace() {
+    return super.getTotalSpace();
+  }
+
+  @Override
+  public long getFreeSpace() {
+    return super.getFreeSpace();
+  }
+
+  @Override
+  public long getUsableSpace() {
+    return super.getUsableSpace();
+  }
+
+  @Override
+  public int compareTo(File file) {
+    if (file instanceof RiakFile) {
+      return 0;
+    } else {
+      return -1;
+    }
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (o instanceof RiakFile) {
+      RiakFile rf = (RiakFile) o;
+      return (rf.getBucket().equals(bucket) && rf.getKey().equals(key));
+    }
+    return false;
+  }
+
+  @Override
+  public int hashCode() {
+    return super.hashCode();
+  }
+
+  @Override
+  public String toString() {
+    return getClass().getSimpleName() + "@" + getUriAsString(true);
+  }
+}
diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/io/RiakInputStream.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/io/RiakInputStream.java
new file mode 100644
index 000000000..279b711f7
--- /dev/null
+++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/io/RiakInputStream.java
@@ -0,0 +1,91 @@
+/*
+ * 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.io;
+
+import org.springframework.data.keyvalue.riak.core.RiakTemplate;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * An {@link java.io.InputStream} implementation that is backed by a resource residing in Riak.
+ *
+ * @author J. Brisbin <jon@jbrisbin.com>
+ */
+public class RiakInputStream<B, K> extends InputStream {
+
+  private RiakTemplate riak;
+  private B bucket;
+  private K key;
+  private ByteArrayInputStream in;
+
+  public RiakInputStream(RiakTemplate riak, B bucket, K key) {
+    this.riak = riak;
+    this.bucket = bucket;
+    this.key = key;
+    this.in = new ByteArrayInputStream(riak.getAsBytes(bucket, key));
+  }
+
+  @Override
+  public int read(byte[] bytes) throws IOException {
+    return in.read(bytes);
+  }
+
+  @Override
+  public int read(byte[] bytes, int i, int i1) throws IOException {
+    return in.read(bytes, i, i1);
+  }
+
+  @Override
+  public long skip(long l) throws IOException {
+    return in.skip(l);
+  }
+
+  @Override
+  public int available() throws IOException {
+    return in.available();
+  }
+
+  @Override
+  public void close() throws IOException {
+    in.close();
+  }
+
+  @Override
+  public void mark(int i) {
+    in.mark(i);
+  }
+
+  @Override
+  public void reset() throws IOException {
+    in.reset();
+  }
+
+  @Override
+  public boolean markSupported() {
+    return in.markSupported();
+  }
+
+  @Override
+  public int read() throws IOException {
+    return in.read();
+  }
+
+}
diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/io/RiakOutputStream.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/io/RiakOutputStream.java
new file mode 100644
index 000000000..bb82608a0
--- /dev/null
+++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/io/RiakOutputStream.java
@@ -0,0 +1,46 @@
+/*
+ * 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.io;
+
+import org.springframework.data.keyvalue.riak.core.RiakTemplate;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+
+/**
+ * @author J. Brisbin <jon@jbrisbin.com>
+ */
+public class RiakOutputStream<B, K> extends ByteArrayOutputStream {
+
+  private RiakTemplate riak;
+  private B bucket;
+  private K key;
+
+  public RiakOutputStream(RiakTemplate riak, B bucket, K key) {
+    this.riak = riak;
+    this.bucket = bucket;
+    this.key = key;
+  }
+
+  @Override
+  public void flush() throws IOException {
+    super.flush();
+    riak.setAsBytes(bucket, key, toByteArray());
+  }
+}
diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/io/RiakResource.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/io/RiakResource.java
new file mode 100644
index 000000000..a5e4b172b
--- /dev/null
+++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/io/RiakResource.java
@@ -0,0 +1,138 @@
+/*
+ * 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.io;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.core.io.Resource;
+import org.springframework.core.io.UrlResource;
+import org.springframework.data.keyvalue.riak.core.RiakTemplate;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+
+/**
+ * An implementation of {@link org.springframework.core.io.UrlResource} that is backed by a
+ * resource in Riak.
+ *
+ * @author J. Brisbin <jon@jbrisbin.com>
+ */
+public class RiakResource<B, K> extends UrlResource {
+
+  private static final Logger log = LoggerFactory.getLogger(RiakResource.class);
+
+  private RiakTemplate riak;
+  private B bucket;
+  private K key;
+  private String description;
+
+  public RiakResource(RiakTemplate riak, B bucket, K key) throws MalformedURLException {
+    super(riak.getDefaultUri());
+    this.bucket = bucket;
+    this.key = key;
+  }
+
+  public RiakTemplate getRiakTemplate() {
+    return this.riak;
+  }
+
+  public B getBucket() {
+    return bucket;
+  }
+
+  public K getKey() {
+    return key;
+  }
+
+  @SuppressWarnings({"unchecked"})
+  @Override
+  public URL getURL() throws IOException {
+    try {
+      return new URL(new RiakFile(riak, bucket, key).getUriAsString(true));
+    } catch (URISyntaxException e) {
+      log.error(e.getMessage(), e);
+    }
+    return null;
+  }
+
+  @SuppressWarnings({"unchecked"})
+  @Override
+  public URI getURI() throws IOException {
+    try {
+      return new RiakFile(riak, bucket, key).toURI();
+    } catch (URISyntaxException e) {
+      log.error(e.getMessage(), e);
+    }
+    return null;
+  }
+
+  @SuppressWarnings({"unchecked"})
+  @Override
+  public Resource createRelative(String relativePath) throws MalformedURLException {
+    if (relativePath.startsWith("../")) {
+      return new RiakResource(riak, bucket, relativePath.substring(3));
+    } else if (!relativePath.startsWith("/")) {
+      return new RiakResource(riak, bucket, relativePath);
+    }
+    return null;
+  }
+
+  @SuppressWarnings({"unchecked"})
+  @Override
+  public String getFilename() {
+    try {
+      return new RiakFile(riak, bucket, key).getUriAsString(true);
+    } catch (URISyntaxException e) {
+      log.error(e.getMessage(), e);
+    }
+    return riak.getDefaultUri();
+  }
+
+  public void setDescription(String description) {
+    this.description = description;
+  }
+
+  @Override
+  public String getDescription() {
+    return this.description;
+  }
+
+  @SuppressWarnings({"unchecked"})
+  @Override
+  public File getFile() throws IOException {
+    try {
+      return new RiakFile(riak, bucket, key);
+    } catch (URISyntaxException e) {
+      log.error(e.getMessage(), e);
+    }
+    return null;
+  }
+
+  @SuppressWarnings({"unchecked"})
+  @Override
+  public InputStream getInputStream() throws IOException {
+    return new RiakInputStream(riak, bucket, key);
+  }
+
+}
diff --git a/spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/RiakTemplateSpec.groovy b/spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/RiakTemplateSpec.groovy
index 82fd81625..a911e64ac 100644
--- a/spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/RiakTemplateSpec.groovy
+++ b/spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/RiakTemplateSpec.groovy
@@ -19,6 +19,7 @@ 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.core.io.RiakFile
 import org.springframework.data.keyvalue.riak.mapreduce.JavascriptMapReduceOperation
 import org.springframework.data.keyvalue.riak.mapreduce.MapReduceJob
 import org.springframework.data.keyvalue.riak.mapreduce.RiakMapReducePhase
@@ -39,7 +40,7 @@ class RiakTemplateSpec extends Specification {
   int run = 1
   @Shared def riakBin = System.properties["bamboo.RIAK_BIN"] ?: "/usr/sbin/riak"
   @Shared def p
-
+/*
   def setupSpec() {
     p = "$riakBin start".execute()
     p.waitFor()
@@ -50,6 +51,7 @@ class RiakTemplateSpec extends Specification {
     p = "$riakBin stop".execute()
     p.waitFor()
   }
+*/
 
   def "Test Map object"() {
 
@@ -235,6 +237,25 @@ class RiakTemplateSpec extends Specification {
 
   }
 
+  def "Test RiakFile"() {
+
+    given:
+    def file = new RiakFile(riak, "test", "test")
+
+    when:
+    def exists = file.exists()
+
+    then:
+    exists
+
+    when:
+    def content = file.toURI().toURL().openConnection().getContent()
+
+    then:
+    null != content
+
+  }
+
   def "Test delete key"() {
 
     given:
diff --git a/src/docbkx/index.xml b/src/docbkx/index.xml
index a24aacbf0..a6fbafe92 100644
--- a/src/docbkx/index.xml
+++ b/src/docbkx/index.xml
@@ -12,6 +12,11 @@
                 <surname>Leau</surname>
                 <affiliation>SpringSource</affiliation>
             </author>
+            <author>
+                <firstname>Jon</firstname>
+                <surname>Brisbin</surname>
+                <affiliation>NPC International, Inc.</affiliation>
+            </author>
         </authorgroup>
 
         <legalnotice>
@@ -42,6 +47,7 @@
        
       <xi:include href="reference/introduction.xml"/>
       <xi:include href="reference/redis.xml"/>
+      <xi:include href="reference/riak.xml"/>
     </part>
 
     <!--
diff --git a/src/docbkx/reference/riak.xml b/src/docbkx/reference/riak.xml
new file mode 100644
index 000000000..15edc0c1e
--- /dev/null
+++ b/src/docbkx/reference/riak.xml
@@ -0,0 +1,200 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE preface PUBLIC "-//OASIS//DTD DocBook XML V4.4//EN"
+                 "http://www.oasis-open.org/docbook/xml/4.4/docbookx.dtd">
+<chapter id="riak">
+  <title>Riak Support</title>
+  
+  <para><ulink url="https://wiki.basho.com/display/RIAK/Riak">Riak</ulink> is a Key/Value datastore that supports <ulink url="https://wiki.basho.com/pages/viewpage.action?pageId=1245320">Internet-scale data replication</ulink> for high performance and high availability. Spring Data Key/Value (SDKV) provides access to the Riak datastore over the <ulink url="https://wiki.basho.com/display/RIAK/REST+API">HTTP REST API</ulink> using a built-in driver based on Spring 3.0's <ulink url="http://static.springsource.org/spring/docs/3.0.x/spring-framework-reference/html/remoting.html#rest-resttemplate"><classname>RestTemplate</classname></ulink>. In addition to making Key/Value datastore access easier from Java, the <classname>RiakTemplate</classname> has been designed, from the ground up, to be used from alternative JVM languages like <ulink url="http://groovy.codehaus.org/">Groovy</ulink> or <ulink url="http://jruby.org/">JRuby</ulink>.</para> 
+  
+  <para>Since the SDKV support for Riak uses the stateless REST API, there are no connection factories to manage or other stateful objects to keep tabs on. The helper you'll spend the most time working with is likely the thread-safe <classname>RiakTemplate</classname> or <classname>RiakKeyValueTemplate</classname>. Your choice of which to use will depend on how you want to manage buckets and keys. SDKV supports two ways to interact with Riak. If you want to use the convention you're likely already familiar with, namely of storing an entry with a given key in a "bucket" by passing the bucket and key name separately, you'll want to use the <classname>RiakTemplate</classname>. If you want to use a single object to represent your bucket and key pair, you can use the <classname>RiakKeyValueTemplate</classname>. It supports a key object that is encoded using one of several different methods:
+  <itemizedlist>
+    <listitem><emphasis>Using a <classname>String</classname></emphasis> - You can concatenate two strings, separated by a colon: "mybucket:mykey".</listitem>
+    <listitem><emphasis>Using a <interfacename>BucketKeyPair</interfacename></emphasis> - You can pass an instance of <interfacename>BucketKeyPair</interfacename>, like <classname>SimpleBucketKeyPair</classname>.</listitem>
+    <listitem><emphasis>Using a <interfacename>Map</interfacename></emphasis> - You can pass a <interfacename>Map</interfacename> with keys for "bucket" and "key".</listitem>
+  </itemizedlist>
+  </para>
+  
+  <section id="riak:template:config">
+    <title>Configuring the <classname>RiakTemplate</classname></title>
+    
+    <para>This is likely the easiest path to using SDKV for Riak, as the bucket and key are passed separately. The examples that follow will assume you're using this version of the the template.</para>
+    
+    <para>There are only two options you need to set to specify the Riak server to use in your <classname>RiakTemplate</classname> object: "defaultUri" and "mapReduceUri". Encoded with the URI should be placeholders for the bucket and the key, which will be filled in by the <classname>RestTemplate</classname> when the request is made. 
+    
+      <important><para>You can also turn the internal, ETag-based object cache off by setting <literal>useCache="false"</literal>. It's generally recommended, however, to leave the internal cache on as the ETag matching will pick up any changes made to the entry on the Riak side and your application will benefit from greatly-increased performance for often-requested objects.</para></important>
+    
+      <programlisting language="xml"><![CDATA[<?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:useCache="true"/>
+    
+</beans>]]></programlisting>
+    </para>
+    
+    <section id="riak:template:config:advanced">
+      <title>Advanced Template Configuration</title>
+      
+      <para>There are a couple additional properties on the <classname>RiakTemplate</classname> that can be changed from their defaults. If you want to specify your own <ulink url="http://static.springsource.org/spring/docs/3.0.x/spring-framework-reference/html/validation.html#core-convert-ConversionService-API">ConversionService</ulink> to use when converting objects for storage inside Riak, then set it on the "conversionService" property:
+
+        <programlisting language="xml"><![CDATA[<?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>]]></programlisting>
+      </para>
+    
+      <para>Depending on the application, it might be useful to set default Quality-of-Service parameters. In Riak paralance, these are the <ulink url="https://wiki.basho.com/display/RIAK/REST+API#RESTAPI-Setbucketproperties">"dw", "w", and "r" parameters</ulink>. 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 <interfacename>QosParameters</interfacename> to the set/get operation you're performing.
+    
+        <programlisting language="xml"><![CDATA[<?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>]]></programlisting>
+      </para>
+    
+      <para>It might also be necessary to replace the default <classname>ExecutorService</classname> (by default a cached <ulink url="http://download.oracle.com/javase/6/docs/api/java/util/concurrent/ThreadPoolExecutor.html">ThreadPoolExecutor</ulink>) with an executor you've explicitly configured. Set your <classname>ExecutorService</classname> on the template's "executorService" property.</para>
+    
+    </section>
+  </section>
+  
+  <section id="riak:template:objects">
+    <title>Working with Objects using the <classname>RiakTemplate</classname></title>
+    
+    <para>One of the primary goals of the SDKV project is to make accessing Key/Value stores easier for the developer by taking away the mundane tasks of basic IO, buffering, type conversion, exception handling, and sundry other logistical concerns so the developer can focus on creating great applications. SDKV for Riak works toward this goal by making basic persistence and data access as easy as using a <classname>Map</classname>.</para>
+    
+    <section id="riak:template:objects:get">
+      <title>Saving data into Riak</title>
+      
+      <para>To store data in Riak, use one of the six different <literal>set</literal> methods:
+        <programlisting language="java"><![CDATA[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
+    }
+
+}
+        ]]></programlisting>
+      </para>
+    
+      <para>Additionally, there is a <literal>setWithMetaData</literal> method that takes a <classname>Map</classname> of metadata that will be set as the outgoing HTTP headers. To set <ulink url="https://wiki.basho.com/display/RIAK/REST+API#RESTAPI-Storeaneworexistingobjectwithakey">custom metadata</ulink>, your key should be prefixed with <literal>X-Riak-Meta-</literal> e.g. <literal>X-Riak-Meta-Custom-Header</literal>.</para>
+    </section>
+    
+    <section id="riak:template:objects:set">
+      <title>Retrieving data from Riak</title>
+    
+      <para>Retrieving data from Riak is just as easy. There are actually 13 different <literal>get</literal> methods on <classname>RiakTemplate</classname> that give the developer a wide range options for accessing and converting your data.</para>
+    
+      <para>Assuming you've stored a POJO using an appropriate <literal>set</literal> method, you can retrieve that object from Riak using a <literal>get</literal>:
+        <programlisting language="java"><![CDATA[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); // No conversion at all
+      }
+
+  }
+        ]]></programlisting>
+      </para>
+    </section>
+  </section>
+  
+  <section id="riak:io">
+    <title>Working with streams</title>
+
+    <para>SDKV for Riak includes a couple of useful helper objects to make reading and writing plain text or binary data in Riak really easy. If you want to store a file in Riak, then you can create a <classname>RiakOutputStream</classname> and simply write your data to it (making sure to call the "flush" method, which actually sends the data to Riak).
+    
+      <programlisting language="java"><![CDATA[import org.springframework.data.keyvalue.riak.core.RiakTemplate;
+import org.springframework.data.keyvalue.riak.core.io.RiakOutputStream;
+        
+public class Example {
+
+    @Autowired
+    RiakTemplate riak;
+
+    public void writeToRiak(String bucket, String key, String data) throws Exception {
+        OutputStream out = new RiakOutputStream(riak, bucket, key);
+        try {
+            out.write(data.getBytes());
+        } finally {
+            out.flush();
+            out.close();
+        }
+    }
+    
+}
+      ]]></programlisting>
+      
+      <para>Reading data from Riak is similarly easy. SDKV provides a <literal>java.io.File</literal> subclass that represents a resource in Riak. There's also a Spring IO Resource abstraction called <classname>RiakResource</classname> that can be used anywhere a <ulink url="http://static.springsource.org/spring/docs/3.0.x/spring-framework-reference/html/resources.html"><interfacename>Resource</interfacename></ulink> is required. There's also an <classname>InputStream</classname> implementation called <classname>RiakInputStream</classname>.</para>
+      
+      <programlisting language="java"><![CDATA[import org.springframework.data.keyvalue.riak.core.RiakTemplate;
+import org.springframework.data.keyvalue.riak.core.io.RiakInputStream;
+        
+public class Example {
+
+    @Autowired
+    RiakTemplate riak;
+
+    public String readFromRiak(String bucket, String key) throws Exception {
+        InputStream in = new RiakInputStream(riak, bucket, key);
+        String data;
+        ...read data and work with it...
+        return data;
+    }
+    
+}
+      ]]></programlisting>
+      
+    </para>
+  </section>
+</chapter>
\ No newline at end of file