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 2017/05/04 21:18:29 UTC
[1/2] beam git commit: [BEAM-59] Remove IOChannelUtils,
IOChannelFactory
Repository: beam
Updated Branches:
refs/heads/master 9fffa7efa -> 44589c6f9
[BEAM-59] Remove IOChannelUtils, IOChannelFactory
Leaving one deprecated function in one deprecated class
to enable migration of DataflowRunner.
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/78ae5287
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/78ae5287
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/78ae5287
Branch: refs/heads/master
Commit: 78ae52870dd906c63cc3574a6ec0e8e8fa8e510e
Parents: 9fffa7e
Author: Dan Halperin <dh...@google.com>
Authored: Thu May 4 12:27:44 2017 -0700
Committer: Dan Halperin <dh...@google.com>
Committed: Thu May 4 14:18:20 2017 -0700
----------------------------------------------------------------------
.../beam/sdk/util/FileIOChannelFactory.java | 223 ----------------
.../sdk/util/FileIOChannelFactoryRegistrar.java | 38 ---
.../apache/beam/sdk/util/IOChannelFactory.java | 127 ---------
.../sdk/util/IOChannelFactoryRegistrar.java | 48 ----
.../apache/beam/sdk/util/IOChannelUtils.java | 259 +------------------
.../util/FileIOChannelFactoryRegistrarTest.java | 44 ----
.../beam/sdk/util/FileIOChannelFactoryTest.java | 242 -----------------
.../beam/sdk/util/IOChannelUtilsTest.java | 118 ---------
.../beam/sdk/util/GcsIOChannelFactory.java | 111 --------
.../sdk/util/GcsIOChannelFactoryRegistrar.java | 38 ---
.../util/GcsIOChannelFactoryRegistrarTest.java | 44 ----
.../beam/sdk/util/GcsIOChannelFactoryTest.java | 43 ---
12 files changed, 3 insertions(+), 1332 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/78ae5287/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
deleted file mode 100644
index 6f6ba37..0000000
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/FileIOChannelFactory.java
+++ /dev/null
@@ -1,223 +0,0 @@
-/*
- * 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.beam.sdk.util;
-
-import static com.google.common.base.Preconditions.checkArgument;
-
-import com.google.common.base.Predicate;
-import com.google.common.base.Predicates;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
-import java.io.BufferedOutputStream;
-import java.io.File;
-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.nio.file.StandardCopyOption;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.regex.Matcher;
-import javax.annotation.Nullable;
-
-import org.apache.beam.sdk.options.PipelineOptions;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Implements {@link IOChannelFactory} for local files.
- */
-public class FileIOChannelFactory implements IOChannelFactory {
- private static final Logger LOG = LoggerFactory.getLogger(FileIOChannelFactory.class);
-
- /**
- * Create a {@link FileIOChannelFactory} with the given {@link PipelineOptions}.
- */
- public static FileIOChannelFactory fromOptions(@Nullable PipelineOptions options) {
- return new FileIOChannelFactory();
- }
-
- private FileIOChannelFactory() {}
-
- /**
- * 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();
- }
- }
-
- /**
- * {@inheritDoc}
- *
- * <p>Wildcards in the directory portion are not supported.
- */
- @Override
- public Collection<String> match(String spec) throws IOException {
- File file = specToFile(spec);
-
- File parent = file.getAbsoluteFile().getParentFile();
- if (!parent.exists()) {
- return Collections.EMPTY_LIST;
- }
-
- // Method getAbsolutePath() on Windows platform may return something like
- // "c:\temp\file.txt". FileSystem.getPathMatcher() call below will treat
- // '\' (backslash) as an escape character, instead of a directory
- // separator. Replacing backslash with double-backslash solves the problem.
- // We perform the replacement on all platforms, even those that allow
- // backslash as a part of the filename, because Globs.toRegexPattern will
- // eat one backslash.
- String pathToMatch = file.getAbsolutePath().replaceAll(Matcher.quoteReplacement("\\"),
- Matcher.quoteReplacement("\\\\"));
-
- final PathMatcher matcher = FileSystems.getDefault().getPathMatcher("glob:" + pathToMatch);
-
- Iterable<File> files = com.google.common.io.Files.fileTreeTraverser().preOrderTraversal(parent);
- Iterable<File> matchedFiles = Iterables.filter(files,
- Predicates.and(
- com.google.common.io.Files.isFile(),
- new Predicate<File>() {
- @Override
- public boolean apply(File input) {
- return matcher.matches(input.toPath());
- }
- }));
-
- List<String> result = new LinkedList<>();
- for (File match : matchedFiles) {
- result.add(match.getPath());
- }
-
- return result;
- }
-
- @Override
- 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(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();
- }
-
- @Override
- public WritableByteChannel create(String spec, String mimeType)
- throws IOException {
- LOG.debug("creating file {}", spec);
- File file = specToFile(spec);
- if (file.getAbsoluteFile().getParentFile() != null
- && !file.getAbsoluteFile().getParentFile().exists()
- && !file.getAbsoluteFile().getParentFile().mkdirs()
- && !file.getAbsoluteFile().getParentFile().exists()) {
- throw new IOException("Unable to create parent directories for '" + spec + "'");
- }
- return Channels.newChannel(
- new BufferedOutputStream(new FileOutputStream(file)));
- }
-
- @Override
- public long getSizeBytes(String spec) throws IOException {
- try {
- return Files.size(specToFile(spec).toPath());
- } catch (NoSuchFileException e) {
- throw new FileNotFoundException(e.getReason());
- }
- }
-
- @Override
- public boolean isReadSeekEfficient(String spec) throws IOException {
- return true;
- }
-
- @Override
- public String resolve(String path, String other) throws IOException {
- return toPath(path).resolve(other).toString();
- }
-
- @Override
- public Path toPath(String path) {
- return specToFile(path).toPath();
- }
-
- @Override
- public void copy(Iterable<String> srcFilenames, Iterable<String> destFilenames) throws
- IOException {
- List<String> srcList = Lists.newArrayList(srcFilenames);
- List<String> destList = Lists.newArrayList(destFilenames);
- checkArgument(
- srcList.size() == destList.size(),
- "Number of source files %s must equal number of destination files %s",
- srcList.size(),
- destList.size());
- int numFiles = srcList.size();
- for (int i = 0; i < numFiles; i++) {
- String src = srcList.get(i);
- String dst = destList.get(i);
- LOG.info("Copying {} to {}", src, dst);
- try {
- // Copy the source file, replacing the existing destination.
- // Paths.get(x) will not work on Windows OSes cause of the ":" after the drive letter.
- Files.copy(
- new File(src).toPath(),
- new File(dst).toPath(),
- StandardCopyOption.REPLACE_EXISTING);
- } catch (NoSuchFileException e) {
- LOG.info("{} does not exist.", src);
- // Suppress exception if file does not exist.
- }
- }
- }
-
- @Override
- public void remove(Collection<String> filesOrDirs) throws IOException {
- for (String fileOrDir : filesOrDirs) {
- LOG.debug("Removing file {}", fileOrDir);
- removeOne(fileOrDir);
- }
- }
-
- private void removeOne(String fileOrDir) throws IOException {
- // Delete the file if it exists.
- boolean exists = Files.deleteIfExists(Paths.get(fileOrDir));
- if (!exists) {
- LOG.debug("Tried to delete {}, but it did not exist", fileOrDir);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/beam/blob/78ae5287/sdks/java/core/src/main/java/org/apache/beam/sdk/util/FileIOChannelFactoryRegistrar.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/FileIOChannelFactoryRegistrar.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/FileIOChannelFactoryRegistrar.java
deleted file mode 100644
index acc0222..0000000
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/FileIOChannelFactoryRegistrar.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * 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.beam.sdk.util;
-
-import com.google.auto.service.AutoService;
-import org.apache.beam.sdk.options.PipelineOptions;
-
-/**
- * {@link AutoService} registrar for the {@link FileIOChannelFactory}.
- */
-@AutoService(IOChannelFactoryRegistrar.class)
-public class FileIOChannelFactoryRegistrar implements IOChannelFactoryRegistrar {
-
- @Override
- public IOChannelFactory fromOptions(PipelineOptions options) {
- return FileIOChannelFactory.fromOptions(options);
- }
-
- @Override
- public String getScheme() {
- return "file";
- }
-}
http://git-wip-us.apache.org/repos/asf/beam/blob/78ae5287/sdks/java/core/src/main/java/org/apache/beam/sdk/util/IOChannelFactory.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/IOChannelFactory.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/IOChannelFactory.java
deleted file mode 100644
index 3a3af17..0000000
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/IOChannelFactory.java
+++ /dev/null
@@ -1,127 +0,0 @@
-/*
- * 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.beam.sdk.util;
-
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.nio.channels.ReadableByteChannel;
-import java.nio.channels.WritableByteChannel;
-import java.nio.file.Path;
-import java.util.Collection;
-
-/**
- * Defines a factory for working with read and write channels.
- *
- * <p>Channels provide an abstract API for IO operations.
- *
- * <p>See <a href="http://docs.oracle.com/javase/7/docs/api/java/nio/channels/package-summary.html"
- * >Java NIO Channels</a>
- *
- * @deprecated This is under redesign, see: https://issues.apache.org/jira/browse/BEAM-59.
- */
-@Deprecated
-public interface IOChannelFactory {
-
- /**
- * Matches a specification, which may contain globs, against available
- * resources.
- *
- * <p>Glob handling is dependent on the implementation. Implementations should
- * all support globs in the final component of a path (eg /foo/bar/*.txt),
- * however they are not required to support globs in the directory paths.
- *
- * <p>The list of resources returned are required to exist and not represent abstract
- * resources such as symlinks and directories.
- */
- Collection<String> match(String spec) throws IOException;
-
- /**
- * Returns a read channel for the given specification.
- *
- * <p>The specification is not expanded; it is used verbatim.
- *
- * <p>If seeking is supported, then this returns a
- * {@link java.nio.channels.SeekableByteChannel}.
- */
- ReadableByteChannel open(String spec) throws IOException;
-
- /**
- * Returns a write channel for the given specification.
- *
- * <p>The specification is not expanded; is it used verbatim.
- */
- WritableByteChannel create(String spec, String mimeType) throws IOException;
-
- /**
- * Returns the size in bytes for the given specification.
- *
- * <p>The specification is not expanded; it is used verbatim.
- *
- * <p>{@link FileNotFoundException} will be thrown if the resource does not exist.
- */
- long getSizeBytes(String spec) throws IOException;
-
- /**
- * Returns {@code true} if the channel created when invoking method {@link #open} for the given
- * file specification is guaranteed to be of type {@link java.nio.channels.SeekableByteChannel
- * SeekableByteChannel} and if seeking into positions of the channel is recommended. Returns
- * {@code false} if the channel returned is not a {@code SeekableByteChannel}. May return
- * {@code false} even if the channel returned is a {@code SeekableByteChannel}, if seeking is not
- * efficient for the given file specification.
- *
- * <p>Only efficiently seekable files can be split into offset ranges.
- *
- * <p>The specification is not expanded; it is used verbatim.
- */
- boolean isReadSeekEfficient(String spec) throws IOException;
-
- /**
- * Resolve the given {@code other} against the {@code path}.
- *
- * <p>If the {@code other} parameter is an absolute path then this method trivially returns
- * other. If {@code other} is an empty path then this method trivially returns the given
- * {@code path}. Otherwise this method considers the given {@code path} to be a directory and
- * resolves the {@code other} path against this path. In the simplest case, the {@code other}
- * path does not have a root component, in which case this method joins the {@code other} path
- * to the given {@code path} and returns a resulting path that ends with the {@code other} path.
- * Where the {@code other} path has a root component then resolution is highly implementation
- * dependent and therefore unspecified.
- */
- String resolve(String path, String other) throws IOException;
-
- /** Converts the given string to a {@link Path}. */
- Path toPath(String path);
-
- /**
- * Copies a collection of files from one location to another.
- *
- * <p>The number of source filenames must equal the number of destination filenames.
- *
- * @param srcFilenames the source filenames.
- * @param destFilenames the destination filenames.
- */
- void copy(Iterable<String> srcFilenames, Iterable<String> destFilenames) throws IOException;
-
- /**
- * Removes a collection of files or directories.
- *
- * <p>Directories are required to be empty. Non-empty directories will not be deleted,
- * and this method may return silently or throw an exception.
- */
- void remove(Collection<String> filesOrDirs) throws IOException;
-}
http://git-wip-us.apache.org/repos/asf/beam/blob/78ae5287/sdks/java/core/src/main/java/org/apache/beam/sdk/util/IOChannelFactoryRegistrar.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/IOChannelFactoryRegistrar.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/IOChannelFactoryRegistrar.java
deleted file mode 100644
index 7776b13..0000000
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/IOChannelFactoryRegistrar.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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.beam.sdk.util;
-
-import com.google.auto.service.AutoService;
-import java.util.ServiceLoader;
-import org.apache.beam.sdk.options.PipelineOptions;
-
-/**
- * A registrar that creates {@link IOChannelFactory} instances from {@link PipelineOptions}.
- *
- * <p>{@link IOChannelFactory} creators have the ability to provide a registrar by creating
- * a {@link ServiceLoader} entry and a concrete implementation of this interface.
- *
- * <p>It is optional but recommended to use one of the many build time tools such as
- * {@link AutoService} to generate the necessary META-INF files automatically.
- */
-public interface IOChannelFactoryRegistrar {
- /**
- * Create a {@link IOChannelFactory} from the given {@link PipelineOptions}.
- */
- IOChannelFactory fromOptions(PipelineOptions options);
-
- /**
- * Get the URI scheme which defines the namespace of the IOChannelFactoryRegistrar.
- *
- * <p>The scheme is required to be unique among all
- * {@link IOChannelFactoryRegistrar IOChannelFactoryRegistrars}.
- *
- * @see <a href="https://www.ietf.org/rfc/rfc2396.txt">RFC 2396</a>
- */
- String getScheme();
-}
http://git-wip-us.apache.org/repos/asf/beam/blob/78ae5287/sdks/java/core/src/main/java/org/apache/beam/sdk/util/IOChannelUtils.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/IOChannelUtils.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/IOChannelUtils.java
index 33913f8..b658983 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/IOChannelUtils.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/IOChannelUtils.java
@@ -17,264 +17,11 @@
*/
package org.apache.beam.sdk.util;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Function;
-import com.google.common.base.Joiner;
-import com.google.common.collect.FluentIterable;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Multimap;
-import com.google.common.collect.Ordering;
-import com.google.common.collect.Sets;
-import com.google.common.collect.TreeMultimap;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.nio.channels.ReadableByteChannel;
-import java.nio.channels.WritableByteChannel;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.ServiceLoader;
-import java.util.Set;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-import javax.annotation.Nonnull;
-import org.apache.beam.sdk.io.DefaultFilenamePolicy;
import org.apache.beam.sdk.options.PipelineOptions;
-import org.apache.beam.sdk.util.common.ReflectHelpers;
-/**
- * Provides utilities for creating read and write channels.
- */
+/** Do not use, being removed. */
+@Deprecated
public class IOChannelUtils {
- // TODO: add registration mechanism for adding new schemas.
- private static final Map<String, IOChannelFactory> FACTORY_MAP =
- Collections.synchronizedMap(new HashMap<String, IOChannelFactory>());
-
- private static final ClassLoader CLASS_LOADER = ReflectHelpers.findClassLoader();
-
- /**
- * Associates a scheme with an {@link IOChannelFactory}.
- *
- * <p>The given factory is used to construct read and write channels when
- * a URI is provided with the given scheme.
- *
- * <p>For example, when reading from "gs://bucket/path", the scheme "gs" is
- * used to lookup the appropriate factory.
- *
- * <p>{@link PipelineOptions} are required to provide dependencies and
- * pipeline level configuration to the individual {@link IOChannelFactory IOChannelFactories}.
- *
- * @throws IllegalStateException if multiple {@link IOChannelFactory IOChannelFactories}
- * for the same scheme are detected.
- */
- @VisibleForTesting
- public static void setIOFactoryInternal(
- String scheme,
- IOChannelFactory factory,
- boolean override) {
- if (!override && FACTORY_MAP.containsKey(scheme)) {
- throw new IllegalStateException(String.format(
- "Failed to register IOChannelFactory: %s. "
- + "Scheme: [%s] is already registered with %s, and override is not allowed.",
- FACTORY_MAP.get(scheme).getClass(),
- scheme,
- factory.getClass()));
- }
- FACTORY_MAP.put(scheme, factory);
- }
-
- /**
- * Deregisters the scheme and the associated {@link IOChannelFactory}.
- */
- @VisibleForTesting
- static void deregisterScheme(String scheme) {
- FACTORY_MAP.remove(scheme);
- }
-
- /**
- * Registers all {@link IOChannelFactory IOChannelFactories} from {@link ServiceLoader}.
- *
- * <p>{@link PipelineOptions} are required to provide dependencies and
- * pipeline level configuration to the individual {@link IOChannelFactory IOChannelFactories}.
- *
- * <p>Multiple {@link IOChannelFactory IOChannelFactories} for the same scheme are not allowed.
- *
- * @throws IllegalStateException if multiple {@link IOChannelFactory IOChannelFactories}
- * for the same scheme are detected.
- */
- public static void registerIOFactories(PipelineOptions options) {
- registerIOFactoriesInternal(options, false /* override */);
- }
-
- /**
- * Registers all {@link IOChannelFactory IOChannelFactories} from {@link ServiceLoader}.
- *
- * <p>This requires {@link PipelineOptions} to provide, e.g., credentials for GCS.
- *
- * <p>Override existing schemes is allowed.
- *
- * @deprecated This is currently to provide different configurations for tests and
- * is still public for IOChannelFactory redesign purposes.
- */
@Deprecated
- @VisibleForTesting
- public static void registerIOFactoriesAllowOverride(PipelineOptions options) {
- registerIOFactoriesInternal(options, true /* override */);
- }
-
- private static void registerIOFactoriesInternal(
- PipelineOptions options, boolean override) {
- Set<IOChannelFactoryRegistrar> registrars =
- Sets.newTreeSet(ReflectHelpers.ObjectsClassComparator.INSTANCE);
- registrars.addAll(Lists.newArrayList(
- ServiceLoader.load(IOChannelFactoryRegistrar.class, CLASS_LOADER)));
-
- checkDuplicateScheme(registrars);
-
- for (IOChannelFactoryRegistrar registrar : registrars) {
- setIOFactoryInternal(
- registrar.getScheme(),
- registrar.fromOptions(options),
- override);
- }
- }
-
- @VisibleForTesting
- static void checkDuplicateScheme(Set<IOChannelFactoryRegistrar> registrars) {
- Multimap<String, IOChannelFactoryRegistrar> registrarsBySchemes =
- TreeMultimap.create(Ordering.<String>natural(), Ordering.arbitrary());
-
- for (IOChannelFactoryRegistrar registrar : registrars) {
- registrarsBySchemes.put(registrar.getScheme(), registrar);
- }
- for (Entry<String, Collection<IOChannelFactoryRegistrar>> entry
- : registrarsBySchemes.asMap().entrySet()) {
- if (entry.getValue().size() > 1) {
- String conflictingRegistrars = Joiner.on(", ").join(
- FluentIterable.from(entry.getValue())
- .transform(new Function<IOChannelFactoryRegistrar, String>() {
- @Override
- public String apply(@Nonnull IOChannelFactoryRegistrar input) {
- return input.getClass().getName();
- }})
- .toSortedList(Ordering.<String>natural()));
- throw new IllegalStateException(String.format(
- "Scheme: [%s] has conflicting registrars: [%s]",
- entry.getKey(),
- conflictingRegistrars));
- }
- }
- }
-
- /**
- * Creates a write channel for the given filename.
- */
- public static WritableByteChannel create(String filename, String mimeType)
- throws IOException {
- return getFactory(filename).create(filename, mimeType);
- }
-
- /**
- * Creates a read channel for the given filename.
- */
- public static ReadableByteChannel open(String filename)
- throws IOException {
- return getFactory(filename).open(filename);
- }
-
- /**
- * Creates a write channel for the given file components.
- *
- * <p>If numShards is specified, then a ShardingWritableByteChannel is
- * returned.
- *
- * <p>Shard numbers are 0 based, meaning they start with 0 and end at the
- * number of shards - 1.
- */
- public static WritableByteChannel create(String prefix, String shardTemplate,
- String suffix, int numShards, String mimeType) throws IOException {
- if (numShards == 1) {
- return create(DefaultFilenamePolicy.constructName(prefix, shardTemplate, suffix, 0, 1),
- mimeType);
- }
-
- // It is the callers responsibility to close this channel.
- @SuppressWarnings("resource")
- ShardingWritableByteChannel shardingChannel =
- new ShardingWritableByteChannel();
-
- Set<String> outputNames = new HashSet<>();
- for (int i = 0; i < numShards; i++) {
- String outputName =
- DefaultFilenamePolicy.constructName(prefix, shardTemplate, suffix, i, numShards);
- if (!outputNames.add(outputName)) {
- throw new IllegalArgumentException(
- "Shard name collision detected for: " + outputName);
- }
- WritableByteChannel channel = create(outputName, mimeType);
- shardingChannel.addChannel(channel);
- }
-
- return shardingChannel;
- }
-
- /**
- * Returns the size in bytes for the given specification.
- *
- * <p>The specification is not expanded; it is used verbatim.
- *
- * <p>{@link FileNotFoundException} will be thrown if the resource does not exist.
- */
- public static long getSizeBytes(String spec) throws IOException {
- return getFactory(spec).getSizeBytes(spec);
- }
-
- private static final Pattern URI_SCHEME_PATTERN = Pattern.compile(
- "(?<scheme>[a-zA-Z][-a-zA-Z0-9+.]*)://.*");
-
- /**
- * Returns the IOChannelFactory associated with an input specification.
- */
- public static IOChannelFactory getFactory(String spec) throws IOException {
- // The spec is almost, but not quite, a URI. In particular,
- // the reserved characters '[', ']', and '?' have meanings that differ
- // from their use in the URI spec. ('*' is not reserved).
- // Here, we just need the scheme, which is so circumscribed as to be
- // very easy to extract with a regex.
- Matcher matcher = URI_SCHEME_PATTERN.matcher(spec);
-
- if (!matcher.matches()) {
- return FileIOChannelFactory.fromOptions(null);
- }
-
- String scheme = matcher.group("scheme");
- IOChannelFactory ioFactory = FACTORY_MAP.get(scheme);
- if (ioFactory != null) {
- return ioFactory;
- }
-
- throw new IOException("Unable to find handler for " + spec);
- }
-
- /**
- * Resolve multiple {@code others} against the {@code path} sequentially.
- *
- * <p>Empty paths in {@code others} are ignored. If {@code others} contains one or more
- * absolute paths, then this method returns a path that starts with the last absolute path
- * in {@code others} joined with the remaining paths. Resolution of paths is highly
- * implementation dependent and therefore unspecified.
- */
- public static String resolve(String path, String... others) throws IOException {
- IOChannelFactory ioFactory = getFactory(path);
- String fullPath = path;
-
- for (String other : others) {
- fullPath = ioFactory.resolve(fullPath, other);
- }
-
- return fullPath;
- }
+ public static void registerIOFactoriesAllowOverride(PipelineOptions options) {}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/78ae5287/sdks/java/core/src/test/java/org/apache/beam/sdk/util/FileIOChannelFactoryRegistrarTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/FileIOChannelFactoryRegistrarTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/FileIOChannelFactoryRegistrarTest.java
deleted file mode 100644
index f8f53e7..0000000
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/FileIOChannelFactoryRegistrarTest.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * 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.beam.sdk.util;
-
-import static org.junit.Assert.fail;
-
-import com.google.common.collect.Lists;
-import java.util.ServiceLoader;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
-
-/**
- * Tests for {@link FileIOChannelFactoryRegistrar}.
- */
-@RunWith(JUnit4.class)
-public class FileIOChannelFactoryRegistrarTest {
-
- @Test
- public void testServiceLoader() {
- for (IOChannelFactoryRegistrar registrar
- : Lists.newArrayList(ServiceLoader.load(IOChannelFactoryRegistrar.class).iterator())) {
- if (registrar instanceof FileIOChannelFactoryRegistrar) {
- return;
- }
- }
- fail("Expected to find " + FileIOChannelFactoryRegistrar.class);
- }
-}
http://git-wip-us.apache.org/repos/asf/beam/blob/78ae5287/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
deleted file mode 100644
index 6062619..0000000
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/FileIOChannelFactoryTest.java
+++ /dev/null
@@ -1,242 +0,0 @@
-/*
- * 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.beam.sdk.util;
-
-import static org.hamcrest.Matchers.containsInAnyOrder;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
-
-import com.google.common.collect.ImmutableList;
-import com.google.common.io.Files;
-import com.google.common.io.LineReader;
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.Reader;
-import java.io.Writer;
-import java.nio.channels.Channels;
-import java.nio.charset.StandardCharsets;
-import java.nio.file.Path;
-import java.util.List;
-import org.hamcrest.Matchers;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
-import org.junit.rules.TemporaryFolder;
-import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
-
-/** Tests for {@link FileIOChannelFactory}. */
-@RunWith(JUnit4.class)
-public class FileIOChannelFactoryTest {
- @Rule public ExpectedException thrown = ExpectedException.none();
- @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder();
- private FileIOChannelFactory factory = FileIOChannelFactory.fromOptions(null);
-
- 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);
- }
- 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
- public void testCreateWithExistingFile() throws Exception {
- File existingFile = temporaryFolder.newFile();
- testCreate(existingFile.toPath());
- }
-
- @Test
- public void testCreateWithinExistingDirectory() throws Exception {
- testCreate(temporaryFolder.getRoot().toPath().resolve("file.txt"));
- }
-
- @Test
- public void testCreateWithNonExistentSubDirectory() throws Exception {
- testCreate(temporaryFolder.getRoot().toPath().resolve("non-existent-dir").resolve("file.txt"));
- }
-
- @Test
- public void testReadWithExistingFile() throws Exception {
- String expected = "my test string";
- File existingFile = temporaryFolder.newFile();
- Files.write(expected, existingFile, StandardCharsets.UTF_8);
- String data;
- try (Reader reader =
- Channels.newReader(factory.open(existingFile.getPath()), StandardCharsets.UTF_8.name())) {
- data = new LineReader(reader).readLine();
- }
- assertEquals(expected, data);
- }
-
- @Test
- public void testReadNonExistentFile() throws Exception {
- thrown.expect(FileNotFoundException.class);
- factory
- .open(
- temporaryFolder
- .getRoot()
- .toPath()
- .resolve("non-existent-file.txt")
- .toString())
- .close();
- }
-
- @Test
- public void testIsReadSeekEfficient() throws Exception {
- assertTrue(factory.isReadSeekEfficient("somePath"));
- }
-
- @Test
- public void testMatchExact() throws Exception {
- List<String> expected = ImmutableList.of(temporaryFolder.newFile("a").toString());
- temporaryFolder.newFile("aa");
- temporaryFolder.newFile("ab");
-
- assertThat(factory.match(temporaryFolder.getRoot().toPath().resolve("a").toString()),
- containsInAnyOrder(expected.toArray(new String[expected.size()])));
- }
-
- @Test
- public void testMatchPatternNone() throws Exception {
- List<String> expected = ImmutableList.of();
- temporaryFolder.newFile("a");
- temporaryFolder.newFile("aa");
- temporaryFolder.newFile("ab");
-
- // Windows doesn't like resolving paths with * in them, so the * is appended after resolve.
- assertThat(factory.match(factory.resolve(temporaryFolder.getRoot().getPath(), "b") + "*"),
- containsInAnyOrder(expected.toArray(new String[expected.size()])));
- }
-
- @Test
- public void testMatchForNonExistentFile() throws Exception {
- List<String> expected = ImmutableList.of();
- temporaryFolder.newFile("aa");
-
- assertThat(factory.match(factory.resolve(temporaryFolder.getRoot().getPath(), "a")),
- containsInAnyOrder(expected.toArray(new String[expected.size()])));
- }
-
- @Test
- public void testMatchMultipleWithoutSubdirectoryExpansion() throws Exception {
- File unmatchedSubDir = temporaryFolder.newFolder("aaa");
- File unmatchedSubDirFile = File.createTempFile("sub-dir-file", "", unmatchedSubDir);
- unmatchedSubDirFile.deleteOnExit();
- List<String> expected = ImmutableList.of(temporaryFolder.newFile("a").toString(),
- temporaryFolder.newFile("aa").toString(), temporaryFolder.newFile("ab").toString());
- temporaryFolder.newFile("ba");
- temporaryFolder.newFile("bb");
-
- // Windows doesn't like resolving paths with * in them, so the * is appended after resolve.
- assertThat(factory.match(factory.resolve(temporaryFolder.getRoot().getPath(), "a") + "*"),
- containsInAnyOrder(expected.toArray(new String[expected.size()])));
- }
-
- @Test
- public void testMatchMultipleWithSubdirectoryExpansion() throws Exception {
- File matchedSubDir = temporaryFolder.newFolder("a");
- File matchedSubDirFile = File.createTempFile("sub-dir-file", "", matchedSubDir);
- matchedSubDirFile.deleteOnExit();
- File unmatchedSubDir = temporaryFolder.newFolder("b");
- File unmatchedSubDirFile = File.createTempFile("sub-dir-file", "", unmatchedSubDir);
- unmatchedSubDirFile.deleteOnExit();
-
- List<String> expected = ImmutableList.of(matchedSubDirFile.toString(),
- temporaryFolder.newFile("aa").toString(), temporaryFolder.newFile("ab").toString());
- temporaryFolder.newFile("ba");
- temporaryFolder.newFile("bb");
-
- // Windows doesn't like resolving paths with * in them, so the ** is appended after resolve.
- assertThat(factory.match(factory.resolve(temporaryFolder.getRoot().getPath(), "a") + "**"),
- Matchers.hasItems(expected.toArray(new String[expected.size()])));
- }
-
- @Test
- public void testMatchWithDirectoryFiltersOutDirectory() throws Exception {
- List<String> expected = ImmutableList.of(temporaryFolder.newFile("a").toString());
- temporaryFolder.newFolder("a_dir_that_should_not_be_matched");
-
- // Windows doesn't like resolving paths with * in them, so the * is appended after resolve.
- assertThat(factory.match(factory.resolve(temporaryFolder.getRoot().getPath(), "a") + "*"),
- containsInAnyOrder(expected.toArray(new String[expected.size()])));
- }
-
- @Test
- public void testMatchWithoutParentDirectory() throws Exception {
- String pattern = factory.resolve(
- factory.resolve(temporaryFolder.getRoot().getPath(), "non_existing_dir"),
- "*");
- assertTrue(factory.match(pattern).isEmpty());
- }
-
- @Test
- public void testResolve() throws Exception {
- 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();
- assertEquals(expected, factory.resolve(expected, expected));
- }
-
- @Test
- public void testResolveOtherIsEmptyPath() throws Exception {
- String expected = temporaryFolder.getRoot().getPath();
- assertEquals(expected, factory.resolve(expected, ""));
- }
-
- @Test
- public void testGetSizeBytes() throws Exception {
- String data = "TestData!!!";
- File file = temporaryFolder.newFile();
- Files.write(data, file, StandardCharsets.UTF_8);
- assertEquals(data.length(), factory.getSizeBytes(file.getPath()));
- }
-
- @Test
- public void testGetSizeBytesForNonExistentFile() throws Exception {
- thrown.expect(FileNotFoundException.class);
- factory.getSizeBytes(
- factory.resolve(temporaryFolder.getRoot().getPath(), "non-existent-file"));
- }
-}
http://git-wip-us.apache.org/repos/asf/beam/blob/78ae5287/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
deleted file mode 100644
index ea4ae87..0000000
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/IOChannelUtilsTest.java
+++ /dev/null
@@ -1,118 +0,0 @@
-/*
- * 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.beam.sdk.util;
-
-import static org.hamcrest.Matchers.instanceOf;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.fail;
-
-import com.google.common.collect.Sets;
-import com.google.common.io.Files;
-import java.io.File;
-import java.nio.charset.StandardCharsets;
-import org.apache.beam.sdk.options.PipelineOptionsFactory;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
-import org.junit.rules.TemporaryFolder;
-import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
-
-/**
- * Tests for IOChannelUtils.
- */
-@RunWith(JUnit4.class)
-public class IOChannelUtilsTest {
- @Rule
- public TemporaryFolder tmpFolder = new TemporaryFolder();
-
- @Rule
- public ExpectedException thrown = ExpectedException.none();
-
- @Test(expected = IllegalArgumentException.class)
- public void testShardNameCollision() throws Exception {
- File outFolder = tmpFolder.newFolder();
- String filename = outFolder.toPath().resolve("output").toString();
-
- IOChannelUtils.create(filename, "", "", 2, "text").close();
- fail("IOChannelUtils.create expected to fail due "
- + "to filename collision");
- }
-
- @Test
- public void testHandlerNoScheme() throws Exception {
- String pathToTempFolder = tmpFolder.getRoot().getAbsolutePath();
- assertThat(IOChannelUtils.getFactory(pathToTempFolder), instanceOf(FileIOChannelFactory.class));
- }
-
- @Test
- public void testGetSizeBytes() throws Exception {
- String data = "TestData";
- File file = tmpFolder.newFile();
- Files.write(data, file, StandardCharsets.UTF_8);
- assertEquals(data.length(), IOChannelUtils.getSizeBytes(file.getPath()));
- }
-
- @Test
- public void testResolveSinglePath() throws Exception {
- String expected = tmpFolder.getRoot().toPath().resolve("aa").toString();
- assertEquals(expected, IOChannelUtils.resolve(tmpFolder.getRoot().toString(), "aa"));
- }
-
- @Test
- public void testResolveMultiplePaths() throws Exception {
- String expected =
- tmpFolder.getRoot().toPath().resolve("aa").resolve("bb").resolve("cc").toString();
- assertEquals(expected,
- IOChannelUtils.resolve(tmpFolder.getRoot().getPath(), "aa", "bb", "cc"));
- }
-
- @Test
- public void testRegisterIOFactoriesAllowOverride() throws Exception {
- IOChannelUtils.registerIOFactoriesAllowOverride(PipelineOptionsFactory.create());
- IOChannelUtils.registerIOFactoriesAllowOverride(PipelineOptionsFactory.create());
- assertNotNull(IOChannelUtils.getFactory("gs"));
- assertNotNull(IOChannelUtils.getFactory("file"));
- }
-
- @Test
- public void testRegisterIOFactories() throws Exception {
- IOChannelUtils.deregisterScheme("gs");
- IOChannelUtils.deregisterScheme("file");
-
- IOChannelUtils.registerIOFactories(PipelineOptionsFactory.create());
- assertNotNull(IOChannelUtils.getFactory("gs"));
- assertNotNull(IOChannelUtils.getFactory("file"));
- thrown.expect(RuntimeException.class);
- thrown.expectMessage("Failed to register IOChannelFactory");
- thrown.expectMessage("override is not allowed");
- IOChannelUtils.registerIOFactories(PipelineOptionsFactory.create());
- }
-
- @Test
- public void testCheckDuplicateScheme() throws Exception {
- thrown.expect(RuntimeException.class);
- thrown.expectMessage("Scheme: [file] has conflicting registrars");
- IOChannelUtils.checkDuplicateScheme(
- Sets.<IOChannelFactoryRegistrar>newHashSet(
- new FileIOChannelFactoryRegistrar(),
- new FileIOChannelFactoryRegistrar()));
- }
-}
http://git-wip-us.apache.org/repos/asf/beam/blob/78ae5287/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/util/GcsIOChannelFactory.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/util/GcsIOChannelFactory.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/util/GcsIOChannelFactory.java
deleted file mode 100644
index 3a12620..0000000
--- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/util/GcsIOChannelFactory.java
+++ /dev/null
@@ -1,111 +0,0 @@
-/*
- * 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.beam.sdk.util;
-
-import java.io.IOException;
-import java.nio.channels.ReadableByteChannel;
-import java.nio.channels.WritableByteChannel;
-import java.nio.file.Path;
-import java.util.Collection;
-import java.util.LinkedList;
-import java.util.List;
-import org.apache.beam.sdk.extensions.gcp.options.GcsOptions;
-import org.apache.beam.sdk.options.PipelineOptions;
-import org.apache.beam.sdk.util.gcsfs.GcsPath;
-
-/**
- * Implements IOChannelFactory for GCS.
- */
-public class GcsIOChannelFactory implements IOChannelFactory {
-
- /**
- * Create a {@link GcsIOChannelFactory} with the given {@link PipelineOptions}.
- */
- public static GcsIOChannelFactory fromOptions(PipelineOptions options) {
- return new GcsIOChannelFactory(options.as(GcsOptions.class));
- }
-
- private final GcsOptions options;
-
- private GcsIOChannelFactory(GcsOptions options) {
- this.options = options;
- }
-
- @Override
- public Collection<String> match(String spec) throws IOException {
- GcsPath path = GcsPath.fromUri(spec);
- GcsUtil util = options.getGcsUtil();
- List<GcsPath> matched = util.expand(path);
-
- List<String> specs = new LinkedList<>();
- for (GcsPath match : matched) {
- specs.add(match.toString());
- }
-
- return specs;
- }
-
- @Override
- public ReadableByteChannel open(String spec) throws IOException {
- GcsPath path = GcsPath.fromUri(spec);
- GcsUtil util = options.getGcsUtil();
- return util.open(path);
- }
-
- @Override
- public WritableByteChannel create(String spec, String mimeType)
- throws IOException {
- GcsPath path = GcsPath.fromUri(spec);
- GcsUtil util = options.getGcsUtil();
- return util.create(path, mimeType);
- }
-
- @Override
- public long getSizeBytes(String spec) throws IOException {
- GcsPath path = GcsPath.fromUri(spec);
- GcsUtil util = options.getGcsUtil();
- return util.fileSize(path);
- }
-
- @Override
- public boolean isReadSeekEfficient(String spec) throws IOException {
- // TODO It is incorrect to return true here for files with content encoding set to gzip.
- return true;
- }
-
- @Override
- public String resolve(String path, String other) throws IOException {
- return toPath(path).resolve(other).toString();
- }
-
- @Override
- public Path toPath(String path) {
- return GcsPath.fromUri(path);
- }
-
- @Override
- public void copy(Iterable<String> srcFilenames, Iterable<String> destFilenames)
- throws IOException {
- options.getGcsUtil().copy(srcFilenames, destFilenames);
- }
-
- @Override
- public void remove(Collection<String> filesOrDirs) throws IOException {
- options.getGcsUtil().remove(filesOrDirs);
- }
-}
http://git-wip-us.apache.org/repos/asf/beam/blob/78ae5287/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/util/GcsIOChannelFactoryRegistrar.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/util/GcsIOChannelFactoryRegistrar.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/util/GcsIOChannelFactoryRegistrar.java
deleted file mode 100644
index b4c457f..0000000
--- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/util/GcsIOChannelFactoryRegistrar.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * 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.beam.sdk.util;
-
-import com.google.auto.service.AutoService;
-import org.apache.beam.sdk.options.PipelineOptions;
-
-/**
- * {@link AutoService} registrar for the {@link GcsIOChannelFactory}.
- */
-@AutoService(IOChannelFactoryRegistrar.class)
-public class GcsIOChannelFactoryRegistrar implements IOChannelFactoryRegistrar {
-
- @Override
- public GcsIOChannelFactory fromOptions(PipelineOptions options) {
- return GcsIOChannelFactory.fromOptions(options);
- }
-
- @Override
- public String getScheme() {
- return "gs";
- }
-}
http://git-wip-us.apache.org/repos/asf/beam/blob/78ae5287/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/util/GcsIOChannelFactoryRegistrarTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/util/GcsIOChannelFactoryRegistrarTest.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/util/GcsIOChannelFactoryRegistrarTest.java
deleted file mode 100644
index a29dd45..0000000
--- a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/util/GcsIOChannelFactoryRegistrarTest.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * 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.beam.sdk.util;
-
-import static org.junit.Assert.fail;
-
-import com.google.common.collect.Lists;
-import java.util.ServiceLoader;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
-
-/**
- * Tests for {@link GcsIOChannelFactoryRegistrar}.
- */
-@RunWith(JUnit4.class)
-public class GcsIOChannelFactoryRegistrarTest {
-
- @Test
- public void testServiceLoader() {
- for (IOChannelFactoryRegistrar registrar
- : Lists.newArrayList(ServiceLoader.load(IOChannelFactoryRegistrar.class).iterator())) {
- if (registrar instanceof GcsIOChannelFactoryRegistrar) {
- return;
- }
- }
- fail("Expected to find " + GcsIOChannelFactoryRegistrar.class);
- }
-}
http://git-wip-us.apache.org/repos/asf/beam/blob/78ae5287/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/util/GcsIOChannelFactoryTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/util/GcsIOChannelFactoryTest.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/util/GcsIOChannelFactoryTest.java
deleted file mode 100644
index f53490a..0000000
--- a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/util/GcsIOChannelFactoryTest.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * 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.beam.sdk.util;
-
-import static org.junit.Assert.assertEquals;
-
-import org.apache.beam.sdk.extensions.gcp.options.GcsOptions;
-import org.apache.beam.sdk.options.PipelineOptionsFactory;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
-
-/** Tests for {@link GcsIOChannelFactoryTest}. */
-@RunWith(JUnit4.class)
-public class GcsIOChannelFactoryTest {
- private GcsIOChannelFactory factory;
-
- @Before
- public void setUp() {
- factory = GcsIOChannelFactory.fromOptions(PipelineOptionsFactory.as(GcsOptions.class));
- }
-
- @Test
- public void testResolve() throws Exception {
- assertEquals("gs://bucket/object", factory.resolve("gs://bucket", "object"));
- }
-}
[2/2] beam git commit: This closes #2898
Posted by dh...@apache.org.
This closes #2898
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/44589c6f
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/44589c6f
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/44589c6f
Branch: refs/heads/master
Commit: 44589c6f9f70dd98c79cea387193812abc128d33
Parents: 9fffa7e 78ae528
Author: Dan Halperin <dh...@google.com>
Authored: Thu May 4 14:18:22 2017 -0700
Committer: Dan Halperin <dh...@google.com>
Committed: Thu May 4 14:18:22 2017 -0700
----------------------------------------------------------------------
.../beam/sdk/util/FileIOChannelFactory.java | 223 ----------------
.../sdk/util/FileIOChannelFactoryRegistrar.java | 38 ---
.../apache/beam/sdk/util/IOChannelFactory.java | 127 ---------
.../sdk/util/IOChannelFactoryRegistrar.java | 48 ----
.../apache/beam/sdk/util/IOChannelUtils.java | 259 +------------------
.../util/FileIOChannelFactoryRegistrarTest.java | 44 ----
.../beam/sdk/util/FileIOChannelFactoryTest.java | 242 -----------------
.../beam/sdk/util/IOChannelUtilsTest.java | 118 ---------
.../beam/sdk/util/GcsIOChannelFactory.java | 111 --------
.../sdk/util/GcsIOChannelFactoryRegistrar.java | 38 ---
.../util/GcsIOChannelFactoryRegistrarTest.java | 44 ----
.../beam/sdk/util/GcsIOChannelFactoryTest.java | 43 ---
12 files changed, 3 insertions(+), 1332 deletions(-)
----------------------------------------------------------------------