Nio byte channel (Azure#18836)

* Some experiments with seekable byte channel

* Updated to use ClosedFileSystemException instead of IOException

* Wrote all tests except for seek and read and write

* Added all tests

* Fixed all tests

* Recordings and cleanup

* ci fixes

* Reverted test file size

* Updated recordings

* fixed a test and responded to feedback

* Fixed warnings

* typo fix

* Update sdk/storage/azure-storage-blob-nio/src/main/java/com/azure/storage/blob/nio/AzureFileSystemProvider.java

Co-authored-by: Alan Zimmer <48699787+alzimmermsft@users.noreply.github.com>

* Pr feedback

* PR feedback

* small javadoc fix for ci

* Unused imports

* Fixed buffer position issue

Co-authored-by: Alan Zimmer <48699787+alzimmermsft@users.noreply.github.com>

Author: rickle-msft

sdk/storage/azure-storage-blob-nio/CHANGELOG.md

@@ -3,6 +3,8 @@
 ## 12.0.0-beta.3 (Unreleased)
 - Added support for FileSystemProvider.checkAccess method
 - Added support for file key on AzureBasicFileAttributes and AzureBlobFileAttributes
+- Added support for SeekableByteChannel
+- When an operation is performed on a closed FileSystem, a ClosedFileSystemException is thrown instead of an IOException
 - Adjusted the required flags for opening an outputstream
 
 ## 12.0.0-beta.2 (2020-08-13)

sdk/storage/azure-storage-blob-nio/src/main/java/com/azure/storage/blob/nio/AzureDirectoryStream.java

@@ -100,11 +100,8 @@ private static class AzureDirectoryIterator implements Iterator<Path> {
 
         @Override
         public boolean hasNext() {
-            try {
-                AzurePath.ensureFileSystemOpen(path);
-            } catch (IOException e) {
-                throw LoggingUtility.logError(logger, new DirectoryIteratorException(e));
-            }
+            AzurePath.ensureFileSystemOpen(path);
+
             // Closing the parent stream halts iteration.
             if (parentStream.closed) {
                 return false;

sdk/storage/azure-storage-blob-nio/src/main/java/com/azure/storage/blob/nio/AzureFileSystemProvider.java

@@ -31,6 +31,7 @@
 import java.nio.file.FileSystem;
 import java.nio.file.FileSystemAlreadyExistsException;
 import java.nio.file.FileSystemNotFoundException;
+import java.nio.file.Files;
 import java.nio.file.LinkOption;
 import java.nio.file.NoSuchFileException;
 import java.nio.file.NotDirectoryException;
@@ -46,9 +47,12 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -158,6 +162,17 @@ public final class AzureFileSystemProvider extends FileSystemProvider {
 
     private static final String ACCOUNT_QUERY_KEY = "account";
     private static final int COPY_TIMEOUT_SECONDS = 30;
+    private static final Set<OpenOption> OUTPUT_STREAM_DEFAULT_OPTIONS =
+        Collections.unmodifiableSet(new HashSet<>(Arrays.asList(StandardOpenOption.CREATE,
+            StandardOpenOption.WRITE,
+            StandardOpenOption.TRUNCATE_EXISTING)));
+    private static final Set<OpenOption> OUTPUT_STREAM_SUPPORTED_OPTIONS =
+        Collections.unmodifiableSet(new HashSet<>(Arrays.asList(
+            StandardOpenOption.CREATE_NEW,
+            StandardOpenOption.CREATE,
+            StandardOpenOption.WRITE,
+            // Though we don't actually truncate, the same result is achieved by overwriting the destination.
+            StandardOpenOption.TRUNCATE_EXISTING)));
 
     private final ConcurrentMap<String, FileSystem> openFileSystems;
 
@@ -252,12 +267,20 @@ public Path getPath(URI uri) {
     }
 
     /**
-     * Unsupported. Use {@link #newInputStream(Path, OpenOption...)} or {@link #newOutputStream(Path, OpenOption...)}
-     * instead.
+     * Opens or creates a file, returning a seekable byte channel to access the file.
+     * <p>
+     * This method is primarily offered to support some jdk convenience methods such as
+     * {@link Files#createFile(Path, FileAttribute[])} which requires opening a channel and closing it. A channel may
+     * only be opened in read mode OR write mode. It may not be opened in read/write mode. Seeking is supported for
+     * reads, but not for writes. Modifications to existing files is not permitted--only creating new files or
+     * overwriting existing files.
+     * <p>
+     * This type is not threadsafe to prevent having to hold locks across network calls.
+     * <p>
      *
-     * @param path the Path
-     * @param set open options
-     * @param fileAttributes attributes
+     * @param path the path of the file to open
+     * @param set options specifying how the file should be opened
+     * @param fileAttributes an optional list of file attributes to set atomically when creating the directory
      * @return a new seekable byte channel
      * @throws UnsupportedOperationException Operation is not supported.
      * @throws IllegalArgumentException if the set contains an invalid combination of options
@@ -269,7 +292,17 @@ public Path getPath(URI uri) {
     @Override
     public SeekableByteChannel newByteChannel(Path path, Set<? extends OpenOption> set,
             FileAttribute<?>... fileAttributes) throws IOException {
-        throw LoggingUtility.logError(logger, new UnsupportedOperationException());
+        if (Objects.isNull(set)) {
+            set = Collections.emptySet();
+        }
+
+        if (set.contains(StandardOpenOption.WRITE)) {
+            return new AzureSeekableByteChannel(
+                (NioBlobOutputStream) this.newOutputStreamInternal(path, set, fileAttributes), path);
+        } else {
+            return new AzureSeekableByteChannel(
+                (NioBlobInputStream) this.newInputStream(path, set.toArray(new OpenOption[0])), path);
+        }
     }
 
     /**
@@ -349,24 +382,19 @@ public InputStream newInputStream(Path path, OpenOption... options) throws IOExc
      */
     @Override
     public OutputStream newOutputStream(Path path, OpenOption... options) throws IOException {
+        return newOutputStreamInternal(path, new HashSet<>(Arrays.asList(options)));
+    }
+
+    OutputStream newOutputStreamInternal(Path path, Set<? extends OpenOption> optionsSet,
+        FileAttribute<?>... fileAttributes) throws IOException {
         // If options are empty, add Create, Write, TruncateExisting as defaults per nio docs.
-        if (options == null || options.length == 0) {
-            options = new OpenOption[] {
-                StandardOpenOption.CREATE,
-                StandardOpenOption.WRITE,
-                StandardOpenOption.TRUNCATE_EXISTING };
+        if (optionsSet == null || optionsSet.size() == 0) {
+            optionsSet = OUTPUT_STREAM_DEFAULT_OPTIONS;
         }
-        List<OpenOption> optionsList = Arrays.asList(options);
 
         // Check for unsupported options.
-        List<OpenOption> supportedOptions = Arrays.asList(
-            StandardOpenOption.CREATE_NEW,
-            StandardOpenOption.CREATE,
-            StandardOpenOption.WRITE,
-            // Though we don't actually truncate, the same result is achieved by overwriting the destination.
-            StandardOpenOption.TRUNCATE_EXISTING);
-        for (OpenOption option : optionsList) {
-            if (!supportedOptions.contains(option)) {
+        for (OpenOption option : optionsSet) {
+            if (!OUTPUT_STREAM_SUPPORTED_OPTIONS.contains(option)) {
                 throw LoggingUtility.logError(logger, new UnsupportedOperationException("Unsupported option: "
                     + option.toString()));
             }
@@ -376,9 +404,9 @@ public OutputStream newOutputStream(Path path, OpenOption... options) throws IOE
         Write must be specified. Either create_new or truncate must be specified. This is to ensure that no edits or
         appends are allowed.
          */
-        if (!optionsList.contains(StandardOpenOption.WRITE)
-            || !(optionsList.contains(StandardOpenOption.TRUNCATE_EXISTING)
-            || optionsList.contains(StandardOpenOption.CREATE_NEW))) {
+        if (!optionsSet.contains(StandardOpenOption.WRITE)
+            || !(optionsSet.contains(StandardOpenOption.TRUNCATE_EXISTING)
+            || optionsSet.contains(StandardOpenOption.CREATE_NEW))) {
             throw LoggingUtility.logError(logger,
                 new IllegalArgumentException("Write and either CreateNew or TruncateExisting must be specified to open "
                     + "an OutputStream"));
@@ -396,14 +424,14 @@ public OutputStream newOutputStream(Path path, OpenOption... options) throws IOE
 
         // Writing to an empty location requires a create option.
         if (status.equals(DirectoryStatus.DOES_NOT_EXIST)
-            && !(optionsList.contains(StandardOpenOption.CREATE)
-            || optionsList.contains(StandardOpenOption.CREATE_NEW))) {
+            && !(optionsSet.contains(StandardOpenOption.CREATE)
+            || optionsSet.contains(StandardOpenOption.CREATE_NEW))) {
             throw LoggingUtility.logError(logger, new IOException("Writing to an empty location requires a create "
                 + "option. Path: " + path.toString()));
         }
 
         // Cannot write to an existing file if create new was specified.
-        if (status.equals(DirectoryStatus.NOT_A_DIRECTORY) && optionsList.contains(StandardOpenOption.CREATE_NEW)) {
+        if (status.equals(DirectoryStatus.NOT_A_DIRECTORY) && optionsSet.contains(StandardOpenOption.CREATE_NEW)) {
             throw LoggingUtility.logError(logger, new IOException("A file already exists at this location and "
                 + "CREATE_NEW was specified. Path: " + path.toString()));
         }
@@ -417,12 +445,17 @@ public OutputStream newOutputStream(Path path, OpenOption... options) throws IOE
 
         // Add an extra etag check for create new
         BlobRequestConditions rq = null;
-        if (optionsList.contains(StandardOpenOption.CREATE_NEW)) {
+        if (optionsSet.contains(StandardOpenOption.CREATE_NEW)) {
             rq = new BlobRequestConditions().setIfNoneMatch("*");
         }
 
-        return new NioBlobOutputStream(resource.getBlobClient().getBlockBlobClient().getBlobOutputStream(pto, null,
-            null, null, rq), resource.getPath());
+        // For parsing properties and metadata
+        if (fileAttributes == null) {
+            fileAttributes = new FileAttribute<?>[0];
+        }
+        resource.setFileAttributes(Arrays.asList(fileAttributes));
+
+        return new NioBlobOutputStream(resource.getBlobOutputStream(pto, rq), resource.getPath());
     }
 
     /**
@@ -992,7 +1025,6 @@ public Map<String, Object> readAttributes(Path path, String attributes, LinkOpti
 
             // If "*" is specified, add all of the attributes from the specified set.
             if (attributeName.equals("*")) {
-                Set<String> attributesToAdd;
                 if (viewType.equals(AzureBasicFileAttributeView.NAME)) {
                     for (String attr : AzureBasicFileAttributes.ATTRIBUTE_STRINGS) {
                         results.put(attr, attributeSuppliers.get(attr).get());

sdk/storage/azure-storage-blob-nio/src/main/java/com/azure/storage/blob/nio/AzurePath.java

@@ -11,6 +11,7 @@
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.nio.file.ClosedFileSystemException;
 import java.nio.file.FileSystem;
 import java.nio.file.InvalidPathException;
 import java.nio.file.LinkOption;
@@ -774,10 +775,10 @@ private String rootToFileStore(String root) {
         return root.substring(0, root.length() - 1); // Remove the ROOT_DIR_SUFFIX
     }
 
-    static void ensureFileSystemOpen(Path p) throws IOException {
+    static void ensureFileSystemOpen(Path p) {
         if (!p.getFileSystem().isOpen()) {
             throw LoggingUtility.logError(((AzurePath) p).logger,
-                new IOException("FileSystem for path has been closed. Path: " + p.toString()));
+                new ClosedFileSystemException());
         }
     }
 }

sdk/storage/azure-storage-blob-nio/src/main/java/com/azure/storage/blob/nio/AzureResource.java

@@ -13,6 +13,9 @@
 import com.azure.storage.blob.models.BlobRequestConditions;
 import com.azure.storage.blob.models.BlobStorageException;
 import com.azure.storage.blob.models.ListBlobsOptions;
+import com.azure.storage.blob.models.ParallelTransferOptions;
+import com.azure.storage.blob.options.BlockBlobOutputStreamOptions;
+import com.azure.storage.blob.specialized.BlobOutputStream;
 import com.azure.storage.common.implementation.Constants;
 
 import java.io.IOException;
@@ -240,6 +243,15 @@ BlobClient getBlobClient() {
         return this.blobClient;
     }
 
+    BlobOutputStream getBlobOutputStream(ParallelTransferOptions pto, BlobRequestConditions rq) {
+        BlockBlobOutputStreamOptions options = new BlockBlobOutputStreamOptions()
+            .setHeaders(this.blobHeaders)
+            .setMetadata(this.blobMetadata)
+            .setParallelTransferOptions(pto)
+            .setRequestConditions(rq);
+        return this.blobClient.getBlockBlobClient().getBlobOutputStream(options);
+    }
+
     private Map<String, String> prepareMetadataForDirectory() {
         if (this.blobMetadata == null) {
             this.blobMetadata = new HashMap<>();