feature: implements newFileChannel and newAsynchronousFileChannel (#488)

* enables creation of FileChannels and AsynchronousFileChannels and provides demos.

Author: markjschreiber

src/examples/java/software/amazon/nio/spi/examples/AsyncFileChannelDemo.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package software.amazon.nio.spi.examples;
+
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousFileChannel;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
+import java.util.concurrent.TimeUnit;
+
+public class AsyncFileChannelDemo {
+    public static void main(String[] args) {
+        var path = Paths.get(URI.create(args[0]));
+        try (var channel = AsynchronousFileChannel.open(path, StandardOpenOption.READ)) {
+            var buffer = ByteBuffer.allocate(1024);
+            long position = 0;
+            int bytesRead;
+            while ((bytesRead = channel.read(buffer, position).get(10, TimeUnit.SECONDS)) != -1) {
+                position += bytesRead;
+                buffer.flip();
+                while (buffer.hasRemaining()) {
+                    System.out.print((char) buffer.get());
+                }
+                buffer.clear();
+            }
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

src/examples/java/software/amazon/nio/spi/examples/FileChannelDemo.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package software.amazon.nio.spi.examples;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
+
+public class FileChannelDemo {
+
+    public static void main(String[] args) {
+        var path = Paths.get(URI.create(args[0]));
+        try (var channel = FileChannel.open(path, StandardOpenOption.READ)) {
+            var buffer = ByteBuffer.allocate(1024);
+            while (channel.read(buffer) != -1) {
+                buffer.flip();
+                while (buffer.hasRemaining()) {
+                    System.out.print((char) buffer.get());
+                }
+                buffer.clear();
+            }
+        } catch (IOException e) {
+            e.printStackTrace();
+            System.exit(1);
+        }
+    }
+}

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

@@ -0,0 +1,120 @@
+/*
+ * 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.nio.ByteBuffer;
+import java.nio.channels.AsynchronousFileChannel;
+import java.nio.channels.CompletionHandler;
+import java.nio.channels.FileLock;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import software.amazon.nio.spi.s3.util.TimeOutUtils;
+
+public class AsyncS3FileChannel extends AsynchronousFileChannel {
+
+    private final S3SeekableByteChannel byteChannel;
+
+    AsyncS3FileChannel(S3SeekableByteChannel byteChannel) {
+        this.byteChannel = byteChannel;
+    }
+
+    @Override
+    public long size() throws IOException {
+        return byteChannel.size();
+    }
+
+    @Override
+    public AsynchronousFileChannel truncate(long size) throws IOException {
+        byteChannel.truncate(size);
+        return this;
+    }
+
+    @Override
+    public void force(boolean metaData) throws IOException {
+        if (byteChannel.getWriteDelegate() != null) {
+            ((S3WritableByteChannel) byteChannel.getWriteDelegate()).force();
+        }
+    }
+
+    @Override
+    public <A> void lock(long position, long size, boolean shared, A attachment, CompletionHandler<FileLock, ? super A> handler) {
+        throw new UnsupportedOperationException("S3 does not support file locking");
+    }
+
+    @Override
+    public Future<FileLock> lock(long position, long size, boolean shared) {
+        return CompletableFuture.failedFuture(new UnsupportedOperationException("S3 does not support file locking"));
+    }
+
+    @Override
+    public FileLock tryLock(long position, long size, boolean shared) throws IOException {
+        throw new IOException(new UnsupportedOperationException("S3 does not support file locking"));
+    }
+
+    @Override
+    public <A> void read(ByteBuffer dst, long position, A attachment, CompletionHandler<Integer, ? super A> handler) {
+        Future<Integer> future =  read(dst, position);
+        try {
+            handler.completed(future.get(TimeOutUtils.TIMEOUT_TIME_LENGTH_5, TimeUnit.MINUTES), attachment);
+        } catch (Exception e) {
+            handler.failed(e, attachment);
+        }
+    }
+
+    @Override
+    public Future<Integer> read(ByteBuffer dst, long position) {
+        if (position < 0) {
+            throw new IllegalArgumentException("position: " + position);
+        }
+
+        return CompletableFuture.supplyAsync(() -> {
+            try {
+                byteChannel.position(position);
+                return byteChannel.read(dst);
+            } catch (IOException e) {
+                throw new RuntimeException(e);
+            }
+        });
+    }
+
+    @Override
+    public <A> void write(ByteBuffer src, long position, A attachment, CompletionHandler<Integer, ? super A> handler) {
+        Future<Integer> future = write(src, position);
+        try {
+            handler.completed(future.get(TimeOutUtils.TIMEOUT_TIME_LENGTH_5, TimeUnit.MINUTES), attachment);
+        } catch (Exception e) {
+            handler.failed(e, attachment);
+        }
+    }
+
+    @Override
+    public Future<Integer> write(ByteBuffer src, long position) {
+        if (position < 0) {
+            throw new IllegalArgumentException("position: " + position);
+        }
+
+        return CompletableFuture.supplyAsync(() -> {
+            try {
+                byteChannel.position(position);
+                return byteChannel.write(src);
+            } catch (IOException e) {
+                throw new RuntimeException(e);
+            }
+        });
+    }
+
+    @Override
+    public boolean isOpen() {
+        return byteChannel.isOpen();
+    }
+
+    @Override
+    public void close() throws IOException {
+        byteChannel.close();
+    }
+}