chore: Make newFileSystem work for s3x provider (#420)

* chore: Test `when bucket is empty, directory stream should be empty

* chore: Test `FileSystems$newFileSystem(URI, Map) should create bucket when name is valid` [skip ci]

* chore: Make `newFileSystem(URI uri, Map<String, ?> env)` support s3x provider

* chore: Delete FileSystemsTest unit test - noop

* fix: Avoid closing the client on readaheadbytechannel close

src/integrationTest/java/software/amazon/nio/spi/s3/FileSystemsNewFSTest.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package software.amazon.nio.spi.s3;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.FileSystems;
+import java.util.Map;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+@DisplayName("FileSystems$newFileSystem(URI, Map) should")
+public class FileSystemsNewFSTest {
+
+    @Test
+    @DisplayName("create bucket when name is valid")
+    public void createBucket() throws IOException {
+        String bucketLocation = Containers.localStackConnectionEndpoint() + "/fresh-bucket";
+        FileSystems.newFileSystem(
+            URI.create(bucketLocation),
+            Map.of("locationConstraint", "us-east-1")
+        );
+    }
+}

src/integrationTest/java/software/amazon/nio/spi/s3/FilesDirectoryStreamTest.java

@@ -5,6 +5,7 @@
 
 package software.amazon.nio.spi.s3;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static software.amazon.nio.spi.s3.Containers.localStackConnectionEndpoint;
 
@@ -20,10 +21,10 @@
 public class FilesDirectoryStreamTest {
 
     @Nested
-    @DisplayName("should throw")
+    @DisplayName("when bucket does not exist")
     class DirectoryDoesNotExist {
 
-        @DisplayName("when bucket does not exist")
+        @DisplayName("should throw")
         @Test
         @SuppressWarnings("resource")
         public void whenBucketNotFound() {
@@ -33,4 +34,20 @@ public void whenBucketNotFound() {
         }
     }
 
+    @Nested
+    @DisplayName("when bucket exists")
+    class DirectoryExists {
+
+        @DisplayName("and is empty, list should be empty")
+        @Test
+        public void listShouldBeEmpty() throws IOException {
+            Containers.createBucket("new-directory-stream");
+            final var path = Paths.get(URI.create(localStackConnectionEndpoint() + "/new-directory-stream/"));
+
+            try(var stream = Files.newDirectoryStream(path, p -> true)) {
+                assertThat(stream).isEmpty();
+            }
+        }
+    }
+
 }

src/main/java/software/amazon/nio/spi/s3/S3ClientProvider.java

@@ -181,10 +181,7 @@ private String getBucketRegionFromResponse(SdkHttpResponse response) {
         );
     }
 
-    private S3AsyncClient configureCrtClientForRegion(String regionName) {
-        var region = getRegionFromRegionName(regionName);
-        logger.debug("bucket region is: '{}'", region);
-
+    S3CrtAsyncClientBuilder configureCrtClient() {
         var endpointUri = configuration.endpointUri();
         if (endpointUri != null) {
             asyncClientBuilder.endpointOverride(endpointUri);
@@ -195,9 +192,13 @@ private S3AsyncClient configureCrtClientForRegion(String regionName) {
             asyncClientBuilder.credentialsProvider(() -> credentials);
         }
 
-        return asyncClientBuilder.forcePathStyle(configuration.getForcePathStyle())
-                .region(region)
-                .build();
+        return asyncClientBuilder.forcePathStyle(configuration.getForcePathStyle());
+    }
+
+    private S3AsyncClient configureCrtClientForRegion(String regionName) {
+        var region = getRegionFromRegionName(regionName);
+        logger.debug("bucket region is: '{}'", region);
+        return configureCrtClient().region(region).build();
     }
 
     private static Region getRegionFromRegionName(String regionName) {

src/main/java/software/amazon/nio/spi/s3/S3FileSystemProvider.java

@@ -138,15 +138,21 @@ public String getScheme() {
      */
     @Override
     public FileSystem newFileSystem(final URI uri, final Map<String, ?> env) throws IOException {
-        if (! uri.getScheme().equals(SCHEME)) {
-            throw new IllegalArgumentException("URI scheme must be s3");
+        if (!uri.getScheme().equals(getScheme())) {
+            throw new IllegalArgumentException("URI scheme must be " + getScheme());
         }
 
         @SuppressWarnings("unchecked")
         var envMap = (Map<String, Object>) env;
 
-        var bucketName = uri.getAuthority();
-        try (var client = S3AsyncClient.create()) {
+        var info = fileSystemInfo(uri);
+        var config = new S3NioSpiConfiguration().withEndpoint(info.endpoint()).withBucketName(info.bucket());
+        if (info.accessKey() != null) {
+            config.withCredentials(info.accessKey(), info.accessSecret());
+        }
+        var bucketName = config.getBucketName();
+
+        try (var client = new S3ClientProvider(config).configureCrtClient().build()) {
             var createBucketResponse = client.createBucket(
                     bucketBuilder -> bucketBuilder.bucket(bucketName)
                             .acl(envMap.getOrDefault("acl", "").toString())
@@ -172,11 +178,7 @@ public FileSystem newFileSystem(final URI uri, final Map<String, ?> env) throws
         } catch (InterruptedException | TimeoutException | SdkException e) {
             throw new IOException(e.getMessage(), e);
         }
-
-
         return getFileSystem(uri, true);
-
-
     }
 
     /**

src/main/java/software/amazon/nio/spi/s3/S3ReadAheadByteChannel.java

@@ -177,7 +177,6 @@ public void close() {
         open = false;
         readAheadBuffersCache.invalidateAll();
         readAheadBuffersCache.cleanUp();
-        client.close();
     }
 
     private void clearPriorFragments(int currentFragIndx) {