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

Added QosParameters

Browse Source
This commit is contained in:
J. Brisbin
2010-12-02 09:56:32 -06:00
parent 7c4e70407d
commit c9ba1e2365
5 changed files with 164 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
spring-data-riak/src/main/java/org/springframework/data/riak/core/KeyValueStoreOperations.java
View File

@@ -35,6 +35,18 @@ public interface KeyValueStoreOperations {
*/
<K, V> KeyValueStoreOperations set(K key, V value);
/**
* Variation on set() that allows the user to specify {@link org.springframework.data.riak.core.QosParameters}.
*
* @param key
* @param value
* @param qosParams
* @param <K>
* @param <V>
* @return
*/
<K, V> KeyValueStoreOperations set(K key, V value, QosParameters qosParams);
/**
* Set a value as a byte array at a specified key.
*
@@ -44,6 +56,20 @@ public interface KeyValueStoreOperations {
*/
<K> KeyValueStoreOperations setAsBytes(K key, byte[] value);
/**
* Variation on setWithMetaData() that allows the user to pass {@link
* org.springframework.data.riak.core.QosParameters}.
*
* @param key
* @param value
* @param metaData
* @param qosParams
* @param <K>
* @param <V>
* @return
*/
<K, V> KeyValueStoreOperations setWithMetaData(K key, V value, Map<String, String> metaData, QosParameters qosParams);
// Get operations
/**
@@ -237,4 +263,7 @@ public interface KeyValueStoreOperations {
*/
<B> Map<String, Object> getBucketSchema(B bucket, boolean listKeys);
<K> KeyValueStoreOperations setAsBytes(K key, byte[] value, QosParameters qosParams);
<K, V> KeyValueStoreOperations setWithMetaData(K key, V value, Map<String, String> metaData);
}

34
spring-data-riak/src/main/java/org/springframework/data/riak/core/QosParameters.java Normal file
View File

@@ -0,0 +1,34 @@
package org.springframework.data.riak.core;
/**
* Specify Quality Of Service parameters.
*
* @author J. Brisbin <jon@jbrisbin.com>
*/
public interface QosParameters {
/**
* Instruct the server on the read threshold.
*
* @param <T>
* @return
*/
public <T> T getReadThreshold();
/**
* Instruct the server on the normal write threshold.
*
* @param <T>
* @return
*/
public <T> T getWriteThreshold();
/**
* Instruct the server on the durable write threshold.
*
* @param <T>
* @return
*/
public <T> T getDurableWriteThreshold();
}

39
spring-data-riak/src/main/java/org/springframework/data/riak/core/RiakQosParameters.java Normal file
View File

@@ -0,0 +1,39 @@
package org.springframework.data.riak.core;
/**
* A generic class for specifying Quality Of Service parameters on operations.
*
* @author J. Brisbin <jon@jbrisbin.com>
*/
@SuppressWarnings({"unchecked"})
public class RiakQosParameters implements QosParameters {
public Object readThreshold = null;
public Object writeThreshold = null;
public Object durableWriteThreshold = null;
public <T> void setReadThreshold(T readThreshold) {
this.readThreshold = readThreshold;
}
public <T> void setWriteThreshold(T writeThreshold) {
this.writeThreshold = writeThreshold;
}
public <T> void setDurableWriteThreshold(T durableWriteThreshold) {
this.durableWriteThreshold = durableWriteThreshold;
}
public <T> T getReadThreshold() {
return (T) this.readThreshold;
}
public <T> T getWriteThreshold() {
return (T) this.writeThreshold;
}
public <T> T getDurableWriteThreshold() {
return (T) this.durableWriteThreshold;
}
}

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

@@ -141,6 +141,10 @@ public class RiakTemplate extends RestGatewaySupport implements KeyValueStoreOpe
* A list of resolvers to turn a single object into a {@link BucketKeyPair}.
*/
protected List<BucketKeyResolver> bucketKeyResolvers;
/**
* The default QosParameters to use for all operations through this template.
*/
protected QosParameters defaultQosParameters = null;
/**
* Take all the defaults.
@@ -156,6 +160,7 @@ public class RiakTemplate extends RestGatewaySupport implements KeyValueStoreOpe
*/
public RiakTemplate(ClientHttpRequestFactory requestFactory) {
super(requestFactory);
setRestTemplate(new RestTemplate());
}
/**
@@ -234,17 +239,27 @@ public class RiakTemplate extends RestGatewaySupport implements KeyValueStoreOpe
return setWithMetaData(key, value, null);
}
public <K, V> KeyValueStoreOperations set(K key, V value, QosParameters qosParams) {
return setWithMetaData(key, value, null, qosParams);
}
public <K> KeyValueStoreOperations setAsBytes(K key, byte[] value) {
return setAsBytes(key, value, null);
}
public <K> KeyValueStoreOperations setAsBytes(K key, byte[] value, QosParameters qosParams) {
Assert.notNull(key, "Can't store an object with a NULL key.");
BucketKeyPair bucketKeyPair = resolveBucketKeyPair(key, value);
String bucketName = (null != bucketKeyPair.getBucket() ? bucketKeyPair.getBucket()
.toString() : "bytes");
String keyName = (null != qosParams ? bucketKeyPair.getKey()
.toString() + extractQosParameters(qosParams) : bucketKeyPair.getKey().toString());
RestTemplate restTemplate = getRestTemplate();
HttpHeaders headers = new HttpHeaders();
headers.set("X-Riak-ClientId", RIAK_CLIENT_ID);
headers.setContentType(MediaType.APPLICATION_OCTET_STREAM);
HttpEntity<byte[]> entity = new HttpEntity<byte[]>(value, headers);
restTemplate.put(defaultUri, entity, bucketName, bucketKeyPair.getKey());
restTemplate.put(defaultUri, entity, bucketName, keyName);
if (log.isDebugEnabled()) {
log.debug(String.format("PUT byte[]: bucket=%s, key=%s",
bucketKeyPair.getBucket(),
@@ -253,8 +268,10 @@ public class RiakTemplate extends RestGatewaySupport implements KeyValueStoreOpe
return this;
}
public <K, V> KeyValueStoreOperations setWithMetaData(K key, V value, Map<String, String> metaData) {
public <K, V> KeyValueStoreOperations setWithMetaData(K key, V value, Map<String, String> metaData, QosParameters qosParams) {
BucketKeyPair bucketKeyPair = resolveBucketKeyPair(key, value);
String keyName = (null != qosParams ? bucketKeyPair.getKey()
.toString() + extractQosParameters(qosParams) : bucketKeyPair.getKey().toString());
RestTemplate restTemplate = getRestTemplate();
HttpHeaders headers = new HttpHeaders();
headers.set("X-Riak-ClientId", RIAK_CLIENT_ID);
@@ -268,7 +285,7 @@ public class RiakTemplate extends RestGatewaySupport implements KeyValueStoreOpe
restTemplate.put(defaultUri,
entity,
bucketKeyPair.getBucket(),
bucketKeyPair.getKey());
keyName);
if (log.isDebugEnabled()) {
log.debug(String.format("PUT object: bucket=%s, key=%s, value=%s",
bucketKeyPair.getBucket(),
@@ -278,6 +295,10 @@ public class RiakTemplate extends RestGatewaySupport implements KeyValueStoreOpe
return this;
}
public <K, V> KeyValueStoreOperations setWithMetaData(K key, V value, Map<String, String> metaData) {
return setWithMetaData(key, value, metaData, null);
}
/*----------------- Get Operations -----------------*/
public <K, T> RiakValue<T> getWithMetaData(K key, Class<T> requiredType) {
@@ -872,7 +893,6 @@ public class RiakTemplate extends RestGatewaySupport implements KeyValueStoreOpe
return meta;
}
protected <K, T> T checkCache(K key, Class<T> requiredType) {
BucketKeyPair bucketKeyPair = resolveBucketKeyPair(key, requiredType);
RiakValue<?> obj = cache.get(bucketKeyPair);
@@ -903,4 +923,27 @@ public class RiakTemplate extends RestGatewaySupport implements KeyValueStoreOpe
}
}
protected String extractQosParameters(QosParameters qosParams) {
List<String> params = new LinkedList<String>();
if (null != qosParams.getReadThreshold()) {
params.add(String.format("r=%s", qosParams.<Object>getReadThreshold()));
} else if (null != defaultQosParameters && null != defaultQosParameters.getReadThreshold()) {
params.add(String.format("r=%s", defaultQosParameters.getReadThreshold()));
}
if (null != qosParams.getWriteThreshold()) {
params.add(String.format("w=%s", qosParams.<Object>getWriteThreshold()));
} else if (null != defaultQosParameters && null != defaultQosParameters.getWriteThreshold()) {
params.add(String.format("w=%s", defaultQosParameters.getWriteThreshold()));
}
if (null != qosParams.getDurableWriteThreshold()) {
params.add(String.format("dw=%s", qosParams.<Object>getDurableWriteThreshold()));
} else if (null != defaultQosParameters && null != defaultQosParameters.getDurableWriteThreshold()) {
params.add(String.format("dw=%s", defaultQosParameters.getDurableWriteThreshold()));
}
return (params.size() > 0 ? "?" + StringUtils.collectionToDelimitedString(
params,
"&") : "");
}
}

15
spring-data-riak/src/test/groovy/org/springframework/data/riak/core/RiakTemplateSpec.groovy
View File

@@ -94,6 +94,21 @@ class RiakTemplateSpec extends Specification {
}
def "Test setting QosParameters"() {
given:
def obj = riak.get("test:test")
when:
def qos = new RiakQosParameters()
qos.durableWriteThreshold = "all"
riak.set("test:test", obj, qos)
then:
true
}
def "Test containsKey"() {
when: