From da7a0d43c294a8a50f35fc568f654a171739f7ae Mon Sep 17 00:00:00 2001 From: "J. Brisbin" Date: Mon, 20 Dec 2010 15:00:33 -0600 Subject: [PATCH] Added asynchronous version of RiakTemplate, Groovy DSL for data access. --- .../riak/core/AbstractAsyncOperation.java | 52 --- .../riak/core/AbstractRiakTemplate.java | 24 +- .../AsyncBucketKeyValueStoreOperations.java | 166 +++++++ .../core/AsyncKeyValueStoreOperation.java | 29 ++ .../keyvalue/riak/core/AsyncRiakTemplate.java | 417 ++++++++++++++++++ .../core/BucketKeyValueStoreOperations.java | 1 + .../keyvalue/riak/groovy/RiakBuilder.java | 184 ++++++++ .../keyvalue/riak/groovy/RiakOperation.java | 238 ++++++++++ .../keyvalue/riak/mapreduce/MapReduceJob.java | 11 +- .../riak/mapreduce/RiakMapReduceJob.java | 4 + .../riak/core/AsyncRiakTemplateSpec.groovy | 87 ++++ .../keyvalue/riak/core/RiakBuilderSpec.groovy | 140 ++++++ .../data/AsyncRiakTemplateTests.xml | 34 ++ 13 files changed, 1329 insertions(+), 58 deletions(-) delete mode 100644 spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AbstractAsyncOperation.java create mode 100644 spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncBucketKeyValueStoreOperations.java create mode 100644 spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncKeyValueStoreOperation.java create mode 100644 spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplate.java create mode 100644 spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakBuilder.java create mode 100644 spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakOperation.java create mode 100644 spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplateSpec.groovy create mode 100644 spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/RiakBuilderSpec.groovy create mode 100644 spring-data-riak/src/test/resources/org/springframework/data/AsyncRiakTemplateTests.xml diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AbstractAsyncOperation.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AbstractAsyncOperation.java deleted file mode 100644 index d2a699eeb..000000000 --- a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AbstractAsyncOperation.java +++ /dev/null @@ -1,52 +0,0 @@ -/* - * Copyright (c) 2010 by J. Brisbin - * 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; - -import org.springframework.beans.factory.InitializingBean; -import org.springframework.util.Assert; - -import java.util.concurrent.Callable; - -/** - * @author J. Brisbin - */ -public abstract class AbstractAsyncOperation implements Callable, InitializingBean { - - protected RiakTemplate riakTemplate; - - protected AbstractAsyncOperation() { - } - - protected AbstractAsyncOperation(RiakTemplate riakTemplate) { - this.riakTemplate = riakTemplate; - } - - public RiakTemplate getRiakTemplate() { - return riakTemplate; - } - - public void setRiakTemplate(RiakTemplate riakTemplate) { - this.riakTemplate = riakTemplate; - } - - public void afterPropertiesSet() throws Exception { - Assert.notNull(riakTemplate, "Must provide a configured RiakTemplate for this operation."); - } - -} 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 aa7156779..f966ff826 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 @@ -59,10 +59,8 @@ import java.util.regex.Pattern; */ public abstract class AbstractRiakTemplate extends RestGatewaySupport implements InitializingBean { - /** - * Client ID used by Riak to correlate updates. - */ - protected static final String RIAK_CLIENT_ID = "org.springframework.data.keyvalue.riak.core.RiakTemplate/1.0"; + protected static final String RIAK_META_CLASSNAME = "X-Riak-Meta-ClassName"; + /** * Regex used to extract host, port, and prefix from the given URI. */ @@ -81,6 +79,12 @@ public abstract class AbstractRiakTemplate extends RestGatewaySupport implements "EEE, d MMM yyyy HH:mm:ss z"); protected final Logger log = LoggerFactory.getLogger(getClass()); + + /** + * Client ID used by Riak to correlate updates. + */ + protected final String RIAK_CLIENT_ID = getClass().getName() + "/1.0"; + /** * For converting objects to/from other kinds of objects. */ @@ -380,4 +384,16 @@ public abstract class AbstractRiakTemplate extends RestGatewaySupport implements "&") : ""); } + protected HttpHeaders defaultHeaders(Map metadata) { + HttpHeaders headers = new HttpHeaders(); + headers.set("X-Riak-ClientId", RIAK_CLIENT_ID); + if (null != metadata) { + for (Map.Entry entry : metadata.entrySet()) { + Object o = entry.getValue(); + headers.set(entry.getKey(), (null != o ? o.toString() : null)); + } + } + return headers; + } + } diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncBucketKeyValueStoreOperations.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncBucketKeyValueStoreOperations.java new file mode 100644 index 000000000..b006fea00 --- /dev/null +++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncBucketKeyValueStoreOperations.java @@ -0,0 +1,166 @@ +/* + * Copyright (c) 2010 by J. Brisbin + * 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; + +import java.util.Map; +import java.util.concurrent.Future; + +/** + * An asynchronous version of {@link BucketKeyValueStoreOperations}. + * + * @author J. Brisbin + */ +public interface AsyncBucketKeyValueStoreOperations { + + /** + * Put an object in Riak at a specific bucket and key and invoke callback with the value + * pulled back out of Riak after the update, which contains full headers and metadata. + * + * @param bucket + * @param key + * @param value + * @param callback Called with the update value pulled from Riak + */ + Future set(B bucket, K key, V value, AsyncKeyValueStoreOperation callback); + + /** + * @param bucket + * @param key + * @param value + * @param qosParams + * @return + */ + Future set(B bucket, K key, V value, QosParameters qosParams, AsyncKeyValueStoreOperation callback); + + /** + * @param bucket + * @param key + * @param value + * @return + */ + Future setAsBytes(B bucket, K key, byte[] value, AsyncKeyValueStoreOperation callback); + + /** + * @param bucket + * @param key + * @param value + * @param qosParams + * @return + */ + Future setAsBytes(B bucket, K key, byte[] value, QosParameters qosParams, AsyncKeyValueStoreOperation callback); + + /** + * @param bucket + * @param key + * @param value + * @param metaData + * @return + */ + Future setWithMetaData(B bucket, K key, V value, Map metaData, AsyncKeyValueStoreOperation callback); + + /** + * @param bucket + * @param key + * @param value + * @param metaData + * @param qosParams + * @return + */ + Future setWithMetaData(B bucket, K key, V value, Map metaData, QosParameters qosParams, AsyncKeyValueStoreOperation callback); + + /** + * @param bucket + * @param key + * @return + */ + Future get(B bucket, K key, AsyncKeyValueStoreOperation callback); + + /** + * @param bucket + * @param key + * @return + */ + Future getAsBytes(B bucket, K key, AsyncKeyValueStoreOperation callback); + + /** + * @param bucket + * @param key + * @param requiredType + * @return + */ + Future getAsType(B bucket, K key, Class requiredType, AsyncKeyValueStoreOperation callback); + + /** + * @param bucket + * @param key + * @param value + * @return + */ + Future getAndSet(B bucket, K key, V value, AsyncKeyValueStoreOperation callback); + + /** + * @param bucket + * @param key + * @param value + * @return + */ + Future getAndSetAsBytes(B bucket, K key, byte[] value, AsyncKeyValueStoreOperation callback); + + /** + * @param bucket + * @param key + * @param value + * @param requiredType + * @return + */ + Future getAndSetAsType(B bucket, K key, V value, Class requiredType, AsyncKeyValueStoreOperation callback); + + /** + * @param bucket + * @param key + * @param value + * @return + */ + Future setIfKeyNonExistent(B bucket, K key, V value, AsyncKeyValueStoreOperation callback); + + /** + * @param bucket + * @param key + * @param value + * @return + */ + Future setIfKeyNonExistentAsBytes(B bucket, K key, byte[] value, AsyncKeyValueStoreOperation callback); + + /** + * @param bucket + * @param key + * @return + */ + Future containsKey(B bucket, K key, AsyncKeyValueStoreOperation callback); + + /** + * Delete a specific entry from this data store. + * + * @param bucket + * @param key + * @return + */ + Future delete(B bucket, K key, AsyncKeyValueStoreOperation callback); + +} diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncKeyValueStoreOperation.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncKeyValueStoreOperation.java new file mode 100644 index 000000000..cdd6e41fc --- /dev/null +++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncKeyValueStoreOperation.java @@ -0,0 +1,29 @@ +/* + * Copyright (c) 2010 by J. Brisbin + * 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; + +/** + * @author J. Brisbin + */ +public interface AsyncKeyValueStoreOperation { + + void completed(KeyValueStoreMetaData meta, V result); + + void failed(Throwable error); +} 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 new file mode 100644 index 000000000..09270c991 --- /dev/null +++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplate.java @@ -0,0 +1,417 @@ +/* + * Copyright (c) 2010 by J. Brisbin + * 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; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.data.keyvalue.riak.DataStoreOperationException; +import org.springframework.http.HttpEntity; +import org.springframework.http.HttpHeaders; +import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; +import org.springframework.http.client.ClientHttpRequestFactory; +import org.springframework.util.Assert; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +/** + * @author J. Brisbin + */ +public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBucketKeyValueStoreOperations { + + protected final Logger log = LoggerFactory.getLogger(getClass()); + + protected ExecutorService workerPool = Executors.newCachedThreadPool(); + protected AsyncKeyValueStoreOperation defaultErrorHandler = new LoggingErrorHandler(); + + public AsyncRiakTemplate() { + super(); + } + + public AsyncRiakTemplate(ClientHttpRequestFactory requestFactory) { + super(requestFactory); + } + + public ExecutorService getWorkerPool() { + return workerPool; + } + + public void setWorkerPool(ExecutorService workerPool) { + this.workerPool = workerPool; + } + + public AsyncKeyValueStoreOperation getDefaultErrorHandler() { + return defaultErrorHandler; + } + + public void setDefaultErrorHandler(AsyncKeyValueStoreOperation defaultErrorHandler) { + this.defaultErrorHandler = defaultErrorHandler; + } + + 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) { + return setWithMetaData(bucket, key, value, null, qosParams, 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) { + 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."); + String keyName = (null != qosParams ? key.toString() + extractQosParameters(qosParams) : key + .toString()); + HttpHeaders headers = defaultHeaders(metaData); + headers.setContentType(extractMediaType(value)); + headers.set(RIAK_META_CLASSNAME, value.getClass().getName()); + HttpEntity entity = new HttpEntity(value, headers); + return (Future) workerPool.submit(new AsyncPost(bucketName, + keyName, + entity, + callback)); + } + + public Future get(B bucket, K key, AsyncKeyValueStoreOperation callback) { + return getWithMetaData(bucket, key, null, callback); + } + + @SuppressWarnings({"unchecked"}) + 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 key."); + if (null == requiredType) { + try { + requiredType = (Class) getType(bucketName, key.toString()); + } catch (ClassNotFoundException e) { + throw new DataStoreOperationException(e.getMessage(), e); + } + } + return workerPool.submit(new AsyncGet(bucketName, + key.toString(), + requiredType, + 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) { + return getWithMetaData(bucket, key, requiredType, 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() { + @SuppressWarnings({"unchecked"}) + public void completed(KeyValueStoreMetaData meta, Object result) { + futures.add(setWithMetaData(bucket, key, value, null, null, null)); + callback.completed(meta, (V) result); + } + + public void failed(Throwable error) { + callback.failed(error); + } + }).get(); + } catch (InterruptedException e) { + log.error(e.getMessage(), e); + } catch (ExecutionException e) { + log.error(e.getMessage(), e); + } + return futures.size() > 0 ? futures.get(0) : null; + } + + 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) { + final List> futures = new ArrayList>(); + getWithMetaData(bucket, key, requiredType, new AsyncKeyValueStoreOperation() { + @SuppressWarnings({"unchecked"}) + public void completed(KeyValueStoreMetaData meta, T result) { + futures.add(setWithMetaData(bucket, key, value, null, null, null)); + callback.completed(meta, (V) result); + } + + public void failed(Throwable error) { + callback.failed(error); + } + }); + return futures.size() > 0 ? futures.get(0) : null; + } + + public Future setIfKeyNonExistent(final B bucket, final K key, final V value, final AsyncKeyValueStoreOperation callback) { + return containsKey(bucket, key, new AsyncKeyValueStoreOperation() { + public void completed(KeyValueStoreMetaData meta, Boolean result) { + if (!result) { + setWithMetaData(bucket, key, value, null, null, callback); + } + } + + public void failed(Throwable error) { + callback.failed(error); + } + }); + } + + public Future setIfKeyNonExistentAsBytes(final B bucket, final K key, final byte[] value, final AsyncKeyValueStoreOperation callback) { + return containsKey(bucket, key, new AsyncKeyValueStoreOperation() { + public void completed(KeyValueStoreMetaData meta, Boolean result) { + if (!result) { + setWithMetaData(bucket, key, value, null, null, callback); + } + } + + public void failed(Throwable error) { + callback.failed(error); + } + }); + } + + 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(), + key.toString(), + new AsyncKeyValueStoreOperation() { + public void completed(KeyValueStoreMetaData meta, HttpHeaders result) { + callback.completed(null, (null != result)); + } + + public void failed(Throwable error) { + callback.failed(error); + } + })); + } + + 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) { + return setWithMetaData(bucket, key, value, null, qosParams, callback); + } + + public Future setWithMetaData(B bucket, K key, V value, Map metaData, AsyncKeyValueStoreOperation callback) { + return setWithMetaData(bucket, key, value, metaData, null, callback); + } + + protected Class getType(String bucket, String key) throws ClassNotFoundException { + HttpHeaders headers = getRestTemplate().headForHeaders(defaultUri, bucket, key); + Class clazz = null; + if (null != headers) { + String s = headers.getFirst(RIAK_META_CLASSNAME); + if (null != s) { + try { + clazz = Class.forName(s); + } catch (ClassNotFoundException ignored) { + if (headers.getContentType().equals(MediaType.APPLICATION_JSON)) { + clazz = Map.class; + } else if (headers.getContentType().equals(MediaType.TEXT_PLAIN)) { + clazz = String.class; + } else { + // handle as bytes + log.error("Need to handle bytes!"); + } + } + } + } + if (null == clazz) { + clazz = byte[].class; + } + return clazz; + } + + protected class AsyncPost implements Runnable { + + private String bucket; + private String key; + private HttpEntity entity = null; + private AsyncKeyValueStoreOperation callback = null; + + public AsyncPost(String bucket, String key, HttpEntity entity, AsyncKeyValueStoreOperation callback) { + this.bucket = bucket; + this.key = key; + this.entity = entity; + this.callback = callback; + } + + @SuppressWarnings({"unchecked"}) + public void run() { + try { + HttpEntity result = getRestTemplate().postForEntity(defaultUri, + entity, + (entity.getBody() instanceof byte[] ? byte[].class : entity.getBody().getClass()), + bucket, + key + "?returnbody=true"); + if (log.isDebugEnabled()) { + log.debug(String.format("PUT object: bucket=%s, key=%s, value=%s", + bucket, + key, + entity)); + } + if (null != callback) { + callback.completed(extractMetaData(result.getHeaders()), (V) result.getBody()); + } + } catch (Throwable t) { + DataStoreOperationException dsoe = new DataStoreOperationException(t.getMessage(), t); + if (null != callback) { + callback.failed(dsoe); + } else { + defaultErrorHandler.failed(dsoe); + } + } + } + + } + + protected class AsyncGet implements Runnable { + + private String bucket; + private String key; + private Class requiredType; + private AsyncKeyValueStoreOperation callback = null; + + public AsyncGet(String bucket, String key, Class requiredType, AsyncKeyValueStoreOperation callback) { + this.bucket = bucket; + this.key = key; + this.requiredType = requiredType; + this.callback = callback; + } + + public void run() { + try { + ResponseEntity result = getRestTemplate().getForEntity(defaultUri, + requiredType, + bucket, + key); + if (result.hasBody()) { + RiakMetaData meta = extractMetaData(result.getHeaders()); + RiakValue val = new RiakValue(result.getBody(), meta); + if (useCache) { + cache.put(new SimpleBucketKeyPair(bucket, key), val); + } + if (null != callback) { + callback.completed(meta, val.get()); + } + if (log.isDebugEnabled()) { + log.debug(String.format("GET object: bucket=%s, key=%s, type=%s", + bucket, + key, + requiredType.getName())); + } + } + } catch (Throwable t) { + DataStoreOperationException dsoe = new DataStoreOperationException(t.getMessage(), t); + if (null != callback) { + callback.failed(dsoe); + } else { + defaultErrorHandler.failed(dsoe); + } + } + } + } + + protected class AsyncHead implements Runnable { + + private String bucket; + private String key; + private AsyncKeyValueStoreOperation callback = null; + + public AsyncHead(String bucket, String key, AsyncKeyValueStoreOperation callback) { + this.bucket = bucket; + this.key = key; + this.callback = callback; + } + + public void run() { + try { + HttpHeaders headers = getRestTemplate().headForHeaders(defaultUri, bucket, key); + if (null != headers) { + if (null != callback) { + callback.completed(null, headers); + } + } + } catch (Throwable t) { + DataStoreOperationException dsoe = new DataStoreOperationException(t.getMessage(), t); + if (null != callback) { + callback.failed(dsoe); + } else { + defaultErrorHandler.failed(dsoe); + } + } + } + } + + protected class AsyncDelete implements Runnable { + + private String bucket; + private String key; + private AsyncKeyValueStoreOperation callback = null; + + public AsyncDelete(String bucket, String key, AsyncKeyValueStoreOperation callback) { + this.bucket = bucket; + this.key = key; + this.callback = callback; + } + + public void run() { + try { + getRestTemplate().delete(defaultUri, bucket, key); + if (null != callback) { + callback.completed(null, true); + } + } catch (Throwable t) { + DataStoreOperationException dsoe = new DataStoreOperationException(t.getMessage(), t); + if (null != callback) { + callback.failed(dsoe); + } else { + defaultErrorHandler.failed(dsoe); + } + } + } + } + + protected class LoggingErrorHandler implements AsyncKeyValueStoreOperation { + public void completed(KeyValueStoreMetaData meta, Throwable result) { + } + + public void failed(Throwable error) { + log.error(error.getMessage(), error); + } + } + +} diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/BucketKeyValueStoreOperations.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/BucketKeyValueStoreOperations.java index 13398d08b..23f370518 100644 --- a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/BucketKeyValueStoreOperations.java +++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/BucketKeyValueStoreOperations.java @@ -201,4 +201,5 @@ public interface BucketKeyValueStoreOperations { * @return */ BucketKeyValueStoreOperations setWithMetaData(B bucket, K key, V value, Map metaData); + } 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 new file mode 100644 index 000000000..7652d3db7 --- /dev/null +++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakBuilder.java @@ -0,0 +1,184 @@ +/* + * Copyright (c) 2010 by J. Brisbin + * 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.groovy; + +import groovy.lang.Closure; +import groovy.util.BuilderSupport; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.data.keyvalue.riak.core.AsyncRiakTemplate; +import org.springframework.data.keyvalue.riak.core.RiakQosParameters; + +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +/** + * @author J. Brisbin + */ +public class RiakBuilder extends BuilderSupport { + + protected final Logger log = LoggerFactory.getLogger(getClass()); + protected AsyncRiakTemplate riak; + protected ExecutorService workerPool = Executors.newCachedThreadPool(); + + public RiakBuilder(AsyncRiakTemplate riak) { + this.riak = riak; + } + + public RiakBuilder(AsyncRiakTemplate riak, ExecutorService workerPool) { + this.riak = riak; + this.workerPool = workerPool; + } + + public RiakBuilder(BuilderSupport proxyBuilder, AsyncRiakTemplate riak) { + super(proxyBuilder); + this.riak = riak; + } + + public RiakBuilder(Closure nameMappingClosure, BuilderSupport proxyBuilder, AsyncRiakTemplate riak) { + super(nameMappingClosure, proxyBuilder); + this.riak = riak; + } + + public ExecutorService getWorkerPool() { + return workerPool; + } + + public void setWorkerPool(ExecutorService workerPool) { + this.workerPool = workerPool; + } + + @Override + protected void setParent(Object parent, Object child) { + log.debug("setParent/2 " + parent + " " + child); + } + + @Override + protected Object createNode(Object name) { + log.debug("createNode/1 " + name); + return this; + } + + @Override + protected Object createNode(Object name, Object value) { + log.debug("createNode/2 " + name + " " + value); + return null; //To change body of implemented methods use File | Settings | File Templates. + } + + @SuppressWarnings({"unchecked"}) + @Override + protected Object createNode(Object name, Map attributes) { + log.debug("createNode/2 (Map) " + name + " " + attributes); + RiakOperation.Type type = RiakOperation.Type.valueOf(name.toString().toUpperCase()); + if (null != type) { + RiakOperation op = new RiakOperation(riak, type); + Object o = attributes.get("bucket"); + op.setBucket((null != o ? o.toString() : null)); + o = attributes.get("key"); + op.setKey((null != o ? o.toString() : null)); + o = attributes.get("value"); + op.setValue(o); + o = attributes.get("qos"); + if (null != o) { + RiakQosParameters qos = new RiakQosParameters(); + Map qosParams = (Map) o; + if (qosParams.containsKey("dw")) { + qos.setDurableWriteThreshold(qosParams.get("dw")); + } + if (qosParams.containsKey("w")) { + qos.setWriteThreshold(qosParams.get("w")); + } + if (qosParams.containsKey("r")) { + qos.setReadThreshold(qosParams.get("r")); + } + op.setQosParameters(qos); + } + + o = attributes.get("wait"); + if (null != o && o instanceof Long) { + op.setTimeout((Long) o); + } + return op; + } + return null; + } + + @Override + protected Object createNode(Object name, Map attributes, Object value) { + log.debug("createNode/3"); + return null; //To change body of implemented methods use File | Settings | File Templates. + } + + @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. + } + + @SuppressWarnings({"unchecked"}) + @Override + public Object invokeMethod(String methodName, Object arg) { + if (log.isDebugEnabled()) { + log.debug("invokeMethod: " + methodName + " " + arg); + } + if ("completed".equals(methodName) || "failed".equals(methodName)) { + RiakOperation op = (RiakOperation) getCurrent(); + Object[] args = (Object[]) arg; + Map params; + Closure handler = null; + Closure guard = null; + for (Object o : args) { + if (o instanceof Map) { + params = (Map) o; + if (params.containsKey("when")) { + guard = (Closure) params.get("when"); + } + } else if (o instanceof Closure) { + handler = (Closure) o; + } + } + op.addHandler(methodName, handler, guard); + return op; + } + return super.invokeMethod(methodName, arg); + } + + @SuppressWarnings({"unchecked"}) + @Override + protected void nodeCompleted(Object parent, Object node) { + log.debug("nodeCompleted: " + parent + " " + node); + if (null == parent && node instanceof RiakOperation) { + RiakOperation op = (RiakOperation) node; + try { + op.call(); + } catch (Exception e) { + log.error(e.getMessage(), e); + } + } else { + super.nodeCompleted(parent, node); + } + } + + @Override + protected Object postNodeCompletion(Object parent, Object node) { + log.debug("postNodeCompletion: " + parent + " " + node); + return super.postNodeCompletion(parent, node); + } +} diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakOperation.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakOperation.java new file mode 100644 index 000000000..86fa06dc4 --- /dev/null +++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakOperation.java @@ -0,0 +1,238 @@ +/* + * Copyright (c) 2010 by J. Brisbin + * 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.groovy; + +import groovy.lang.Closure; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.data.keyvalue.riak.core.AsyncKeyValueStoreOperation; +import org.springframework.data.keyvalue.riak.core.AsyncRiakTemplate; +import org.springframework.data.keyvalue.riak.core.KeyValueStoreMetaData; +import org.springframework.data.keyvalue.riak.core.QosParameters; + +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; + +/** + * @author J. Brisbin + */ +public class RiakOperation implements Callable { + + static enum Type { + SET, SETASBYTES, PUT, GET, GETASBYTES, CONTAINSKEY, DELETE + } + + static String COMPLETED = "completed"; + static String FAILED = "failed"; + + protected final Logger log = LoggerFactory.getLogger(getClass()); + + protected AsyncRiakTemplate riak; + protected Type type; + protected String bucket; + protected String key; + protected T value; + protected long timeout = -1L; + protected QosParameters qosParameters; + protected Map> callbacks = new LinkedHashMap>(); + protected ClosureInvokingCallback callbackInvoker = new ClosureInvokingCallback(); + + public RiakOperation(AsyncRiakTemplate riak, Type type) { + this.riak = riak; + this.type = type; + } + + public Type getType() { + return type; + } + + public Map> getCallbacks() { + return callbacks; + } + + public QosParameters getQosParameters() { + return qosParameters; + } + + public void setQosParameters(QosParameters qosParameters) { + this.qosParameters = qosParameters; + } + + public String getBucket() { + return bucket; + } + + public void setBucket(String bucket) { + this.bucket = bucket; + } + + public String getKey() { + return key; + } + + public void setKey(String key) { + this.key = key; + } + + public T getValue() { + return value; + } + + public void setValue(T value) { + this.value = value; + } + + public long getTimeout() { + return timeout; + } + + public void setTimeout(long timeout) { + this.timeout = timeout; + } + + public void addHandler(String type, Closure handler, Closure guard) { + List guardedClosures = callbacks.get(type); + if (null == guardedClosures) { + guardedClosures = new ArrayList(); + callbacks.put(type, guardedClosures); + } + guardedClosures.add(new GuardedClosure(handler, guard)); + } + + @SuppressWarnings({"unchecked"}) + public T call() throws Exception { + Future f = null; + switch (type) { + case GET: + f = riak.get(bucket, key, callbackInvoker); + break; + case GETASBYTES: + f = riak.getAsBytes(bucket, key, callbackInvoker); + break; + case PUT: + throw new IllegalStateException("PUT not yet implemented in AsyncRiakTemplate"); + case SET: + f = riak.set(bucket, key, value, callbackInvoker); + break; + case SETASBYTES: + if (value instanceof byte[]) { + f = riak.setAsBytes(bucket, key, (byte[]) value, callbackInvoker); + } else { + log.error("Need to convert obj to byte array first!"); + } + break; + case CONTAINSKEY: + f = riak.containsKey(bucket, key, callbackInvoker); + break; + case DELETE: + f = riak.delete(bucket, key, callbackInvoker); + break; + } + return null != f && timeout > 0 ? (T) f.get(timeout, TimeUnit.MILLISECONDS) : null; + } + + class GuardedClosure { + + private Closure delegate; + private Closure guard; + + GuardedClosure(Closure delegate, Closure guard) { + this.delegate = delegate; + this.guard = guard; + } + + public Closure getDelegate() { + return delegate; + } + + public Closure getGuard() { + return guard; + } + + } + + class ClosureInvokingCallback implements AsyncKeyValueStoreOperation { + + public void completed(KeyValueStoreMetaData meta, Object result) { + for (GuardedClosure cl : callbacks.get(COMPLETED)) { + boolean execute = true; + + Closure guardExpr = cl.getGuard(); + if (null != guardExpr) { + int noOfParams = guardExpr.getParameterTypes().length; + Object guardResult; + if (noOfParams == 2) { + guardResult = guardExpr.call(new Object[]{result, meta}); + } else { + guardResult = guardExpr.call(result); + } + if (null != guardResult) { + if (guardResult instanceof Boolean) { + execute = (Boolean) guardResult; + } else { + execute = true; + } + } + } + + if (execute) { + Closure callback = cl.getDelegate(); + if (callback.getParameterTypes().length == 2) { + // Pass value and metadata + callback.call(new Object[]{result, meta}); + } else { + callback.call(result); + } + break; + } + } + } + + public void failed(Throwable error) { + for (GuardedClosure cl : callbacks.get(FAILED)) { + boolean execute = true; + Object param; + + Closure guardExpr = cl.getGuard(); + if (null != guardExpr) { + Object guardResult = guardExpr.call(error); + if (null != guardResult) { + if (guardResult instanceof Boolean) { + execute = (Boolean) guardResult; + } else { + execute = true; + } + } + } + + if (execute) { + Closure callback = cl.getDelegate(); + callback.call(error); + } + } + } + + } + +} diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/MapReduceJob.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/MapReduceJob.java index 9acaaa9dc..103214b19 100644 --- a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/MapReduceJob.java +++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/MapReduceJob.java @@ -22,8 +22,8 @@ import java.util.List; import java.util.concurrent.Callable; /** - * A generic interface to representing a Map/Reduce job to a data store that - * supports that operation. + * A generic interface to representing a Map/Reduce job to a data store that supports that + * operation. * * @author J. Brisbin */ @@ -53,6 +53,13 @@ public interface MapReduceJob extends Callable { */ MapReduceJob addPhase(MapReducePhase phase); + /** + * Get the list of phases for this job. + * + * @return + */ + List getPhases(); + /** * Set the static argument for this job. * diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/RiakMapReduceJob.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/RiakMapReduceJob.java index 0fa3c2d38..c05902286 100644 --- a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/RiakMapReduceJob.java +++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/RiakMapReduceJob.java @@ -73,6 +73,10 @@ public class RiakMapReduceJob implements MapReduceJob { return this; } + public List getPhases() { + return this.phases; + } + public void setArg(Object arg) { this.arg = arg; } diff --git a/spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplateSpec.groovy b/spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplateSpec.groovy new file mode 100644 index 000000000..4d24c5eb9 --- /dev/null +++ b/spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplateSpec.groovy @@ -0,0 +1,87 @@ +/* + * Copyright (c) 2010 by J. Brisbin + * 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 + +import java.util.concurrent.Future +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.context.ApplicationContext +import org.springframework.test.context.ContextConfiguration +import spock.lang.Specification + +/** + * @author J. Brisbin + */ +@ContextConfiguration(locations = "/org/springframework/data/AsyncRiakTemplateTests.xml") +class AsyncRiakTemplateSpec extends Specification { + + @Autowired + ApplicationContext appCtx + @Autowired + AsyncRiakTemplate riak + + def "Test async setWithMetaData"() { + + given: + def obj = [test: "value", integer: 12] + def success = false + def failure = false + def testValue = "bad value" + def callback = [ + completed: { v -> + success = true + testValue = v.get().test + }, + failed: { e -> + failure = true + } + ] as AsyncKeyValueStoreOperation + + when: + Future future = riak.setWithMetaData("test", "test", obj, null, null, callback) + println "Waiting for result: ${future.get()}" + + then: + success && !failure + "value" == testValue + + } + + def "Test async getWithMetaData"() { + + given: + def result = null + def callback = [ + completed: { meta, v -> + println "got value: $meta $v" + result = v + }, + failed: { e -> + println "got error: $e" + } + ] as AsyncKeyValueStoreOperation + + when: + riak.getWithMetaData("test", "test", Map, callback).get() + + then: + null != result + + } + +} 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 new file mode 100644 index 000000000..af1d7d860 --- /dev/null +++ b/spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/RiakBuilderSpec.groovy @@ -0,0 +1,140 @@ +/* + * Copyright (c) 2010 by J. Brisbin + * 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 + +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.context.ApplicationContext +import org.springframework.data.keyvalue.riak.groovy.RiakBuilder +import org.springframework.test.context.ContextConfiguration +import spock.lang.Specification + +/** + * @author J. Brisbin + */ +@ContextConfiguration(locations = "/org/springframework/data/AsyncRiakTemplateTests.xml") +class RiakBuilderSpec extends Specification { + + @Autowired + ApplicationContext appCtx + @Autowired + AsyncRiakTemplate riakTemplate + + def "Test builder set"() { + + given: + def obj = [test: "value", integer: 12] + def riak = new RiakBuilder(riakTemplate) + def result = null + + when: + riak.set(bucket: "test", key: "test", qos: [dw: "all"], value: obj, wait: 3000L) { + + completed(when: { v -> v.integer == 12 }) { v, meta -> + result = v.test + } + completed { v -> result = "otherwise" } + + failed { e -> println "failure: $e" } + + } + + then: + "value" == result + + } + + def "Test builder get"() { + + given: + def riak = new RiakBuilder(riakTemplate) + def result = null + + when: + riak.get(bucket: "test", key: "test", wait: 3000L) { + + completed(when: { v -> v.integer == 12 }) { v, meta -> + result = v.test + } + completed { v -> result = "otherwise" } + + failed { e -> println "failure: $e" } + + } + + then: + "value" == result + + } + + def "Test builder setAsBytes"() { + + given: + def obj = "test bytes".bytes + def riak = new RiakBuilder(riakTemplate) + def result = null + + when: + riak.setAsBytes(bucket: "test", key: "test", value: obj, qos: [dw: "all"], wait: 3000L) { + completed { v -> result = "success" } + failed { e -> result = "failure" } + } + + then: + null != result + "success" == result + + } + + def "Test builder get with bytes"() { + + given: + def riak = new RiakBuilder(riakTemplate) + def result = null + + when: + riak.get(bucket: "test", key: "test", wait: 3000L) { + completed { v -> result = v } + failed { e -> println "failure: $e" } + } + + then: + null != result + "test bytes".bytes == result + + } + + def "Test builder delete"() { + + given: + def riak = new RiakBuilder(riakTemplate) + def result = null + + when: + riak.delete(bucket: "test", key: "test", wait: 3000L) { + completed { v -> result = v } + failed { e -> println "failure: $e" } + } + + then: + null != result + result + + } + +} diff --git a/spring-data-riak/src/test/resources/org/springframework/data/AsyncRiakTemplateTests.xml b/spring-data-riak/src/test/resources/org/springframework/data/AsyncRiakTemplateTests.xml new file mode 100644 index 000000000..86e08abe5 --- /dev/null +++ b/spring-data-riak/src/test/resources/org/springframework/data/AsyncRiakTemplateTests.xml @@ -0,0 +1,34 @@ + + + + + + + + + + +