cloud-storage-contrib-nio: Add support for newFileChannel

google-cloud-clients/google-cloud-contrib/google-cloud-nio/src/main/java/com/google/cloud/storage/contrib/nio/CloudStorageFileSystemProvider.java

@@ -42,6 +42,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URI;
+import java.nio.channels.FileChannel;
 import java.nio.channels.SeekableByteChannel;
 import java.nio.file.AccessMode;
 import java.nio.file.AtomicMoveNotSupportedException;
@@ -300,6 +301,48 @@ public SeekableByteChannel newByteChannel(
     }
   }
 
+  /**
+   * Open a file for reading OR writing. The {@link FileChannel} that is returned will only allow
+   * reads or writes depending on the {@link OpenOption}s that are specified. If any of the
+   * following have been specified, the {@link FileChannel} will be write-only: {@link
+   * StandardOpenOption#CREATE}
+   *
+   * <ul>
+   *   <li>{@link StandardOpenOption#CREATE}
+   *   <li>{@link StandardOpenOption#CREATE_NEW}
+   *   <li>{@link StandardOpenOption#WRITE}
+   *   <li>{@link StandardOpenOption#TRUNCATE_EXISTING}
+   * </ul>
+   *
+   * In all other cases the {@link FileChannel} will be read-only.
+   *
+   * @param path The path to the file to open or create
+   * @param options The options specifying how the file should be opened, and whether the {@link
+   *     FileChannel} should be read-only or write-only.
+   * @param attrs (not supported, the values will be ignored)
+   * @throws IOException
+   */
+  @Override
+  public FileChannel newFileChannel(
+      Path path, Set<? extends OpenOption> options, FileAttribute<?>... attrs) throws IOException {
+    checkNotNull(path);
+    initStorage();
+    CloudStorageUtil.checkNotNullArray(attrs);
+    if (options.contains(StandardOpenOption.CREATE_NEW)) {
+      Files.createFile(path, attrs);
+    } else if (options.contains(StandardOpenOption.CREATE) && !Files.exists(path)) {
+      Files.createFile(path, attrs);
+    }
+    if (options.contains(StandardOpenOption.WRITE)
+        || options.contains(StandardOpenOption.CREATE)
+        || options.contains(StandardOpenOption.CREATE_NEW)
+        || options.contains(StandardOpenOption.TRUNCATE_EXISTING)) {
+      return new CloudStorageWriteFileChannel(newWriteChannel(path, options));
+    } else {
+      return new CloudStorageReadFileChannel(newReadChannel(path, options));
+    }
+  }
+
   private SeekableByteChannel newReadChannel(Path path, Set<? extends OpenOption> options)
       throws IOException {
     initStorage();

google-cloud-clients/google-cloud-contrib/google-cloud-nio/src/main/java/com/google/cloud/storage/contrib/nio/CloudStorageReadFileChannel.java

@@ -0,0 +1,148 @@
+/*
+ * Copyright 2019 Google LLC
+ *
+ * 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 com.google.cloud.storage.contrib.nio;
+
+import com.google.common.base.Preconditions;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.MappedByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.channels.WritableByteChannel;
+
+class CloudStorageReadFileChannel extends FileChannel {
+  private static final String READ_ONLY = "This FileChannel is read-only";
+  private final SeekableByteChannel readChannel;
+
+  CloudStorageReadFileChannel(SeekableByteChannel readChannel) {
+    Preconditions.checkNotNull(readChannel);
+    this.readChannel = readChannel;
+  }
+
+  @Override
+  public int read(ByteBuffer dst) throws IOException {
+    return readChannel.read(dst);
+  }
+
+  @Override
+  public synchronized long read(ByteBuffer[] dsts, int offset, int length) throws IOException {
+    long res = 0L;
+    for (int i = offset; i < offset + length; i++) {
+      res += readChannel.read(dsts[i]);
+    }
+    return res;
+  }
+
+  @Override
+  public int write(ByteBuffer src) throws IOException {
+    throw new UnsupportedOperationException(READ_ONLY);
+  }
+
+  @Override
+  public long write(ByteBuffer[] srcs, int offset, int length) throws IOException {
+    throw new UnsupportedOperationException(READ_ONLY);
+  }
+
+  @Override
+  public long position() throws IOException {
+    return readChannel.position();
+  }
+
+  @Override
+  public FileChannel position(long newPosition) throws IOException {
+    readChannel.position(newPosition);
+    return this;
+  }
+
+  @Override
+  public long size() throws IOException {
+    return readChannel.size();
+  }
+
+  @Override
+  public FileChannel truncate(long size) throws IOException {
+    throw new UnsupportedOperationException(READ_ONLY);
+  }
+
+  @Override
+  public void force(boolean metaData) throws IOException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public synchronized long transferTo(
+      long transferFromPosition, long count, WritableByteChannel target) throws IOException {
+    long res = 0L;
+    long originalPosition = position();
+    position(transferFromPosition);
+    int blockSize = (int) Math.min(count, 0xfffffL);
+    int bytesRead = 0;
+    ByteBuffer buffer = ByteBuffer.allocate(blockSize);
+    while (res < count && bytesRead >= 0) {
+      buffer.position(0);
+      bytesRead = read(buffer);
+      if (bytesRead > 0) {
+        buffer.position(0);
+        buffer.limit(bytesRead);
+        target.write(buffer);
+        res += bytesRead;
+      }
+    }
+    position(originalPosition);
+    return res;
+  }
+
+  @Override
+  public long transferFrom(ReadableByteChannel src, long position, long count) throws IOException {
+    throw new UnsupportedOperationException(READ_ONLY);
+  }
+
+  @Override
+  public synchronized int read(ByteBuffer dst, long readFromPosition) throws IOException {
+    long originalPosition = position();
+    position(readFromPosition);
+    int res = readChannel.read(dst);
+    position(originalPosition);
+    return res;
+  }
+
+  @Override
+  public int write(ByteBuffer src, long position) throws IOException {
+    throw new UnsupportedOperationException(READ_ONLY);
+  }
+
+  @Override
+  public MappedByteBuffer map(MapMode mode, long position, long size) throws IOException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public FileLock lock(long position, long size, boolean shared) throws IOException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public FileLock tryLock(long position, long size, boolean shared) throws IOException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  protected void implCloseChannel() throws IOException {
+    readChannel.close();
+  }
+}

google-cloud-clients/google-cloud-contrib/google-cloud-nio/src/main/java/com/google/cloud/storage/contrib/nio/CloudStorageWriteFileChannel.java

@@ -0,0 +1,180 @@
+/*
+ * Copyright 2019 Google LLC
+ *
+ * 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 com.google.cloud.storage.contrib.nio;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.MappedByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.channels.WritableByteChannel;
+
+class CloudStorageWriteFileChannel extends FileChannel {
+  private static final String WRITE_ONLY = "This FileChannel is write-only";
+  private SeekableByteChannel writeChannel;
+  private boolean valid = true;
+
+  CloudStorageWriteFileChannel(SeekableByteChannel writeChannel) {
+    this.writeChannel = writeChannel;
+  }
+
+  private void checkValid() throws IOException {
+    if (!valid) {
+      // These methods are only supported to be called once, because the underlying channel does not
+      // support changing the position.
+      throw new IOException(
+          "This FileChannel is no longer valid. "
+              + "A Cloud Storage FileChannel is invalidated after calling one of "
+              + "the methods FileChannel#write(ByteBuffer, long) or "
+              + "FileChannel#transferFrom(ReadableByteChannel, long, long)");
+    }
+    if (!writeChannel.isOpen()) {
+      throw new IOException("This FileChannel is closed");
+    }
+  }
+
+  @Override
+  public int read(ByteBuffer dst) throws IOException {
+    throw new UnsupportedOperationException(WRITE_ONLY);
+  }
+
+  @Override
+  public long read(ByteBuffer[] dsts, int offset, int length) throws IOException {
+    throw new UnsupportedOperationException(WRITE_ONLY);
+  }
+
+  @Override
+  public synchronized int write(ByteBuffer src) throws IOException {
+    checkValid();
+    return writeChannel.write(src);
+  }
+
+  @Override
+  public synchronized long write(ByteBuffer[] srcs, int offset, int length) throws IOException {
+    checkValid();
+    long res = 0L;
+    for (int i = offset; i < offset + length; i++) {
+      res += writeChannel.write(srcs[i]);
+    }
+    return res;
+  }
+
+  @Override
+  public synchronized long position() throws IOException {
+    checkValid();
+    return writeChannel.position();
+  }
+
+  @Override
+  public synchronized FileChannel position(long newPosition) throws IOException {
+    if (newPosition != position()) {
+      writeChannel.position(newPosition);
+    }
+    return this;
+  }
+
+  @Override
+  public synchronized long size() throws IOException {
+    checkValid();
+    return writeChannel.size();
+  }
+
+  @Override
+  public synchronized FileChannel truncate(long size) throws IOException {
+    checkValid();
+    writeChannel.truncate(size);
+    return this;
+  }
+
+  @Override
+  public void force(boolean metaData) throws IOException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public long transferTo(long position, long count, WritableByteChannel target) throws IOException {
+    throw new UnsupportedOperationException(WRITE_ONLY);
+  }
+
+  @Override
+  public synchronized long transferFrom(ReadableByteChannel src, long position, long count)
+      throws IOException {
+    if (position != position()) {
+      throw new UnsupportedOperationException(
+          "This FileChannel only supports transferFrom at the current position");
+    }
+    int blockSize = (int) Math.min(count, 0xfffffL);
+    long res = 0L;
+    int bytesRead = 0;
+    ByteBuffer buffer = ByteBuffer.allocate(blockSize);
+    while (res < count && bytesRead >= 0) {
+      buffer.position(0);
+      bytesRead = src.read(buffer);
+      if (bytesRead > 0) {
+        buffer.position(0);
+        buffer.limit(bytesRead);
+        write(buffer);
+        res += bytesRead;
+      }
+    }
+    // The channel is no longer valid as the position has been updated, and there is no way of
+    // resetting it, but this way we at least support the write-at-position and transferFrom
+    // methods being called once.
+    this.valid = false;
+    return res;
+  }
+
+  @Override
+  public int read(ByteBuffer dst, long position) throws IOException {
+    throw new UnsupportedOperationException(WRITE_ONLY);
+  }
+
+  @Override
+  public synchronized int write(ByteBuffer src, long position) throws IOException {
+    if (position != position()) {
+      throw new UnsupportedOperationException(
+          "This FileChannel only supports write at the current position");
+    }
+    int res = writeChannel.write(src);
+    // The channel is no longer valid as the position has been updated, and there is no way of
+    // resetting it, but this way we at least support the write-at-position and transferFrom
+    // methods being called once.
+    this.valid = false;
+    return res;
+  }
+
+  @Override
+  public MappedByteBuffer map(MapMode mode, long position, long size) throws IOException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public FileLock lock(long position, long size, boolean shared) throws IOException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public FileLock tryLock(long position, long size, boolean shared) throws IOException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  protected void implCloseChannel() throws IOException {
+    writeChannel.close();
+  }
+}