Added java.io and Spring IO resource abstractions

This commit is contained in:
J. Brisbin
2010-12-13 16:37:28 -06:00
parent 1db9e72fde
commit 78d8819f9c
8 changed files with 706 additions and 5 deletions

View File

@@ -94,15 +94,15 @@ For those in a hurry:
-----
MyObject obj = new MyObject("value1", "value2");
riakTemplate.set("mybucket:mykey", obj);
riakTemplate.set("mybucket", "mykey", obj);
Map returnObj = riakTemplate.getAsType("mybucket:mykey", Map.class);
Map returnObj = riakTemplate.getAsType("mybucket", "mykey", Map.class);
Groovy:
-----
def obj = [first: "value1", second: "value2"]
riakTemplate.set([bucket: "mybucket", key: "mykey"], obj)
riakTemplate.set("mybucket", "mykey", obj)
Contributing to Spring Data

View File

@@ -129,7 +129,7 @@
<artifactId>com.springsource.bundlor.maven</artifactId>
</plugin>
<!-- For running Groovy/Spock tests
<!-- For running Groovy/Spock tests
<plugin>
<groupId>org.spockframework</groupId>
<artifactId>spock-maven</artifactId>

View File

@@ -176,6 +176,22 @@ public abstract class AbstractRiakTemplate extends RestGatewaySupport implements
this.defaultQosParameters = defaultQosParameters;
}
public String getHost() {
Matcher m = prefix.matcher(defaultUri);
if (m.matches()) {
return m.group(1);
}
return "localhost";
}
public Integer getPort() {
Matcher m = prefix.matcher(defaultUri);
if (m.matches()) {
return new Integer(m.group(2));
}
return 8098;
}
/**
* Extract the prefix from the URI for use in creating links.
*

View File

@@ -0,0 +1,391 @@
/*
* Copyright (c) 2010 by J. Brisbin <jon@jbrisbin.com>
* 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.io;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.keyvalue.riak.DataStoreOperationException;
import org.springframework.data.keyvalue.riak.core.KeyValueStoreMetaData;
import org.springframework.data.keyvalue.riak.core.RiakTemplate;
import org.springframework.data.keyvalue.riak.core.RiakValue;
import java.io.File;
import java.io.FileFilter;
import java.io.FilenameFilter;
import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
/**
* @author J. Brisbin <jon@jbrisbin.com>
*/
public class RiakFile<B, K> extends File {
private static final Logger log = LoggerFactory.getLogger(RiakFile.class);
private RiakTemplate riak;
private B bucket;
private K key;
public RiakFile(RiakTemplate riak, B bucket, K key) throws URISyntaxException {
super(riak.getDefaultUri());
this.riak = riak;
this.bucket = bucket;
this.key = key;
}
public String getUriAsString(boolean includeKey) {
String protocol = riak.getDefaultUri().substring(0, riak.getDefaultUri().indexOf(":"));
String uri = String.format("%s://%s:%s%s/%s/%s",
protocol,
riak.getHost(),
riak.getPort(),
riak.getPrefix(),
bucket,
(includeKey ? key : ""));
return uri;
}
public RiakTemplate getRiak() {
return riak;
}
public void setRiak(RiakTemplate riak) {
this.riak = riak;
}
public B getBucket() {
return bucket;
}
public void setBucket(B bucket) {
this.bucket = bucket;
}
public K getKey() {
return key;
}
public void setKey(K key) {
this.key = key;
}
@Override
public String getName() {
return getUriAsString(true);
}
@Override
public String getParent() {
return getUriAsString(false);
}
@SuppressWarnings({"unchecked"})
@Override
public File getParentFile() {
try {
return new RiakFile(riak, bucket, "");
} catch (URISyntaxException e) {
log.error(e.getMessage(), e);
}
return null;
}
@Override
public String getPath() {
return getUriAsString(true);
}
@Override
public boolean isAbsolute() {
return true;
}
@Override
public String getAbsolutePath() {
return getUriAsString(true);
}
@Override
public File getAbsoluteFile() {
return this;
}
@Override
public String getCanonicalPath() throws IOException {
return getUriAsString(true);
}
@Override
public File getCanonicalFile() throws IOException {
return this;
}
@SuppressWarnings({"deprecation"})
@Override
public URL toURL() throws MalformedURLException {
return new URL(getUriAsString(true));
}
@Override
public URI toURI() {
try {
return new URI(getUriAsString(true));
} catch (URISyntaxException e) {
log.error(e.getMessage(), e);
}
return null;
}
@Override
public boolean canRead() {
return true;
}
@Override
public boolean canWrite() {
return true;
}
@Override
public boolean exists() {
return riak.containsKey(bucket, key);
}
@Override
public boolean isDirectory() {
return (key != null && "".equals(key));
}
@Override
public boolean isFile() {
return (key != null && !("".equals(key)));
}
@Override
public boolean isHidden() {
return false;
}
@Override
public long lastModified() {
KeyValueStoreMetaData meta = riak.getMetaData(bucket, key);
return (null != meta ? meta.getLastModified() : null);
}
@Override
public long length() {
return riak.getAsBytes(bucket, key).length;
}
@Override
public boolean createNewFile() throws IOException {
return true;
}
@Override
public boolean delete() {
return riak.delete(bucket, key);
}
@Override
public void deleteOnExit() {
// NO-OP
}
@Override
public String[] list() {
if (isDirectory()) {
Map<String, Object> schema = riak.getBucketSchema(bucket, true);
String baseUri = getUriAsString(false);
List<String> uris = new LinkedList<String>();
for (Object key : (List) ((Map) schema.get("props")).get("keys")) {
uris.add(baseUri + key);
}
return (String[]) uris.toArray();
}
return new String[]{};
}
@Override
public String[] list(FilenameFilter filenameFilter) {
List<String> uris = new LinkedList<String>();
for (String s : list()) {
if (filenameFilter.accept(this, s)) {
uris.add(s);
}
}
return (String[]) uris.toArray();
}
@SuppressWarnings({"unchecked"})
@Override
public File[] listFiles() {
List<RiakFile> uris = new LinkedList<RiakFile>();
for (String s : list()) {
try {
uris.add(new RiakFile(riak, bucket, s.substring(s.lastIndexOf("/"))));
} catch (URISyntaxException e) {
log.error(e.getMessage(), e);
}
}
return (File[]) uris.toArray();
}
@SuppressWarnings({"unchecked"})
@Override
public File[] listFiles(FilenameFilter filenameFilter) {
List<RiakFile> uris = new LinkedList<RiakFile>();
for (String s : list(filenameFilter)) {
try {
uris.add(new RiakFile(riak, bucket, s.substring(s.lastIndexOf("/"))));
} catch (URISyntaxException e) {
log.error(e.getMessage(), e);
}
}
return (File[]) uris.toArray();
}
@Override
public File[] listFiles(FileFilter fileFilter) {
return super.listFiles(fileFilter); //To change body of overridden methods use File | Settings | File Templates.
}
@Override
public boolean mkdir() {
return true;
}
@Override
public boolean mkdirs() {
return true;
}
@SuppressWarnings({"unchecked"})
@Override
public boolean renameTo(File file) {
if (file instanceof RiakFile) {
RiakFile f = (RiakFile) file;
RiakValue<byte[]> v = riak.getAsBytesWithMetaData(bucket, key);
try {
riak.setWithMetaData(f.getBucket(), f.getKey(), v.get(),
(Map<String, String>) v.getMetaData());
} catch (DataStoreOperationException e) {
log.error(e.getMessage(), e);
return false;
}
} else {
throw new IllegalArgumentException("Renaming to a non-Riak file is not yet supported.");
}
return true;
}
@Override
public boolean setLastModified(long l) {
return false;
}
@Override
public boolean setReadOnly() {
return false;
}
@Override
public boolean setWritable(boolean b, boolean b1) {
return true;
}
@Override
public boolean setWritable(boolean b) {
return true;
}
@Override
public boolean setReadable(boolean b, boolean b1) {
return true;
}
@Override
public boolean setReadable(boolean b) {
return true;
}
@Override
public boolean setExecutable(boolean b, boolean b1) {
return false;
}
@Override
public boolean setExecutable(boolean b) {
return false;
}
@Override
public boolean canExecute() {
return false;
}
@Override
public long getTotalSpace() {
return super.getTotalSpace();
}
@Override
public long getFreeSpace() {
return super.getFreeSpace();
}
@Override
public long getUsableSpace() {
return super.getUsableSpace();
}
@Override
public int compareTo(File file) {
if (file instanceof RiakFile) {
return 0;
} else {
return -1;
}
}
@Override
public boolean equals(Object o) {
if (o instanceof RiakFile) {
RiakFile rf = (RiakFile) o;
return (rf.getBucket().equals(bucket) && rf.getKey().equals(key));
}
return false;
}
@Override
public int hashCode() {
return super.hashCode();
}
@Override
public String toString() {
return getClass().getSimpleName() + "@" + getUriAsString(true);
}
}

View File

@@ -0,0 +1,91 @@
/*
* Copyright (c) 2010 by J. Brisbin <jon@jbrisbin.com>
* 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.io;
import org.springframework.data.keyvalue.riak.core.RiakTemplate;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
/**
* An {@link java.io.InputStream} implementation that is backed by a resource residing in Riak.
*
* @author J. Brisbin <jon@jbrisbin.com>
*/
public class RiakInputStream<B, K> extends InputStream {
private RiakTemplate riak;
private B bucket;
private K key;
private ByteArrayInputStream in;
public RiakInputStream(RiakTemplate riak, B bucket, K key) {
this.riak = riak;
this.bucket = bucket;
this.key = key;
this.in = new ByteArrayInputStream(riak.getAsBytes(bucket, key));
}
@Override
public int read(byte[] bytes) throws IOException {
return in.read(bytes);
}
@Override
public int read(byte[] bytes, int i, int i1) throws IOException {
return in.read(bytes, i, i1);
}
@Override
public long skip(long l) throws IOException {
return in.skip(l);
}
@Override
public int available() throws IOException {
return in.available();
}
@Override
public void close() throws IOException {
in.close();
}
@Override
public void mark(int i) {
in.mark(i);
}
@Override
public void reset() throws IOException {
in.reset();
}
@Override
public boolean markSupported() {
return in.markSupported();
}
@Override
public int read() throws IOException {
return in.read();
}
}

View File

@@ -0,0 +1,46 @@
/*
* Copyright (c) 2010 by J. Brisbin <jon@jbrisbin.com>
* 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.io;
import org.springframework.data.keyvalue.riak.core.RiakTemplate;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
/**
* @author J. Brisbin <jon@jbrisbin.com>
*/
public class RiakOutputStream<B, K> extends ByteArrayOutputStream {
private RiakTemplate riak;
private B bucket;
private K key;
public RiakOutputStream(RiakTemplate riak, B bucket, K key) {
this.riak = riak;
this.bucket = bucket;
this.key = key;
}
@Override
public void flush() throws IOException {
super.flush();
riak.setAsBytes(bucket, key, toByteArray());
}
}

View File

@@ -0,0 +1,136 @@
/*
* Copyright (c) 2010 by J. Brisbin <jon@jbrisbin.com>
* 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.io;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.Resource;
import org.springframework.core.io.UrlResource;
import org.springframework.data.keyvalue.riak.core.RiakTemplate;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.net.MalformedURLException;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;
/**
* An implementation of {@link org.springframework.core.io.UrlResource} that is backed by a
* resource in Riak.
*
* @author J. Brisbin <jon@jbrisbin.com>
*/
public class RiakResource<B, K> extends UrlResource {
private static final Logger log = LoggerFactory.getLogger(RiakResource.class);
private RiakTemplate riak;
private B bucket;
private K key;
private String description;
public RiakResource(RiakTemplate riak, B bucket, K key) throws MalformedURLException {
super(riak.getDefaultUri());
this.bucket = bucket;
this.key = key;
}
public RiakTemplate getRiakTemplate() {
return this.riak;
}
public B getBucket() {
return bucket;
}
public K getKey() {
return key;
}
@SuppressWarnings({"unchecked"})
@Override
public URL getURL() throws IOException {
try {
return new URL(new RiakFile(riak, bucket, key).getUriAsString(true));
} catch (URISyntaxException e) {
log.error(e.getMessage(), e);
}
return null;
}
@SuppressWarnings({"unchecked"})
@Override
public URI getURI() throws IOException {
try {
return new URI(new RiakFile(riak, bucket, key).getUriAsString(true));
} catch (URISyntaxException e) {
log.error(e.getMessage(), e);
}
return null;
}
@SuppressWarnings({"unchecked"})
@Override
public Resource createRelative(String relativePath) throws MalformedURLException {
if (relativePath.startsWith("../")) {
return new RiakResource(riak, bucket, relativePath.substring(3));
}
return null;
}
@SuppressWarnings({"unchecked"})
@Override
public String getFilename() {
try {
return new RiakFile(riak, bucket, key).getUriAsString(true);
} catch (URISyntaxException e) {
log.error(e.getMessage(), e);
}
return riak.getDefaultUri();
}
public void setDescription(String description) {
this.description = description;
}
@Override
public String getDescription() {
return this.description;
}
@SuppressWarnings({"unchecked"})
@Override
public File getFile() throws IOException {
try {
return new RiakFile(riak, bucket, key);
} catch (URISyntaxException e) {
log.error(e.getMessage(), e);
}
return null;
}
@SuppressWarnings({"unchecked"})
@Override
public InputStream getInputStream() throws IOException {
return new RiakInputStream(riak, bucket, key);
}
}

View File

@@ -19,6 +19,7 @@ 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.core.io.RiakFile
import org.springframework.data.keyvalue.riak.mapreduce.JavascriptMapReduceOperation
import org.springframework.data.keyvalue.riak.mapreduce.MapReduceJob
import org.springframework.data.keyvalue.riak.mapreduce.RiakMapReducePhase
@@ -39,7 +40,7 @@ class RiakTemplateSpec extends Specification {
int run = 1
@Shared def riakBin = System.properties["bamboo.RIAK_BIN"] ?: "/usr/sbin/riak"
@Shared def p
/*
def setupSpec() {
p = "$riakBin start".execute()
p.waitFor()
@@ -50,6 +51,7 @@ class RiakTemplateSpec extends Specification {
p = "$riakBin stop".execute()
p.waitFor()
}
*/
def "Test Map object"() {
@@ -235,6 +237,25 @@ class RiakTemplateSpec extends Specification {
}
def "Test RiakFile"() {
given:
def file = new RiakFile(riak, "test", "test")
when:
def exists = file.exists()
then:
exists
when:
def content = file.toURI().toURL().openConnection().getContent()
then:
null != content
}
def "Test delete key"() {
given: