Repository: metamodel Updated Branches: refs/heads/master ddb2454ef -> 358379e93
METAMODEL-173: Added Resource.append() and Resource.write() methods implemented an abstract Resource class for reuse of common functionality Project: http://git-wip-us.apache.org/repos/asf/metamodel/repo Commit: http://git-wip-us.apache.org/repos/asf/metamodel/commit/7d21fba0 Tree: http://git-wip-us.apache.org/repos/asf/metamodel/tree/7d21fba0 Diff: http://git-wip-us.apache.org/repos/asf/metamodel/diff/7d21fba0 Branch: refs/heads/master Commit: 7d21fba0a5579c96a0ab1feb5818621cfa299e9a Parents: 15c94fc Author: Kasper Sørensen <[email protected]> Authored: Sat Aug 15 20:27:24 2015 +0200 Committer: Kasper Sørensen <[email protected]> Committed: Sat Aug 15 20:27:24 2015 +0200 ---------------------------------------------------------------------- .../apache/metamodel/util/AbstractResource.java | 83 ++++++++++++++ .../metamodel/util/ClasspathResource.java | 32 ++---- .../org/apache/metamodel/util/FileResource.java | 49 +-------- .../apache/metamodel/util/InMemoryResource.java | 66 +++++------- .../org/apache/metamodel/util/Resource.java | 26 +++++ .../org/apache/metamodel/util/UrlResource.java | 31 +----- .../org/apache/metamodel/util/HdfsResource.java | 108 +++++++++---------- 7 files changed, 205 insertions(+), 190 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/metamodel/blob/7d21fba0/core/src/main/java/org/apache/metamodel/util/AbstractResource.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/metamodel/util/AbstractResource.java b/core/src/main/java/org/apache/metamodel/util/AbstractResource.java new file mode 100644 index 0000000..7f96973 --- /dev/null +++ b/core/src/main/java/org/apache/metamodel/util/AbstractResource.java @@ -0,0 +1,83 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.apache.metamodel.util; + +import java.io.InputStream; +import java.io.OutputStream; + +/** + * Abstract implementation of many methods in {@link Resource} + */ +public abstract class AbstractResource implements Resource { + + + @Override + public final void read(Action<InputStream> readCallback) { + final InputStream in = read(); + try { + readCallback.run(in); + } catch (Exception e) { + throw new ResourceException(this, "Error occurred in read callback", e); + } finally { + FileHelper.safeClose(in); + } + } + + @Override + public final <E> E read(Func<InputStream, E> readCallback) { + final InputStream in = read(); + try { + final E result = readCallback.eval(in); + return result; + } catch (Exception e) { + throw new ResourceException(this, "Error occurred in read callback", e); + } finally { + FileHelper.safeClose(in); + } + } + + @Override + public final void write(Action<OutputStream> writeCallback) throws ResourceException { + final OutputStream out = write(); + try { + writeCallback.run(out); + } catch (Exception e) { + throw new ResourceException(this, "Error occurred in write callback", e); + } finally { + FileHelper.safeClose(out); + } + } + + @Override + public final void append(Action<OutputStream> appendCallback) throws ResourceException { + final OutputStream out = append(); + try { + appendCallback.run(out); + } catch (Exception e) { + throw new ResourceException(this, "Error occurred in append callback", e); + } finally { + FileHelper.safeClose(out); + } + } + + @Override + public String toString() { + return getClass().getSimpleName() + "[" + getQualifiedPath() + "]"; + } +} http://git-wip-us.apache.org/repos/asf/metamodel/blob/7d21fba0/core/src/main/java/org/apache/metamodel/util/ClasspathResource.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/metamodel/util/ClasspathResource.java b/core/src/main/java/org/apache/metamodel/util/ClasspathResource.java index e9e1234..32dbb59 100644 --- a/core/src/main/java/org/apache/metamodel/util/ClasspathResource.java +++ b/core/src/main/java/org/apache/metamodel/util/ClasspathResource.java @@ -26,7 +26,7 @@ import java.net.URL; /** * A {@link Resource} based on a classpath entry */ -public class ClasspathResource implements Resource, Serializable { +public class ClasspathResource extends AbstractResource implements Serializable { private static final long serialVersionUID = 1L; @@ -90,6 +90,16 @@ public class ClasspathResource implements Resource, Serializable { } return new UrlResource(url); } + + @Override + public OutputStream append() throws ResourceException { + return getUrlResourceDelegate().append(); + } + + @Override + public OutputStream write() throws ResourceException { + return getUrlResourceDelegate().write(); + } @Override public boolean isReadOnly() { @@ -124,28 +134,8 @@ public class ClasspathResource implements Resource, Serializable { } @Override - public void write(Action<OutputStream> writeCallback) throws ResourceException { - getUrlResourceDelegate().write(writeCallback); - } - - @Override - public void append(Action<OutputStream> appendCallback) throws ResourceException { - getUrlResourceDelegate().append(appendCallback); - } - - @Override public InputStream read() throws ResourceException { return getUrlResourceDelegate().read(); } - @Override - public void read(Action<InputStream> readCallback) throws ResourceException { - getUrlResourceDelegate().read(readCallback); - } - - @Override - public <E> E read(Func<InputStream, E> readCallback) throws ResourceException { - return getUrlResourceDelegate().read(readCallback); - } - } http://git-wip-us.apache.org/repos/asf/metamodel/blob/7d21fba0/core/src/main/java/org/apache/metamodel/util/FileResource.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/metamodel/util/FileResource.java b/core/src/main/java/org/apache/metamodel/util/FileResource.java index ab2d8c3..a5c80e9 100644 --- a/core/src/main/java/org/apache/metamodel/util/FileResource.java +++ b/core/src/main/java/org/apache/metamodel/util/FileResource.java @@ -29,7 +29,7 @@ import java.util.Arrays; /** * {@link File} based {@link Resource} implementation. */ -public class FileResource implements Resource, Serializable { +public class FileResource extends AbstractResource implements Serializable { private class DirectoryInputStream extends AbstractDirectoryInputStream<File> { @@ -95,52 +95,13 @@ public class FileResource implements Resource, Serializable { } @Override - public void write(Action<OutputStream> writeCallback) throws ResourceException { - final OutputStream out = FileHelper.getOutputStream(_file); - try { - writeCallback.run(out); - } catch (Exception e) { - throw new ResourceException(this, "Error occurred in write callback", e); - } finally { - FileHelper.safeClose(out); - } - } - - @Override - public void append(Action<OutputStream> appendCallback) { - final OutputStream out = FileHelper.getOutputStream(_file, true); - try { - appendCallback.run(out); - } catch (Exception e) { - throw new ResourceException(this, "Error occurred in append callback", e); - } finally { - FileHelper.safeClose(out); - } + public OutputStream write() throws ResourceException { + return FileHelper.getOutputStream(_file); } @Override - public void read(Action<InputStream> readCallback) { - final InputStream in = read(); - try { - readCallback.run(in); - } catch (Exception e) { - throw new ResourceException(this, "Error occurred in read callback", e); - } finally { - FileHelper.safeClose(in); - } - } - - @Override - public <E> E read(Func<InputStream, E> readCallback) { - final InputStream in = read(); - try { - final E result = readCallback.eval(in); - return result; - } catch (Exception e) { - throw new ResourceException(this, "Error occurred in read callback", e); - } finally { - FileHelper.safeClose(in); - } + public OutputStream append() throws ResourceException { + return FileHelper.getOutputStream(_file, true); } public File getFile() { http://git-wip-us.apache.org/repos/asf/metamodel/blob/7d21fba0/core/src/main/java/org/apache/metamodel/util/InMemoryResource.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/metamodel/util/InMemoryResource.java b/core/src/main/java/org/apache/metamodel/util/InMemoryResource.java index 3dafadf..7a1c58f 100644 --- a/core/src/main/java/org/apache/metamodel/util/InMemoryResource.java +++ b/core/src/main/java/org/apache/metamodel/util/InMemoryResource.java @@ -20,6 +20,7 @@ package org.apache.metamodel.util; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; +import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; import java.io.Serializable; @@ -27,7 +28,7 @@ import java.io.Serializable; /** * An entirely in-memory kept {@link Resource}. */ -public class InMemoryResource implements Resource, Serializable { +public class InMemoryResource extends AbstractResource implements Serializable { private static final long serialVersionUID = 1L; @@ -57,7 +58,7 @@ public class InMemoryResource implements Resource, Serializable { _contents = contents; _lastModified = lastModified; } - + @Override public String toString() { return "InMemoryResource[" + _path + "]"; @@ -113,53 +114,36 @@ public class InMemoryResource implements Resource, Serializable { } @Override - public void write(Action<OutputStream> writeCallback) throws ResourceException { - try { - ByteArrayOutputStream baos = new ByteArrayOutputStream(); - writeCallback.run(baos); - _contents = baos.toByteArray(); - _lastModified = System.currentTimeMillis(); - } catch (Exception e) { - throw new ResourceException(this, e); - } - } - - @Override - public void append(Action<OutputStream> appendCallback) throws ResourceException { - try { - ByteArrayOutputStream baos = new ByteArrayOutputStream(); - baos.write(_contents); - appendCallback.run(baos); - _contents = baos.toByteArray(); - _lastModified = System.currentTimeMillis(); - } catch (Exception e) { - throw new ResourceException(this, e); - } + public OutputStream write() throws ResourceException { + return createOutputStream(false); } @Override - public InputStream read() throws ResourceException { - return new ByteArrayInputStream(_contents); + public OutputStream append() throws ResourceException { + return createOutputStream(true); } - @Override - public void read(Action<InputStream> readCallback) throws ResourceException { - final InputStream inputStream = read(); - try { - readCallback.run(inputStream); - } catch (Exception e) { - throw new ResourceException(this, e); + private OutputStream createOutputStream(boolean append) throws ResourceException { + ByteArrayOutputStream out = new ByteArrayOutputStream() { + @Override + public void flush() throws IOException { + super.flush(); + _contents = toByteArray(); + _lastModified = System.currentTimeMillis(); + } + }; + if (append) { + try { + out.write(_contents); + } catch (IOException e) { + throw new ResourceException(this, e); + } } + return out; } @Override - public <E> E read(Func<InputStream, E> readCallback) throws ResourceException { - final InputStream inputStream = read(); - try { - return readCallback.eval(inputStream); - } catch (Exception e) { - throw new ResourceException(this, e); - } + public InputStream read() throws ResourceException { + return new ByteArrayInputStream(_contents); } - } http://git-wip-us.apache.org/repos/asf/metamodel/blob/7d21fba0/core/src/main/java/org/apache/metamodel/util/Resource.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/metamodel/util/Resource.java b/core/src/main/java/org/apache/metamodel/util/Resource.java index f1a3dc3..becb79f 100644 --- a/core/src/main/java/org/apache/metamodel/util/Resource.java +++ b/core/src/main/java/org/apache/metamodel/util/Resource.java @@ -88,6 +88,19 @@ public interface Resource extends HasName { public void write(Action<OutputStream> writeCallback) throws ResourceException; /** + * Opens up an {@link OutputStream} to write to the resource. Consumers of + * this method are expected to invoke the {@link OutputStream#close()} + * method manually. + * + * If possible, the other write(...) method is preferred over this one, + * since it guarantees proper closing of the resource's handles. + * + * @return + * @throws ResourceException + */ + public OutputStream write() throws ResourceException; + + /** * Opens up an {@link InputStream} to append (write at the end of the * existing stream) to the resource. * @@ -99,6 +112,19 @@ public interface Resource extends HasName { public void append(Action<OutputStream> appendCallback) throws ResourceException; /** + * Opens up an {@link OutputStream} to append to the resource. Consumers of + * this method are expected to invoke the {@link OutputStream#close()} + * method manually. + * + * If possible, the other append(...) method is preferred over this one, + * since it guarantees proper closing of the resource's handles. + * + * @return + * @throws ResourceException + */ + public OutputStream append() throws ResourceException; + + /** * Opens up an {@link InputStream} to read from the resource. Consumers of * this method are expected to invoke the {@link InputStream#close()} method * manually. http://git-wip-us.apache.org/repos/asf/metamodel/blob/7d21fba0/core/src/main/java/org/apache/metamodel/util/UrlResource.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/metamodel/util/UrlResource.java b/core/src/main/java/org/apache/metamodel/util/UrlResource.java index b0702aa..f2894fc 100644 --- a/core/src/main/java/org/apache/metamodel/util/UrlResource.java +++ b/core/src/main/java/org/apache/metamodel/util/UrlResource.java @@ -28,7 +28,7 @@ import java.net.URL; /** * Resource based on URL or URI. */ -public class UrlResource implements Resource, Serializable { +public class UrlResource extends AbstractResource implements Serializable { private static final long serialVersionUID = 1L; @@ -94,41 +94,16 @@ public class UrlResource implements Resource, Serializable { } @Override - public void write(Action<OutputStream> writeCallback) throws ResourceException { + public OutputStream write() throws ResourceException { throw new UnsupportedOperationException(); } @Override - public void append(Action<OutputStream> appendCallback) throws ResourceException { + public OutputStream append() throws ResourceException { throw new UnsupportedOperationException(); } @Override - public void read(Action<InputStream> readCallback) throws ResourceException { - final InputStream in = read(); - try { - readCallback.run(in); - } catch (Exception e) { - throw new ResourceException(this, "Error occurred in read callback", e); - } finally { - FileHelper.safeClose(in); - } - } - - @Override - public <E> E read(Func<InputStream, E> readCallback) throws ResourceException { - final InputStream in = read(); - try { - E result = readCallback.eval(in); - return result; - } catch (Exception e) { - throw new ResourceException(this, "Error occurred in read callback", e); - } finally { - FileHelper.safeClose(in); - } - } - - @Override public boolean isExists() { return true; } http://git-wip-us.apache.org/repos/asf/metamodel/blob/7d21fba0/hadoop/src/main/java/org/apache/metamodel/util/HdfsResource.java ---------------------------------------------------------------------- diff --git a/hadoop/src/main/java/org/apache/metamodel/util/HdfsResource.java b/hadoop/src/main/java/org/apache/metamodel/util/HdfsResource.java index 263ac1d..316cef8 100644 --- a/hadoop/src/main/java/org/apache/metamodel/util/HdfsResource.java +++ b/hadoop/src/main/java/org/apache/metamodel/util/HdfsResource.java @@ -38,8 +38,10 @@ import org.apache.metamodel.MetaModelException; * A {@link Resource} implementation that connects to Apache Hadoop's HDFS * distributed file system. */ -public class HdfsResource implements Resource, Serializable { +public class HdfsResource extends AbstractResource implements Serializable { + private static class HdfsFileInputStream extends InputStream { + private final InputStream _in; private final FileSystem _fs; @@ -96,12 +98,49 @@ public class HdfsResource implements Resource, Serializable { } } + private static class HdfsFileOutputStream extends OutputStream { + + private final OutputStream _out; + private final FileSystem _fs; + + public HdfsFileOutputStream(final OutputStream out, final FileSystem fs) { + _out = out; + _fs = fs; + } + + @Override + public void write(int b) throws IOException { + _out.write(b); + } + + @Override + public void write(byte[] b, int off, int len) throws IOException { + _out.write(b, off, len); + } + + @Override + public void write(byte[] b) throws IOException { + _out.write(b); + } + + @Override + public void flush() throws IOException { + _out.flush(); + } + + @Override + public void close() throws IOException { + super.close(); + // need to close 'fs' when output stream is closed + FileHelper.safeClose(_fs); + } + } + private class HdfsDirectoryInputStream extends AbstractDirectoryInputStream<FileStatus> { private final Path _hadoopPath; private final FileSystem _fs; - public HdfsDirectoryInputStream(final Path hadoopPath, - final FileSystem fs) { + public HdfsDirectoryInputStream(final Path hadoopPath, final FileSystem fs) { _hadoopPath = hadoopPath; _fs = fs; FileStatus[] fileStatuses; @@ -252,36 +291,28 @@ public class HdfsResource implements Resource, Serializable { } @Override - public void write(final Action<OutputStream> writeCallback) throws ResourceException { + public OutputStream write() throws ResourceException { final FileSystem fs = getHadoopFileSystem(); try { final FSDataOutputStream out = fs.create(getHadoopPath(), true); - try { - writeCallback.run(out); - } finally { - FileHelper.safeClose(out); - } - } catch (Exception e) { - throw wrapException(e); - } finally { + return new HdfsFileOutputStream(out, fs); + } catch (IOException e) { + // we can close 'fs' in case of an exception FileHelper.safeClose(fs); + throw wrapException(e); } } @Override - public void append(Action<OutputStream> appendCallback) throws ResourceException { + public OutputStream append() throws ResourceException { final FileSystem fs = getHadoopFileSystem(); try { final FSDataOutputStream out = fs.append(getHadoopPath()); - try { - appendCallback.run(out); - } finally { - FileHelper.safeClose(out); - } - } catch (Exception e) { - throw wrapException(e); - } finally { + return new HdfsFileOutputStream(out, fs); + } catch (IOException e) { + // we can close 'fs' in case of an exception FileHelper.safeClose(fs); + throw wrapException(e); } } @@ -303,41 +334,6 @@ public class HdfsResource implements Resource, Serializable { FileHelper.safeClose(fs); throw wrapException(e); } - - } - - @Override - public void read(Action<InputStream> readCallback) throws ResourceException { - final FileSystem fs = getHadoopFileSystem(); - try { - final InputStream in = read(); - try { - readCallback.run(in); - } finally { - FileHelper.safeClose(in); - } - } catch (Exception e) { - throw wrapException(e); - } finally { - FileHelper.safeClose(fs); - } - } - - @Override - public <E> E read(Func<InputStream, E> readCallback) throws ResourceException { - final FileSystem fs = getHadoopFileSystem(); - try { - final InputStream in = read(); - try { - return readCallback.eval(in); - } finally { - FileHelper.safeClose(in); - } - } catch (Exception e) { - throw wrapException(e); - } finally { - FileHelper.safeClose(fs); - } } private RuntimeException wrapException(Exception e) {
