You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by dh...@apache.org on 2016/07/22 03:18:05 UTC
[2/3] incubator-beam git commit: [BEAM-316] Add file URI handling
[BEAM-316] Add file URI handling
* Register FileIOChannelFactory for file scheme
* Modify FileIOChannelFactory to dynamically remove the file:// scheme string.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/38061d35
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/38061d35
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/38061d35
Branch: refs/heads/master
Commit: 38061d3531a1ca13ef825a6ee50872e420ab1494
Parents: 7337ecf
Author: Dan Halperin <dh...@google.com>
Authored: Mon Jul 18 15:35:06 2016 -0700
Committer: Dan Halperin <dh...@google.com>
Committed: Thu Jul 21 20:17:54 2016 -0700
----------------------------------------------------------------------
.../beam/sdk/util/FileIOChannelFactory.java | 31 ++++++++++++++++----
.../beam/sdk/util/FileIOChannelFactoryTest.java | 26 +++++++++++++---
.../beam/sdk/util/IOChannelUtilsTest.java | 9 +++---
3 files changed, 52 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/38061d35/sdks/java/core/src/main/java/org/apache/beam/sdk/util/FileIOChannelFactory.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/FileIOChannelFactory.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/FileIOChannelFactory.java
index 5033dca..92f351b 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/FileIOChannelFactory.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/FileIOChannelFactory.java
@@ -30,12 +30,14 @@ import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
+import java.net.URI;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;
import java.nio.file.FileSystems;
import java.nio.file.Files;
import java.nio.file.NoSuchFileException;
+import java.nio.file.Path;
import java.nio.file.PathMatcher;
import java.nio.file.Paths;
import java.util.Collection;
@@ -44,16 +46,32 @@ import java.util.List;
import java.util.regex.Matcher;
/**
- * Implements IOChannelFactory for local files.
+ * Implements {@link IOChannelFactory} for local files.
*/
public class FileIOChannelFactory implements IOChannelFactory {
private static final Logger LOG = LoggerFactory.getLogger(FileIOChannelFactory.class);
+ /**
+ * Converts the given file spec to a java {@link File}. If {@code spec} is actually a URI with
+ * the {@code file} scheme, then this function will ensure that the returned {@link File}
+ * has the correct path.
+ */
+ private static File specToFile(String spec) {
+ try {
+ // Handle URI.
+ URI uri = URI.create(spec);
+ return Paths.get(uri).toFile();
+ } catch (IllegalArgumentException e) {
+ // Fall back to assuming this is actually a file.
+ return Paths.get(spec).toFile();
+ }
+ }
+
// This implementation only allows for wildcards in the file name.
// The directory portion must exist as-is.
@Override
public Collection<String> match(String spec) throws IOException {
- File file = new File(spec);
+ File file = specToFile(spec);
File parent = file.getAbsoluteFile().getParentFile();
if (!parent.exists()) {
@@ -95,7 +113,7 @@ public class FileIOChannelFactory implements IOChannelFactory {
public ReadableByteChannel open(String spec) throws IOException {
LOG.debug("opening file {}", spec);
@SuppressWarnings("resource") // The caller is responsible for closing the channel.
- FileInputStream inputStream = new FileInputStream(spec);
+ FileInputStream inputStream = new FileInputStream(specToFile(spec));
// Use this method for creating the channel (rather than new FileChannel) so that we get
// regular FileNotFoundException. Closing the underyling channel will close the inputStream.
return inputStream.getChannel();
@@ -105,7 +123,7 @@ public class FileIOChannelFactory implements IOChannelFactory {
public WritableByteChannel create(String spec, String mimeType)
throws IOException {
LOG.debug("creating file {}", spec);
- File file = new File(spec);
+ File file = specToFile(spec);
if (file.getAbsoluteFile().getParentFile() != null
&& !file.getAbsoluteFile().getParentFile().exists()
&& !file.getAbsoluteFile().getParentFile().mkdirs()
@@ -119,7 +137,7 @@ public class FileIOChannelFactory implements IOChannelFactory {
@Override
public long getSizeBytes(String spec) throws IOException {
try {
- return Files.size(FileSystems.getDefault().getPath(spec));
+ return Files.size(specToFile(spec).toPath());
} catch (NoSuchFileException e) {
throw new FileNotFoundException(e.getReason());
}
@@ -132,6 +150,7 @@ public class FileIOChannelFactory implements IOChannelFactory {
@Override
public String resolve(String path, String other) throws IOException {
- return Paths.get(path).resolve(other).toString();
+ Path p = specToFile(path).toPath();
+ return p.resolve(other).toString();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/38061d35/sdks/java/core/src/test/java/org/apache/beam/sdk/util/FileIOChannelFactoryTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/FileIOChannelFactoryTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/FileIOChannelFactoryTest.java
index b510408..79e6e5c 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/FileIOChannelFactoryTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/FileIOChannelFactoryTest.java
@@ -52,6 +52,7 @@ public class FileIOChannelFactoryTest {
private void testCreate(Path path) throws Exception {
String expected = "my test string";
+ // First with the path string
try (Writer writer = Channels.newWriter(
factory.create(path.toString(), MimeTypes.TEXT), StandardCharsets.UTF_8.name())) {
writer.write(expected);
@@ -59,6 +60,18 @@ public class FileIOChannelFactoryTest {
assertThat(
Files.readLines(path.toFile(), StandardCharsets.UTF_8),
containsInAnyOrder(expected));
+
+ // Delete the file before trying as URI
+ assertTrue("Unable to delete file " + path, path.toFile().delete());
+
+ // Second with the path URI
+ try (Writer writer = Channels.newWriter(
+ factory.create(path.toUri().toString(), MimeTypes.TEXT), StandardCharsets.UTF_8.name())) {
+ writer.write(expected);
+ }
+ assertThat(
+ Files.readLines(path.toFile(), StandardCharsets.UTF_8),
+ containsInAnyOrder(expected));
}
@Test
@@ -194,19 +207,24 @@ public class FileIOChannelFactoryTest {
@Test
public void testResolve() throws Exception {
- String expected = temporaryFolder.getRoot().toPath().resolve("aa").toString();
- assertEquals(expected, factory.resolve(temporaryFolder.getRoot().toString(), "aa"));
+ Path rootPath = temporaryFolder.getRoot().toPath();
+ String rootString = rootPath.toString();
+
+ String expected = rootPath.resolve("aa").toString();
+ assertEquals(expected, factory.resolve(rootString, "aa"));
+ assertEquals(expected, factory.resolve("file:" + rootString, "aa"));
+ assertEquals(expected, factory.resolve("file://" + rootString, "aa"));
}
@Test
public void testResolveOtherIsFullPath() throws Exception {
- String expected = temporaryFolder.getRoot().getPath().toString();
+ String expected = temporaryFolder.getRoot().getPath();
assertEquals(expected, factory.resolve(expected, expected));
}
@Test
public void testResolveOtherIsEmptyPath() throws Exception {
- String expected = temporaryFolder.getRoot().getPath().toString();
+ String expected = temporaryFolder.getRoot().getPath();
assertEquals(expected, factory.resolve(expected, ""));
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/38061d35/sdks/java/core/src/test/java/org/apache/beam/sdk/util/IOChannelUtilsTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/IOChannelUtilsTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/IOChannelUtilsTest.java
index 9168fd6..8a7eb02 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/IOChannelUtilsTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/IOChannelUtilsTest.java
@@ -17,7 +17,9 @@
*/
package org.apache.beam.sdk.util;
+import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
import com.google.common.io.Files;
@@ -80,10 +82,9 @@ public class IOChannelUtilsTest {
}
@Test
- public void testFilePrefix() throws Exception {
- IOChannelUtils.getFactory("file://tmp");
- IOChannelUtils.getFactory("file:/tmp");
- IOChannelUtils.getFactory("file:tmp");
+ public void testHandlerNoScheme() throws Exception {
+ String pathToTempFolder = tmpFolder.getRoot().getAbsolutePath();
+ assertThat(IOChannelUtils.getFactory(pathToTempFolder), instanceOf(FileIOChannelFactory.class));
}
@Test