From 8e75fbfaf04a294b8307d22c36fd8e202ffd6a72 Mon Sep 17 00:00:00 2001 From: "J. Brisbin" Date: Wed, 22 Dec 2010 16:40:19 -0600 Subject: [PATCH] Implement Map/Reduce for AsyncRiakTemplate, Groovy DSL, nesting operations into a node that serves as a default bucket. --- .../keyvalue/riak/core/AsyncRiakTemplate.java | 51 ++- .../keyvalue/riak/groovy/RiakBuilder.java | 298 ++++++++++++++---- .../riak/groovy/RiakMapReduceOperation.java | 110 +++++++ .../keyvalue/riak/groovy/RiakOperation.java | 4 +- .../mapreduce/AbstractRiakMapReduceJob.java | 142 +++++++++ .../mapreduce/AsyncMapReduceOperations.java | 41 +++ .../riak/mapreduce/AsyncRiakMapReduceJob.java | 48 +++ .../keyvalue/riak/mapreduce/MapReduceJob.java | 15 - .../riak/mapreduce/MapReducePhase.java | 13 + .../riak/mapreduce/RiakMapReduceJob.java | 120 +------ .../riak/mapreduce/RiakMapReducePhase.java | 9 + .../keyvalue/riak/core/RiakBuilderSpec.groovy | 77 ++++- 12 files changed, 716 insertions(+), 212 deletions(-) create mode 100644 spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakMapReduceOperation.java create mode 100644 spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/AbstractRiakMapReduceJob.java create mode 100644 spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/AsyncMapReduceOperations.java create mode 100644 spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/AsyncRiakMapReduceJob.java 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 64d061984..09145e747 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 @@ -22,8 +22,11 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.dao.DataAccessResourceFailureException; import org.springframework.data.keyvalue.riak.DataStoreOperationException; +import org.springframework.data.keyvalue.riak.mapreduce.AsyncMapReduceOperations; +import org.springframework.data.keyvalue.riak.mapreduce.MapReduceJob; 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; @@ -43,7 +46,7 @@ import java.util.concurrent.Future; /** * @author J. Brisbin */ -public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBucketKeyValueStoreOperations { +public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBucketKeyValueStoreOperations, AsyncMapReduceOperations { protected final Logger log = LoggerFactory.getLogger(getClass()); @@ -301,6 +304,17 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck return setWithMetaData(bucket, key, value, metaData, null, callback); } + /* ---------------- Map/Reduce ---------------- */ + + @SuppressWarnings({"unchecked"}) + public Future execute(MapReduceJob job, AsyncKeyValueStoreOperation> callback) { + HttpHeaders headers = defaultHeaders(null); + headers.setContentType(MediaType.APPLICATION_JSON); + HttpEntity json = new HttpEntity(job.toJson(), headers); + return workerPool.submit(new AsyncMapReduce(json, callback)); + } + + /* ---------------- Runnable helpers ---------------- */ protected class AsyncPut implements Runnable { private String bucket; @@ -384,6 +398,41 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck } + protected class AsyncMapReduce implements Runnable { + + private HttpEntity entity = null; + private AsyncKeyValueStoreOperation> callback = null; + + public AsyncMapReduce(HttpEntity entity, AsyncKeyValueStoreOperation> callback) { + this.entity = entity; + this.callback = callback; + } + + @SuppressWarnings({"unchecked"}) + public void run() { + try { + HttpEntity result = getRestTemplate().postForEntity(mapReduceUri, + entity, + List.class); + if (log.isDebugEnabled()) { + log.debug(String.format("M/R: json=%s", entity.getBody())); + } + if (null != callback) { + RiakMetaData meta = extractMetaData(result.getHeaders()); + callback.completed(meta, 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; 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 d3acd9f18..e5bbb3ce8 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 @@ -26,7 +26,11 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.keyvalue.riak.DataStoreOperationException; import org.springframework.data.keyvalue.riak.core.AsyncRiakTemplate; import org.springframework.data.keyvalue.riak.core.RiakQosParameters; +import org.springframework.data.keyvalue.riak.core.SimpleBucketKeyPair; +import org.springframework.data.keyvalue.riak.mapreduce.*; +import java.util.ArrayList; +import java.util.List; import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -37,10 +41,14 @@ import java.util.concurrent.Executors; public class RiakBuilder extends BuilderSupport { protected final Logger log = LoggerFactory.getLogger(getClass()); - @Autowired + @Autowired(required = false) protected AsyncRiakTemplate riak; - @Autowired + @Autowired(required = false) protected ExecutorService workerPool = Executors.newCachedThreadPool(); + protected String defaultBucketName; + + public RiakBuilder() { + } public RiakBuilder(AsyncRiakTemplate riak) { this.riak = riak; @@ -61,6 +69,14 @@ public class RiakBuilder extends BuilderSupport { this.riak = riak; } + public AsyncRiakTemplate getAsyncTemplate() { + return riak; + } + + public void setAsyncTemplate(AsyncRiakTemplate riak) { + this.riak = riak; + } + public ExecutorService getWorkerPool() { return workerPool; } @@ -74,15 +90,65 @@ public class RiakBuilder extends BuilderSupport { log.debug("setParent/2 " + parent + " " + child); } + @SuppressWarnings({"unchecked"}) @Override protected Object createNode(Object name) { log.debug("createNode/1 " + name); - return this; + if ("call".equals(name)) { + // IGNORED + } else if ("foreach".equals(name)) { + RiakOperation op = new RiakOperation(riak, RiakOperation.Type.FOREACH); + op.setBucket(defaultBucketName); + return op; + } else if ("mapreduce".equals(name)) { + return createMapReduceJob(); + } else if ("query".equals(name)) { + QueryPhase p = new QueryPhase(); + p.job = ((RiakMapReduceOperation) getCurrent()).getJob(); + return getCurrent(); + } else if ("map".equals(name) || "reduce".equals(name)) { + QueryPhase p = new QueryPhase(); + p.job = ((RiakMapReduceOperation) getCurrent()).getJob(); + p.phase = name.toString(); + return p; + } else { + defaultBucketName = name.toString(); + } + + return null; } + @SuppressWarnings({"unchecked"}) @Override protected Object createNode(Object name, Object value) { log.debug("createNode/2 " + name + " " + value); + if ("inputs".equals(name)) { + AsyncRiakMapReduceJob job = ((RiakMapReduceOperation) getCurrent()).getJob(); + if (null != value && value instanceof String) { + List keys = new ArrayList(); + keys.add(value.toString()); + job.addInputs(keys); + } else if (value instanceof List) { + job.addInputs((List) value); + } + return job; + } else if ("language".equals(name)) { + QueryPhase p = (QueryPhase) getCurrent(); + p.language = value.toString(); + return p; + } else if ("source".equals(name)) { + QueryPhase p = (QueryPhase) getCurrent(); + p.source = value.toString(); + return p; + } else if ("keep".equals(name)) { + QueryPhase p = (QueryPhase) getCurrent(); + p.keep = (value instanceof Boolean ? (Boolean) value : new Boolean(value.toString())); + return p; + } else if ("arg".equals(name)) { + QueryPhase p = (QueryPhase) getCurrent(); + p.arg = value; + return p; + } return null; //To change body of implemented methods use File | Settings | File Templates. } @@ -90,59 +156,96 @@ public class RiakBuilder extends BuilderSupport { @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("type"); - if (null != o) { - if (o instanceof Class) { - op.setRequiredType((Class) o); - } else if (o instanceof String) { - try { - op.setRequiredType(Class.forName((String) o)); - } catch (ClassNotFoundException e) { - throw new DataStoreOperationException(e.getMessage(), e); - } - } else { - op.setRequiredType(o.getClass()); - } - } - 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 ("mapreduce".equals(name)) { + RiakMapReduceOperation oper = createMapReduceJob(); + // Set timeout + Object o = attributes.get("wait"); if (null != o) { if (o instanceof Long) { - op.setTimeout((Long) o); + oper.setTimeout((Long) o); } else if (o instanceof String) { - op.setTimeout(new Long(o.toString())); + oper.setTimeout(new Long(o.toString())); } else if (o instanceof Integer) { - op.setTimeout(new Long((Integer) o)); + oper.setTimeout(new Long((Integer) o)); } else { throw new IllegalArgumentException( "Timeout should be an Integer, a Long, or a String denoting milliseconds"); } } - return op; + return oper; + } else if ("map".equals(name) || "reduce".equals(name)) { + QueryPhase p = new QueryPhase(); + p.job = ((RiakMapReduceOperation) getCurrent()).getJob(); + p.phase = name.toString(); + // Set arg + p.arg = attributes.get("arg"); + return p; + } else { + RiakOperation.Type type = RiakOperation.Type.valueOf(name.toString().toUpperCase()); + if (null != type) { + RiakOperation op = new RiakOperation(riak, type); + // Set a bucket name + Object o = attributes.get("bucket"); + if (null == o && null != defaultBucketName) { + op.setBucket(defaultBucketName); + } else { + op.setBucket((null != o ? o.toString() : null)); + } + // Set the object's key + o = attributes.get("key"); + op.setKey((null != o ? o.toString() : null)); + // Set the value + o = attributes.get("value"); + op.setValue(o); + // Set the type of object (for getAsType) + o = attributes.get("type"); + if (null != o) { + if (o instanceof Class) { + op.setRequiredType((Class) o); + } else if (o instanceof String) { + try { + op.setRequiredType(Class.forName((String) o)); + } catch (ClassNotFoundException e) { + throw new DataStoreOperationException(e.getMessage(), e); + } + } else { + op.setRequiredType(o.getClass()); + } + } + // Set QOS parameters + 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); + } + // Set timeout + o = attributes.get("wait"); + if (null != o) { + if (o instanceof Long) { + op.setTimeout((Long) o); + } else if (o instanceof String) { + op.setTimeout(new Long(o.toString())); + } else if (o instanceof Integer) { + op.setTimeout(new Long((Integer) o)); + } else { + throw new IllegalArgumentException( + "Timeout should be an Integer, a Long, or a String denoting milliseconds"); + } + } + + return op; + } } return null; } @@ -163,38 +266,45 @@ public class RiakBuilder extends BuilderSupport { @Override public Object invokeMethod(String methodName, Object arg) { if (log.isDebugEnabled()) { - log.debug("invokeMethod: " + methodName + " " + arg); + log.debug("invokeMethod/2: " + methodName + " " + arg); } - - Object[] args = (Object[]) arg; - Map params; - Closure handler = null; - RiakOperation op; - if ("completed".equals(methodName) || "failed".equals(methodName)) { - op = (RiakOperation) getCurrent(); - Closure guard = null; - for (Object o : args) { - if (o instanceof Map) { - params = (Map) o; - if (params.containsKey("when")) { - guard = (Closure) params.get("when"); + if (getCurrent() instanceof RiakOperation) { + 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; } - } else if (o instanceof Closure) { - handler = (Closure) o; } + op.addHandler(methodName, handler, guard); + return op; + } else if (getCurrent() instanceof RiakMapReduceOperation) { + RiakMapReduceOperation oper = (RiakMapReduceOperation) getCurrent(); + Object[] args = (Object[]) arg; + if ("completed".equals(methodName)) { + oper.setCompleted((Closure) args[0]); + } else if ("failed".equals(methodName)) { + oper.setFailed((Closure) args[0]); + } + return oper; } - 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); + log.debug("nodeCompleted: parent=" + parent + ", node=" + node); if (node instanceof RiakOperation) { RiakOperation op = (RiakOperation) node; try { @@ -202,6 +312,26 @@ public class RiakBuilder extends BuilderSupport { } catch (Exception e) { log.error(e.getMessage(), e); } + } else if (parent instanceof RiakMapReduceOperation && node instanceof QueryPhase) { + QueryPhase p = (QueryPhase) node; + MapReduceOperation oper = null; + if ("javascript".equals(p.language)) { + if (null != p.source) { + oper = new JavascriptMapReduceOperation(p.source); + } else if (null != p.bucket && null != p.key) { + oper = new JavascriptMapReduceOperation(new SimpleBucketKeyPair(p.bucket, p.key)); + } + } else { + oper = new ErlangMapReduceOperation(p.module, p.func); + } + if (null != oper) { + RiakMapReducePhase phase = new RiakMapReducePhase(p.phase, p.language, oper); + if (null != p.keep) { + phase.setKeepResults(p.keep); + } + phase.setArg(p.arg); + p.job.addPhase(phase); + } } else { super.nodeCompleted(parent, node); } @@ -211,7 +341,43 @@ public class RiakBuilder extends BuilderSupport { @Override protected Object postNodeCompletion(Object parent, Object node) { log.debug("postNodeCompletion: " + parent + " " + node); + if (null == parent && node instanceof RiakMapReduceOperation) { + RiakMapReduceOperation oper = (RiakMapReduceOperation) node; + try { + return oper.call(); + } catch (Exception e) { + log.error(e.getMessage(), e); + } + } else if (null == parent && node == parent) { + defaultBucketName = null; + } + return super.postNodeCompletion(parent, node); } + protected RiakMapReduceOperation createMapReduceJob() { + AsyncRiakMapReduceJob job = new AsyncRiakMapReduceJob(riak); + if (null != defaultBucketName) { + List keys = new ArrayList(); + keys.add(defaultBucketName); + job.addInputs(keys); + } + return new RiakMapReduceOperation(riak, job); + } + + private class QueryPhase { + + AsyncRiakMapReduceJob job; + String phase; + String language = "javascript"; + String source = null; + String bucket = null; + String key = null; + String module = null; + String func = null; + Boolean keep = null; + Object arg = null; + + } + } diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakMapReduceOperation.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakMapReduceOperation.java new file mode 100644 index 000000000..2e72b0b0c --- /dev/null +++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakMapReduceOperation.java @@ -0,0 +1,110 @@ +/* + * 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.mapreduce.AsyncRiakMapReduceJob; + +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +/** + * @author J. Brisbin + */ +public class RiakMapReduceOperation implements Callable { + + protected final Logger log = LoggerFactory.getLogger(getClass()); + + protected AsyncRiakTemplate riak; + protected AsyncRiakMapReduceJob job; + protected Long timeout = -1L; + protected Closure completed = null; + protected Closure failed = null; + + public RiakMapReduceOperation(AsyncRiakTemplate riak, AsyncRiakMapReduceJob job) { + this.riak = riak; + this.job = job; + } + + public AsyncRiakMapReduceJob getJob() { + return job; + } + + public void setJob(AsyncRiakMapReduceJob job) { + this.job = job; + } + + public Long getTimeout() { + return timeout; + } + + public void setTimeout(Long timeout) { + this.timeout = timeout; + } + + public Closure getCompleted() { + return completed; + } + + public void setCompleted(Closure completed) { + this.completed = completed; + } + + public Closure getFailed() { + return failed; + } + + public void setFailed(Closure failed) { + this.failed = failed; + } + + public Object call() throws Exception { + Future f = riak.execute(job, new AsyncKeyValueStoreOperation>() { + public void completed(KeyValueStoreMetaData meta, List result) { + Object arg = new Object[]{result, meta}; + if (null != completed) { + completed.call(arg); + } + } + + public void failed(Throwable error) { + if (null != failed) { + failed.call(error); + } else { + throw new RuntimeException(error); + } + } + }); + + if (timeout == 0) { + return f; + } else if (timeout < 0) { + return f.get(); + } else { + return f.get(timeout, TimeUnit.MILLISECONDS); + } + } +} 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 index 0a22420eb..33d93f65d 100644 --- 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 @@ -39,7 +39,7 @@ import java.util.concurrent.*; public class RiakOperation implements Callable { static enum Type { - SET, SETASBYTES, PUT, GET, GETASBYTES, GETASTYPE, CONTAINSKEY, DELETE, EACH + SET, SETASBYTES, PUT, GET, GETASBYTES, GETASTYPE, CONTAINSKEY, DELETE, FOREACH } static String COMPLETED = "completed"; @@ -162,7 +162,7 @@ public class RiakOperation implements Callable { case DELETE: f = riak.delete(bucket, key, callbackInvoker); break; - case EACH: + case FOREACH: f = riak.getBucketSchema(bucket, null, new AsyncKeyValueStoreOperation>() { diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/AbstractRiakMapReduceJob.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/AbstractRiakMapReduceJob.java new file mode 100644 index 000000000..7de70e631 --- /dev/null +++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/AbstractRiakMapReduceJob.java @@ -0,0 +1,142 @@ +/* + * 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.mapreduce; + +import org.codehaus.jackson.JsonFactory; +import org.codehaus.jackson.JsonGenerator; +import org.codehaus.jackson.map.ObjectMapper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.data.keyvalue.riak.core.BucketKeyPair; + +import java.io.IOException; +import java.io.StringWriter; +import java.util.ArrayList; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; + +/** + * An implementation of {@link MapReduceJob} for the Riak data store. + * + * @author J. Brisbin + */ +@SuppressWarnings({"unchecked"}) +public abstract class AbstractRiakMapReduceJob implements MapReduceJob { + + protected final Logger log = LoggerFactory.getLogger(getClass()); + protected List inputs = new LinkedList(); + protected List phases = new ArrayList(); + + public List getInputs() { + return this.inputs; + } + + public MapReduceJob addInputs(List keys) { + inputs.addAll(keys); + return this; + } + + public MapReduceJob addPhase(MapReducePhase phase) { + phases.add(phase); + return this; + } + + public List getPhases() { + return this.phases; + } + + public String toJson() { + StringWriter out = new StringWriter(); + try { + JsonGenerator json = new JsonFactory().createJsonGenerator(out); + json.setCodec(new ObjectMapper()); + json.writeStartObject(); + + // Inputs + json.writeFieldName("inputs"); + if (1 == inputs.size() && !(inputs.get(0) instanceof List)) { + json.writeString(inputs.get(0).toString()); + } else if (inputs.size() > 0) { + json.writeStartArray(); + for (Object obj : inputs) { + List pair = (List) obj; + json.writeStartArray(); + json.writeString(pair.get(0).toString()); + json.writeString(pair.get(1).toString()); + json.writeEndArray(); + } + json.writeEndArray(); + } + + // Query + json.writeFieldName("query"); + json.writeStartArray(); + for (MapReducePhase phase : phases) { + json.writeStartObject(); + switch (phase.getPhase()) { + case MAP: + json.writeFieldName("map"); + break; + case REDUCE: + json.writeFieldName("reduce"); + break; + } + + json.writeStartObject(); + json.writeStringField("language", phase.getLanguage()); + Object repr = phase.getOperation().getRepresentation(); + if (repr instanceof String) { + // Using source + json.writeStringField("source", + String.format("%s", phase.getOperation().getRepresentation())); + } else if (repr instanceof BucketKeyPair) { + BucketKeyPair pair = (BucketKeyPair) repr; + json.writeStringField("bucket", + String.format("%s", pair.getBucket())); + json.writeStringField("key", String.format("%s", pair.getKey())); + } else if (repr instanceof Map) { + for (Map.Entry entry : ((Map) repr).entrySet()) { + json.writeStringField(entry.getKey().toString(), + entry.getValue().toString()); + } + } + if (phase.getKeepResults()) { + json.writeBooleanField("keep", true); + } + // Arg + if (null != phase.getArg()) { + json.writeObjectField("arg", phase.getArg()); + } + + json.writeEndObject(); + json.writeEndObject(); + } + json.writeEndArray(); + + json.writeEndObject(); + json.flush(); + + } catch (IOException e) { + log.error(e.getMessage(), e); + } + return out.toString(); + } + +} diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/AsyncMapReduceOperations.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/AsyncMapReduceOperations.java new file mode 100644 index 000000000..a42162d90 --- /dev/null +++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/AsyncMapReduceOperations.java @@ -0,0 +1,41 @@ +/* + * 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.mapreduce; + +import org.springframework.data.keyvalue.riak.core.AsyncKeyValueStoreOperation; + +import java.util.List; +import java.util.concurrent.Future; + +/** + * Generic interface to Map/Reduce in data stores that support it. + * + * @author J. Brisbin + */ +public interface AsyncMapReduceOperations { + + /** + * Execute a {@link MapReduceJob} synchronously. + * + * @param job + * @return + */ + Future execute(MapReduceJob job, AsyncKeyValueStoreOperation> callback); + +} diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/AsyncRiakMapReduceJob.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/AsyncRiakMapReduceJob.java new file mode 100644 index 000000000..0679c6222 --- /dev/null +++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/AsyncRiakMapReduceJob.java @@ -0,0 +1,48 @@ +/* + * 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.mapreduce; + +import org.springframework.data.keyvalue.riak.core.AsyncRiakTemplate; + +/** + * An implementation of {@link MapReduceJob} for the Riak data store. + * + * @author J. Brisbin + */ +@SuppressWarnings({"unchecked"}) +public class AsyncRiakMapReduceJob extends AbstractRiakMapReduceJob { + + protected AsyncRiakTemplate riakTemplate; + + public AsyncRiakMapReduceJob(AsyncRiakTemplate riakTemplate) { + this.riakTemplate = riakTemplate; + } + + public AsyncRiakTemplate getAsyncRiakTemplate() { + return riakTemplate; + } + + public void setAsyncRiakTemplate(AsyncRiakTemplate riakTemplate) { + this.riakTemplate = riakTemplate; + } + + public Object call() throws Exception { + return riakTemplate.execute(this, null); + } +} 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 103214b19..19ceebd6f 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 @@ -60,21 +60,6 @@ public interface MapReduceJob extends Callable { */ List getPhases(); - /** - * Set the static argument for this job. - * - * @param arg - */ - void setArg(T arg); - - /** - * Get the static argument for this job. - * - * @param - * @return - */ - T getArg(); - /** * Convert this job into the appropriate JSON to send to the server. * diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/MapReducePhase.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/MapReducePhase.java index a7afe522c..030d75c16 100644 --- a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/MapReducePhase.java +++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/MapReducePhase.java @@ -52,4 +52,17 @@ public interface MapReducePhase { */ MapReduceOperation getOperation(); + /** + * Set the static argument for this job. + * + * @param arg + */ + void setArg(Object arg); + + /** + * Get the static argument for this phase. + * + * @return + */ + Object getArg(); } 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 c05902286..424a4dad6 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 @@ -18,20 +18,8 @@ package org.springframework.data.keyvalue.riak.mapreduce; -import org.codehaus.jackson.JsonFactory; -import org.codehaus.jackson.JsonGenerator; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.data.keyvalue.riak.core.BucketKeyPair; import org.springframework.data.keyvalue.riak.core.RiakTemplate; -import java.io.IOException; -import java.io.StringWriter; -import java.util.ArrayList; -import java.util.LinkedList; -import java.util.List; -import java.util.Map; - /** * An implementation of {@link org.springframework.data.keyvalue.riak.mapreduce.MapReduceJob} * for the Riak data store. @@ -39,12 +27,8 @@ import java.util.Map; * @author J. Brisbin */ @SuppressWarnings({"unchecked"}) -public class RiakMapReduceJob implements MapReduceJob { +public class RiakMapReduceJob extends AbstractRiakMapReduceJob { - protected final Logger log = LoggerFactory.getLogger(getClass()); - protected List inputs = new LinkedList(); - protected List phases = new ArrayList(); - protected Object arg = null; protected RiakTemplate riakTemplate; public RiakMapReduceJob(RiakTemplate riakTemplate) { @@ -59,108 +43,6 @@ public class RiakMapReduceJob implements MapReduceJob { this.riakTemplate = riakTemplate; } - public List getInputs() { - return this.inputs; - } - - public MapReduceJob addInputs(List keys) { - inputs.addAll(keys); - return this; - } - - public MapReduceJob addPhase(MapReducePhase phase) { - phases.add(phase); - return this; - } - - public List getPhases() { - return this.phases; - } - - public void setArg(Object arg) { - this.arg = arg; - } - - public Object getArg() { - return this.arg; - } - - public String toJson() { - StringWriter out = new StringWriter(); - try { - JsonGenerator json = new JsonFactory().createJsonGenerator(out); - json.writeStartObject(); - - // Inputs - json.writeFieldName("inputs"); - if (1 == inputs.size() && !(inputs.get(0) instanceof List)) { - json.writeString(inputs.get(0).toString()); - } else if (inputs.size() > 0) { - json.writeStartArray(); - for (Object obj : inputs) { - List pair = (List) obj; - json.writeStartArray(); - json.writeString(pair.get(0).toString()); - json.writeString(pair.get(1).toString()); - json.writeEndArray(); - } - json.writeEndArray(); - } - - // Query - json.writeFieldName("query"); - json.writeStartArray(); - for (MapReducePhase phase : phases) { - json.writeStartObject(); - switch (phase.getPhase()) { - case MAP: - json.writeFieldName("map"); - break; - case REDUCE: - json.writeFieldName("reduce"); - break; - } - - json.writeStartObject(); - json.writeStringField("language", phase.getLanguage()); - Object repr = phase.getOperation().getRepresentation(); - if (repr instanceof String) { - // Using source - json.writeStringField("source", - String.format("%s", phase.getOperation().getRepresentation())); - } else if (repr instanceof BucketKeyPair) { - BucketKeyPair pair = (BucketKeyPair) repr; - json.writeStringField("bucket", - String.format("%s", pair.getBucket())); - json.writeStringField("key", String.format("%s", pair.getKey())); - } else if (repr instanceof Map) { - for (Map.Entry entry : ((Map) repr).entrySet()) { - json.writeStringField(entry.getKey().toString(), - entry.getValue().toString()); - } - } - if (phase.getKeepResults()) { - json.writeBooleanField("keep", true); - } - json.writeEndObject(); - json.writeEndObject(); - } - json.writeEndArray(); - - // Arg - if (null != arg) { - json.writeObjectField("arg", arg); - } - - json.writeEndObject(); - json.flush(); - - } catch (IOException e) { - log.error(e.getMessage(), e); - } - return out.toString(); - } - public Object call() throws Exception { return riakTemplate.execute(this); } diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/RiakMapReducePhase.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/RiakMapReducePhase.java index 9f330ca0e..dfdee5ab9 100644 --- a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/RiakMapReducePhase.java +++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/RiakMapReducePhase.java @@ -30,6 +30,7 @@ public class RiakMapReducePhase implements MapReducePhase { protected String language; protected MapReduceOperation operation; protected boolean keepResults = false; + protected Object arg; public RiakMapReducePhase(String phase, String language, MapReduceOperation oper) { this.phase = Phase.valueOf(phase.toUpperCase()); @@ -67,4 +68,12 @@ public class RiakMapReducePhase implements MapReducePhase { this.operation = oper; } + + public Object getArg() { + return arg; + } + + public void setArg(Object arg) { + this.arg = arg; + } } 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 93ba07b91..d2e1c9cdf 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 @@ -147,14 +147,14 @@ class RiakBuilderSpec extends Specification { } - def "Test builder each"() { + def "Test builder foreach"() { given: def riak = new RiakBuilder(riakTemplate) def idCnt = 0 when: - riak.each(bucket: "test") { + riak.foreach(bucket: "test") { completed { idCnt++ } failed { it.printStackTrace() } } @@ -176,7 +176,7 @@ class RiakBuilderSpec extends Specification { put(bucket: "test", value: [test: "value 2"]) put(bucket: "test", value: [test: "value 3"]) - each(bucket: "test") { + foreach(bucket: "test") { completed { v, meta -> ids << meta.key } failed { it.printStackTrace() } } @@ -188,6 +188,61 @@ class RiakBuilderSpec extends Specification { } + def "Test builder bucket as node"() { + + given: + def riak = new RiakBuilder(riakTemplate) + def ids = [] + + when: + riak { + "test" { + put(value: [test: "value 1"]) + put(value: [test: "value 2"]) + put(value: [test: "value 3"]) + + foreach { + completed { v, meta -> ids << meta.key } + failed { it.printStackTrace() } + } + } + } + + then: + null != ids + 3 <= ids.size() + + } + + def "Test builder Map/Reduce"() { + + given: + def riak = new RiakBuilder(riakTemplate) + def result = [] + + when: + riak { + mapreduce { + inputs "test" + query { + map(arg: [test: "arg", alist: [1, 2, 3, 4]]) { + source "function(v, keyInfo, arg){ ejsLog('/tmp/mapred.log', JSON.stringify(v)); ejsLog('/tmp/mapred.log', JSON.stringify(keyInfo)); ejsLog('/tmp/mapred.log', JSON.stringify(arg)); return [1]; }" + } + reduce { + source "function(v){ ejsLog('/tmp/mapred.log', JSON.stringify(arguments)); return Riak.reduceSum(v); }" + } + } + completed { result = it } + failed { it.printStackTrace() } + } + } + + then: + null != result + 1 <= result.size() + + } + def "Test builder delete"() { given: @@ -195,14 +250,18 @@ class RiakBuilderSpec extends Specification { def deleted = false when: - riak.each(bucket: "test") { - completed { v, meta -> - delete(bucket: meta.bucket, key: meta.key) { - completed { deleted = true } - failed { deleted = false } + riak { + "test" { + foreach { + completed { v, meta -> + delete(bucket: meta.bucket, key: meta.key) { + completed { deleted = true } + failed { deleted = false } + } + } + failed { it.printStackTrace() } } } - failed { it.printStackTrace() } } then: