Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Added PUT for generating IDs to AsyncRiakTemplate, put and each on RiakBuilder, bucket/key on metadata, fixes in both template styles to accommodate new metadata.

Browse Source
This commit is contained in:
J. Brisbin
2010-12-21 14:17:00 -06:00
parent 0bd601f91f
commit 7e66bc36ec
7 changed files with 228 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -31,6 +31,7 @@ import org.springframework.web.client.ResourceAccessException;
import org.springframework.web.client.RestTemplate;
import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -112,6 +113,27 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck
callback));
}
public <B, V> Future<V> put(B bucket, V value, AsyncKeyValueStoreOperation<V> callback) {
return put(bucket, value, null, null, callback);
}
public <B, V> Future<V> put(B bucket, V value, Map<String, String> metaData, AsyncKeyValueStoreOperation<V> callback) {
return put(bucket, value, metaData, null, callback);
}
@SuppressWarnings({"unchecked"})
public <B, V> Future<V> put(B bucket, V value, Map<String, String> metaData, QosParameters qosParams, AsyncKeyValueStoreOperation<V> callback) {
Assert.notNull(bucket, "Bucket cannot be null");
String bucketName = (null != qosParams ? bucket.toString() + extractQosParameters(qosParams) : bucket
.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 AsyncPut<V>(bucketName, entity, callback));
}
public <B, K, V> Future<?> get(B bucket, K key, AsyncKeyValueStoreOperation<V> callback) {
return getWithMetaData(bucket, key, null, callback);
}
@@ -121,7 +143,10 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck
HttpHeaders headers;
try {
headers = restTemplate.headForHeaders(defaultUri, bucket, key);
return extractMetaData(headers);
RiakMetaData meta = extractMetaData(headers);
meta.setBucket((null != bucket ? bucket.toString() : null));
meta.setKey((null != key ? key.toString() : null));
return meta;
} catch (ResourceAccessException e) {
} catch (IOException e) {
throw new DataAccessResourceFailureException(e.getMessage(), e);
@@ -129,11 +154,36 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck
return null;
}
@SuppressWarnings({"unchecked"})
public <B> Future<?> getBucketSchema(B bucket, QosParameters qosParams, final AsyncKeyValueStoreOperation<Map<String, Object>> callback) {
Assert.notNull(bucket, "Bucket cannot be null");
Assert.notNull(callback, "Callback cannot be null");
String bucketName = (null != qosParams ? bucket.toString() + extractQosParameters(qosParams) : bucket
.toString());
return workerPool.submit(new AsyncGet(bucketName,
"?keys=true",
Map.class,
new AsyncKeyValueStoreOperation<Object>() {
@SuppressWarnings({"unchecked"})
public void completed(KeyValueStoreMetaData meta, Object result) {
callback.completed(meta, (Map<String, Object>) result);
}
public void failed(Throwable error) {
callback.failed(error);
}
}));
}
@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.");
Assert.notNull(key, "Cannot use a null key.");
Assert.notNull(callback, "Callback cannot be null");
if (null == requiredType) {
requiredType = (Class<T>) getType(bucketName, key.toString());
}
@@ -251,6 +301,43 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck
return setWithMetaData(bucket, key, value, metaData, null, callback);
}
protected class AsyncPut<V> implements Runnable {
private String bucket;
private HttpEntity<V> entity = null;
private AsyncKeyValueStoreOperation<V> callback = null;
public AsyncPut(String bucket, HttpEntity<V> entity, AsyncKeyValueStoreOperation<V> callback) {
this.bucket = bucket;
this.entity = entity;
this.callback = callback;
}
public void run() {
try {
URI location = getRestTemplate().postForLocation(defaultUri, entity, bucket, "");
String path = location.getPath();
String key = path.substring(path.lastIndexOf("/") + 1);
HttpHeaders headers = getRestTemplate().headForHeaders(defaultUri, bucket, key);
if (null != callback) {
RiakMetaData meta = extractMetaData(headers);
meta.setBucket((null != bucket ? bucket.toString() : null));
meta.setKey((null != key ? key.toString() : null));
callback.completed(meta, entity.getBody());
}
} catch (Throwable t) {
DataStoreOperationException dsoe = new DataStoreOperationException(t.getMessage(), t);
if (null != callback) {
callback.failed(dsoe);
} else {
defaultErrorHandler.failed(dsoe);
}
}
}
}
protected class AsyncPost<V> implements Runnable {
private String bucket;
@@ -280,7 +367,10 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck
entity));
}
if (null != callback) {
callback.completed(extractMetaData(result.getHeaders()), (V) result.getBody());
RiakMetaData meta = extractMetaData(result.getHeaders());
meta.setBucket((null != bucket ? bucket.toString() : null));
meta.setKey((null != key ? key.toString() : null));
callback.completed(meta, (V) result.getBody());
}
} catch (Throwable t) {
DataStoreOperationException dsoe = new DataStoreOperationException(t.getMessage(), t);
@@ -316,6 +406,8 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck
key);
if (result.hasBody()) {
RiakMetaData meta = extractMetaData(result.getHeaders());
meta.setBucket((null != bucket ? bucket.toString() : null));
meta.setKey((null != key ? key.toString() : null));
RiakValue<T> val = new RiakValue<T>(result.getBody(), meta);
if (useCache) {
cache.put(new SimpleBucketKeyPair<Object, Object>(bucket, key), val);

4
spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/KeyValueStoreMetaData.java
View File

@@ -29,6 +29,10 @@ import java.util.Map;
*/
public interface KeyValueStoreMetaData {
String getBucket();
String getKey();
/**
* Get the Content-Type of this object.
*

25
spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/RiakMetaData.java
View File

@@ -33,6 +33,8 @@ public class RiakMetaData implements KeyValueStoreMetaData {
private MediaType mediaType = MediaType.APPLICATION_JSON;
private Map<String, Object> properties;
private String bucket = null;
private String key = null;
public RiakMetaData(Map<String, Object> properties) {
this.properties = properties;
@@ -43,6 +45,29 @@ public class RiakMetaData implements KeyValueStoreMetaData {
this.properties = properties;
}
public RiakMetaData(MediaType mediaType, Map<String, Object> properties, String bucket, String key) {
this.mediaType = mediaType;
this.properties = properties;
this.bucket = bucket;
this.key = key;
}
public void setBucket(String bucket) {
this.bucket = bucket;
}
public void setKey(String key) {
this.key = key;
}
public String getBucket() {
return this.bucket;
}
public String getKey() {
return this.key;
}
public MediaType getContentType() {
return mediaType;
}

9
spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/RiakTemplate.java
View File

@@ -211,7 +211,10 @@ public class RiakTemplate extends AbstractRiakTemplate implements BucketKeyValue
HttpHeaders headers;
try {
headers = restTemplate.headForHeaders(defaultUri, bucket, key);
return extractMetaData(headers);
RiakMetaData meta = extractMetaData(headers);
meta.setBucket((null != bucket ? bucket.toString() : null));
meta.setKey((null != key ? key.toString() : null));
return meta;
} catch (ResourceAccessException e) {
} catch (IOException e) {
throw new DataAccessResourceFailureException(e.getMessage(), e);
@@ -309,7 +312,7 @@ public class RiakTemplate extends AbstractRiakTemplate implements BucketKeyValue
}
@SuppressWarnings({"unchecked"})
public <B, K> RiakValue<byte[]> getAsBytesWithMetaData(B bucket, K key) {
public <B, K> RiakValue<byte[]> getAsBytesWithMetaData(final B bucket, final K key) {
final RestTemplate restTemplate = getRestTemplate();
if (log.isDebugEnabled()) {
log.debug(String.format("GET object: bucket=%s, key=%s, type=byte[]",
@@ -343,6 +346,8 @@ public class RiakTemplate extends AbstractRiakTemplate implements BucketKeyValue
HttpHeaders headers = response.getHeaders();
RiakMetaData meta = extractMetaData(headers);
meta.setBucket((null != bucket ? bucket.toString() : null));
meta.setKey((null != key ? key.toString() : null));
RiakValue<byte[]> val = new RiakValue<byte[]>(out.toByteArray(),
meta);
return val;

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

@@ -150,11 +150,14 @@ public class RiakBuilder extends BuilderSupport {
if (log.isDebugEnabled()) {
log.debug("invokeMethod: " + methodName + " " + arg);
}
Object[] args = (Object[]) arg;
Map<String, Object> params;
Closure handler = null;
RiakOperation<Object> op;
if ("completed".equals(methodName) || "failed".equals(methodName)) {
RiakOperation<Object> op = (RiakOperation<Object>) getCurrent();
Object[] args = (Object[]) arg;
Map<String, Object> params;
Closure handler = null;
op = (RiakOperation<Object>) getCurrent();
Closure guard = null;
for (Object o : args) {
if (o instanceof Map) {
@@ -169,6 +172,7 @@ public class RiakBuilder extends BuilderSupport {
op.addHandler(methodName, handler, guard);
return op;
}
return super.invokeMethod(methodName, arg);
}
@@ -188,9 +192,11 @@ public class RiakBuilder extends BuilderSupport {
}
}
@SuppressWarnings({"unchecked"})
@Override
protected Object postNodeCompletion(Object parent, Object node) {
log.debug("postNodeCompletion: " + parent + " " + node);
return super.postNodeCompletion(parent, node);
}
}

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

@@ -21,6 +21,7 @@ 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.DataStoreOperationException;
import org.springframework.data.keyvalue.riak.core.AsyncKeyValueStoreOperation;
import org.springframework.data.keyvalue.riak.core.AsyncRiakTemplate;
import org.springframework.data.keyvalue.riak.core.KeyValueStoreMetaData;
@@ -30,9 +31,7 @@ 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;
import java.util.concurrent.*;
/**
* @author J. Brisbin <jon@jbrisbin.com>
@@ -40,7 +39,7 @@ import java.util.concurrent.TimeUnit;
public class RiakOperation<T> implements Callable {
static enum Type {
SET, SETASBYTES, PUT, GET, GETASBYTES, CONTAINSKEY, DELETE
SET, SETASBYTES, PUT, GET, GETASBYTES, CONTAINSKEY, DELETE, EACH
}
static String COMPLETED = "completed";
@@ -131,16 +130,19 @@ public class RiakOperation<T> implements Callable {
f = riak.getAsBytes(bucket, key, callbackInvoker);
break;
case PUT:
throw new IllegalStateException("PUT not yet implemented in AsyncRiakTemplate");
f = riak.put(bucket, value, callbackInvoker);
break;
case SET:
f = riak.set(bucket, key, value, callbackInvoker);
break;
case SETASBYTES:
byte[] bytes;
if (value instanceof byte[]) {
f = riak.setAsBytes(bucket, key, (byte[]) value, callbackInvoker);
bytes = (byte[]) value;
} else {
log.error("Need to convert obj to byte array first!");
bytes = riak.getConversionService().convert(value, byte[].class);
}
f = riak.setAsBytes(bucket, key, bytes, callbackInvoker);
break;
case CONTAINSKEY:
f = riak.containsKey(bucket, key, callbackInvoker);
@@ -148,6 +150,35 @@ public class RiakOperation<T> implements Callable {
case DELETE:
f = riak.delete(bucket, key, callbackInvoker);
break;
case EACH:
f = riak.getBucketSchema(bucket,
null,
new AsyncKeyValueStoreOperation<Map<String, Object>>() {
public void completed(KeyValueStoreMetaData meta, Map<String, Object> result) {
List<String> keys = (List<String>) result.get("keys");
for (String key : keys) {
try {
Future<?> getFut = riak.get(bucket, key, callbackInvoker);
if (timeout > 0) {
getFut.get(timeout, TimeUnit.MILLISECONDS);
} else if (timeout < 0) {
getFut.get();
}
} catch (InterruptedException e) {
throw new DataStoreOperationException(e.getMessage(), e);
} catch (ExecutionException e) {
throw new DataStoreOperationException(e.getMessage(), e);
} catch (TimeoutException e) {
throw new DataStoreOperationException(e.getMessage(), e);
}
}
}
public void failed(Throwable error) {
log.error(error.getMessage(), error);
}
});
break;
}
if (null != f) {
@@ -189,6 +220,9 @@ public class RiakOperation<T> implements Callable {
class ClosureInvokingCallback implements AsyncKeyValueStoreOperation {
public void completed(KeyValueStoreMetaData meta, Object result) {
if (!callbacks.containsKey(COMPLETED)) {
return;
}
for (GuardedClosure cl : callbacks.get(COMPLETED)) {
boolean execute = true;

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

@@ -119,21 +119,61 @@ class RiakBuilderSpec extends Specification {
}
def "Test builder delete"() {
def "Test builder put"() {
given:
def obj = [test: "value", integer: 12]
def riak = new RiakBuilder(riakTemplate)
def id = null
when:
riak.put(bucket: "test", qos: [dw: "all"], value: obj) {
completed { v, meta ->
id = meta.key
}
failed { e -> println "failure: $e" }
}
then:
null != id
}
def "Test builder each"() {
given:
def riak = new RiakBuilder(riakTemplate)
def result = null
def idCnt = 0
when:
riak.delete(bucket: "test", key: "test", wait: 3000L) {
completed { v -> result = v }
riak.each(bucket: "test") {
completed { v, meta -> idCnt++ }
failed { e -> println "failure: $e" }
}
then:
null != result
result
idCnt > 0
}
def "Test builder delete"() {
given:
def riak = new RiakBuilder(riakTemplate)
when:
riak.each(bucket: "test") {
completed { v, meta ->
delete(bucket: meta.bucket, key: meta.key)
}
failed { e -> println "failure: $e" }
}
then:
true
}