From efc566036a03fa184aef9d6a189c1c9f4bfeb707 Mon Sep 17 00:00:00 2001 From: "J. Brisbin" Date: Tue, 28 Dec 2010 21:42:37 -0600 Subject: [PATCH] More fixes for NPEs when getting vclock info, fix for duplicate results in RiakBuilder. --- .../keyvalue/riak/core/AsyncRiakTemplate.java | 108 ++++++++++++------ .../data/keyvalue/riak/core/RiakTemplate.java | 9 +- .../keyvalue/riak/groovy/RiakBuilder.java | 8 +- .../keyvalue/riak/core/RiakBuilderSpec.groovy | 13 +-- 4 files changed, 91 insertions(+), 47 deletions(-) diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplate.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplate.java index 55fe982c6..f8f1f8fe5 100644 --- a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplate.java +++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplate.java @@ -89,24 +89,31 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck return defaultErrorHandler; } - public void setDefaultErrorHandler(AsyncKeyValueStoreOperation defaultErrorHandler) { + public void setDefaultErrorHandler( + AsyncKeyValueStoreOperation defaultErrorHandler) { this.defaultErrorHandler = defaultErrorHandler; } - public Future set(B bucket, K key, V value, AsyncKeyValueStoreOperation callback) { + public Future set(B bucket, K key, V value, + AsyncKeyValueStoreOperation callback) { return setWithMetaData(bucket, key, value, null, null, callback); } - public Future set(B bucket, K key, V value, QosParameters qosParams, AsyncKeyValueStoreOperation callback) { + public Future set(B bucket, K key, V value, QosParameters qosParams, + AsyncKeyValueStoreOperation callback) { return setWithMetaData(bucket, key, value, null, qosParams, callback); } - public Future setAsBytes(B bucket, K key, byte[] value, AsyncKeyValueStoreOperation callback) { + public Future setAsBytes(B bucket, K key, byte[] value, + AsyncKeyValueStoreOperation callback) { return setWithMetaData(bucket, key, value, null, null, callback); } @SuppressWarnings({"unchecked"}) - public Future setWithMetaData(B bucket, K key, V value, Map metaData, QosParameters qosParams, AsyncKeyValueStoreOperation callback) { + public Future setWithMetaData(B bucket, K key, V value, + Map metaData, + QosParameters qosParams, + AsyncKeyValueStoreOperation 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 key."); @@ -116,7 +123,13 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck KeyValueStoreMetaData origMeta = getMetaData(bucket, keyName); String vclock = null; if (null != origMeta) { - vclock = origMeta.getProperties().get(RIAK_VCLOCK).toString(); + Map mprops = origMeta.getProperties(); + if (null != mprops) { + Object o = mprops.get(RIAK_VCLOCK); + if (null != o) { + vclock = o.toString(); + } + } } HttpHeaders headers = defaultHeaders(metaData); @@ -132,18 +145,23 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck callback)); } - public Future put(B bucket, V value, AsyncKeyValueStoreOperation callback) { + public Future put(B bucket, V value, + AsyncKeyValueStoreOperation callback) { return put(bucket, value, null, null, callback); } - public Future put(B bucket, V value, Map metaData, AsyncKeyValueStoreOperation callback) { + public Future put(B bucket, V value, Map metaData, + AsyncKeyValueStoreOperation callback) { return put(bucket, value, metaData, null, callback); } @SuppressWarnings({"unchecked"}) - public Future put(B bucket, V value, Map metaData, QosParameters qosParams, AsyncKeyValueStoreOperation callback) { + public Future put(B bucket, V value, Map metaData, + QosParameters qosParams, + AsyncKeyValueStoreOperation callback) { Assert.notNull(bucket, "Bucket cannot be null"); - String bucketName = (null != qosParams ? bucket.toString() + extractQosParameters(qosParams) : bucket + String bucketName = (null != qosParams ? bucket.toString() + extractQosParameters( + qosParams) : bucket .toString()); HttpHeaders headers = defaultHeaders(metaData); @@ -153,7 +171,8 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck return (Future) workerPool.submit(new AsyncPut(bucketName, entity, callback)); } - public Future get(B bucket, K key, AsyncKeyValueStoreOperation callback) { + public Future get(B bucket, K key, + AsyncKeyValueStoreOperation callback) { return getWithMetaData(bucket, key, null, callback); } @@ -174,11 +193,13 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck } @SuppressWarnings({"unchecked"}) - public Future getBucketSchema(B bucket, QosParameters qosParams, final AsyncKeyValueStoreOperation, R> callback) { + public Future getBucketSchema(B bucket, QosParameters qosParams, + final AsyncKeyValueStoreOperation, R> callback) { Assert.notNull(bucket, "Bucket cannot be null"); Assert.notNull(callback, "Callback cannot be null"); - String bucketName = (null != qosParams ? bucket.toString() + extractQosParameters(qosParams) : bucket + String bucketName = (null != qosParams ? bucket.toString() + extractQosParameters( + qosParams) : bucket .toString()); return workerPool.submit(new AsyncGet(bucketName, @@ -197,7 +218,8 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck } @SuppressWarnings({"unchecked"}) - public Future getWithMetaData(B bucket, K key, Class requiredType, AsyncKeyValueStoreOperation callback) { + public Future getWithMetaData(B bucket, K key, Class requiredType, + AsyncKeyValueStoreOperation 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."); @@ -212,15 +234,18 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck callback)); } - public Future getAsBytes(B bucket, K key, AsyncKeyValueStoreOperation callback) { + public Future getAsBytes(B bucket, K key, + AsyncKeyValueStoreOperation callback) { return getWithMetaData(bucket, key, byte[].class, callback); } - public Future getAsType(B bucket, K key, Class requiredType, AsyncKeyValueStoreOperation callback) { + public Future getAsType(B bucket, K key, Class requiredType, + AsyncKeyValueStoreOperation callback) { return getWithMetaData(bucket, key, requiredType, callback); } - public Future getAndSet(final B bucket, final K key, final V value, final AsyncKeyValueStoreOperation callback) { + public Future getAndSet(final B bucket, final K key, final V value, + final AsyncKeyValueStoreOperation callback) { final List> futures = new ArrayList>(); try { getWithMetaData(bucket, key, null, new AsyncKeyValueStoreOperation() { @@ -242,11 +267,14 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck return futures.size() > 0 ? futures.get(0) : null; } - public Future getAndSetAsBytes(B bucket, K key, byte[] value, AsyncKeyValueStoreOperation callback) { + public Future getAndSetAsBytes(B bucket, K key, byte[] value, + AsyncKeyValueStoreOperation callback) { return getAndSet(bucket, key, value, callback); } - public Future getAndSetAsType(final B bucket, final K key, final V value, final Class requiredType, final AsyncKeyValueStoreOperation callback) { + public Future getAndSetAsType(final B bucket, final K key, final V value, + final Class requiredType, + final AsyncKeyValueStoreOperation callback) { final List> futures = new ArrayList>(); getWithMetaData(bucket, key, requiredType, new AsyncKeyValueStoreOperation() { @SuppressWarnings({"unchecked"}) @@ -268,7 +296,8 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck return futures.size() > 0 ? futures.get(0) : null; } - public Future setIfKeyNonExistent(final B bucket, final K key, final V value, final AsyncKeyValueStoreOperation callback) { + public Future setIfKeyNonExistent(final B bucket, final K key, final V value, + final AsyncKeyValueStoreOperation callback) { return containsKey(bucket, key, new AsyncKeyValueStoreOperation() { public Object completed(KeyValueStoreMetaData meta, Boolean result) { if (!result) { @@ -284,7 +313,9 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck }); } - public Future setIfKeyNonExistentAsBytes(final B bucket, final K key, final byte[] value, final AsyncKeyValueStoreOperation callback) { + public Future setIfKeyNonExistentAsBytes(final B bucket, final K key, + final byte[] value, + final AsyncKeyValueStoreOperation callback) { return containsKey(bucket, key, new AsyncKeyValueStoreOperation() { public Object completed(KeyValueStoreMetaData meta, Boolean result) { if (!result) { @@ -301,7 +332,8 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck } @SuppressWarnings({"unchecked"}) - public Future containsKey(B bucket, K key, final AsyncKeyValueStoreOperation callback) { + public Future containsKey(B bucket, K key, + final AsyncKeyValueStoreOperation 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(), @@ -318,24 +350,29 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck } @SuppressWarnings({"unchecked"}) - public Future delete(B bucket, K key, AsyncKeyValueStoreOperation callback) { + public Future delete(B bucket, K key, + AsyncKeyValueStoreOperation 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 Future setAsBytes(B bucket, K key, byte[] value, QosParameters qosParams, AsyncKeyValueStoreOperation callback) { + public Future setAsBytes(B bucket, K key, byte[] value, QosParameters qosParams, + AsyncKeyValueStoreOperation callback) { return setWithMetaData(bucket, key, value, null, qosParams, callback); } - public Future setWithMetaData(B bucket, K key, V value, Map metaData, AsyncKeyValueStoreOperation callback) { + public Future setWithMetaData(B bucket, K key, V value, + Map metaData, + AsyncKeyValueStoreOperation callback) { return setWithMetaData(bucket, key, value, metaData, null, callback); } /* ---------------- Map/Reduce ---------------- */ @SuppressWarnings({"unchecked"}) - public Future execute(MapReduceJob job, AsyncKeyValueStoreOperation, R> callback) { + public Future execute(MapReduceJob job, + AsyncKeyValueStoreOperation, R> callback) { HttpHeaders headers = defaultHeaders(null); headers.setContentType(MediaType.APPLICATION_JSON); HttpEntity json = new HttpEntity(job.toJson(), headers); @@ -343,13 +380,15 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck } /* ---------------- Runnable helpers ---------------- */ + protected class AsyncPut implements Callable { private String bucket; private HttpEntity entity = null; private AsyncKeyValueStoreOperation callback = null; - public AsyncPut(String bucket, HttpEntity entity, AsyncKeyValueStoreOperation callback) { + public AsyncPut(String bucket, HttpEntity entity, + AsyncKeyValueStoreOperation callback) { this.bucket = bucket; this.entity = entity; this.callback = callback; @@ -388,7 +427,8 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck private HttpEntity entity = null; private AsyncKeyValueStoreOperation callback = null; - public AsyncPost(String bucket, String key, HttpEntity entity, AsyncKeyValueStoreOperation callback) { + public AsyncPost(String bucket, String key, HttpEntity entity, + AsyncKeyValueStoreOperation callback) { this.bucket = bucket; this.key = key; this.entity = entity; @@ -433,7 +473,8 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck private HttpEntity entity = null; private AsyncKeyValueStoreOperation, R> callback = null; - public AsyncMapReduce(HttpEntity entity, AsyncKeyValueStoreOperation, R> callback) { + public AsyncMapReduce(HttpEntity entity, + AsyncKeyValueStoreOperation, R> callback) { this.entity = entity; this.callback = callback; } @@ -471,7 +512,8 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck private Class requiredType; private AsyncKeyValueStoreOperation callback = null; - public AsyncGet(String bucket, String key, Class requiredType, AsyncKeyValueStoreOperation callback) { + public AsyncGet(String bucket, String key, Class requiredType, + AsyncKeyValueStoreOperation callback) { this.bucket = bucket; this.key = key; this.requiredType = requiredType; @@ -520,7 +562,8 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck private String key; private AsyncKeyValueStoreOperation callback = null; - public AsyncHead(String bucket, String key, AsyncKeyValueStoreOperation callback) { + public AsyncHead(String bucket, String key, + AsyncKeyValueStoreOperation callback) { this.bucket = bucket; this.key = key; this.callback = callback; @@ -552,7 +595,8 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck private String key; private AsyncKeyValueStoreOperation callback = null; - public AsyncDelete(String bucket, String key, AsyncKeyValueStoreOperation callback) { + public AsyncDelete(String bucket, String key, + AsyncKeyValueStoreOperation callback) { this.bucket = bucket; this.key = key; this.callback = callback; diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/RiakTemplate.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/RiakTemplate.java index d00abd718..7164ef759 100644 --- a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/RiakTemplate.java +++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/RiakTemplate.java @@ -129,9 +129,12 @@ public class RiakTemplate extends AbstractRiakTemplate implements BucketKeyValue KeyValueStoreMetaData origMeta = getMetaData(bucket, keyName); String vclock = null; if (null != origMeta) { - Object o = origMeta.getProperties().get(RIAK_VCLOCK); - if (null != o) { - vclock = o.toString(); + Map mprops = origMeta.getProperties(); + if (null != mprops) { + Object o = mprops.get(RIAK_VCLOCK); + if (null != o) { + vclock = o.toString(); + } } } RestTemplate restTemplate = getRestTemplate(); diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakBuilder.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakBuilder.java index 5dced9278..5dcf2181a 100644 --- a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakBuilder.java +++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakBuilder.java @@ -81,7 +81,8 @@ public class RiakBuilder extends BuilderSupport { this.riak = riak; } - public RiakBuilder(Closure nameMappingClosure, BuilderSupport proxyBuilder, AsyncRiakTemplate riak) { + public RiakBuilder(Closure nameMappingClosure, BuilderSupport proxyBuilder, + AsyncRiakTemplate riak) { super(nameMappingClosure, proxyBuilder); this.riak = riak; } @@ -276,7 +277,8 @@ public class RiakBuilder extends BuilderSupport { @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. + return super.invokeMethod( + methodName); //To change body of overridden methods use File | Settings | File Templates. } @SuppressWarnings({"unchecked"}) @@ -363,7 +365,7 @@ public class RiakBuilder extends BuilderSupport { RiakOperation op = (RiakOperation) node; try { Object o = op.call(); - if (null != o) { + if (null != o && !o.equals(results)) { results.add(o); } return o; diff --git a/spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/RiakBuilderSpec.groovy b/spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/RiakBuilderSpec.groovy index ae6e43393..08e882c53 100644 --- a/spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/RiakBuilderSpec.groovy +++ b/spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/RiakBuilderSpec.groovy @@ -266,22 +266,17 @@ class RiakBuilderSpec extends Specification { def riak = new RiakBuilder(riakTemplate) when: - def deleted = riak { + riak { "test" { foreach { - completed { v, meta -> - delete(bucket: meta.bucket, key: meta.key) { - completed { deleted = true } - failed { deleted = false } - } - } - failed { it.printStackTrace() } + completed { v, meta -> delete(key: meta.key) } + failed { deleted = false } } } } then: - deleted + !riak.results.find { !it } }