ChunkedNioFile can use absolute FileChannel::read to read chunks (#9592)

Motivation:

Users can reuse the same FileChannel for different ChunkedNioFile
instances without being worried that FileChannel::position will be
changed concurrently by them.
In addition, FileChannel::read with absolute position allows to
use on *nix pread that is more efficient then fread.

Modifications:

Always use absolute FileChannel::read ops

Result:

Faster and more flexible uses of FileChannel for ChunkedNioFile

handler/src/main/java/io/netty/handler/stream/ChunkedNioFile.java

@@ -23,6 +23,7 @@ import io.netty.channel.FileRegion;
 import java.io.File;
 import java.io.IOException;
 import java.io.RandomAccessFile;
+import java.nio.channels.ClosedChannelException;
 import java.nio.channels.FileChannel;
 
 /**
@@ -101,9 +102,8 @@ public class ChunkedNioFile implements ChunkedInput<ByteBuf> {
                     "chunkSize: " + chunkSize +
                     " (expected: a positive integer)");
         }
-
-        if (offset != 0) {
-            in.position(offset);
+        if (!in.isOpen()) {
+            throw new ClosedChannelException();
         }
         this.in = in;
         this.chunkSize = chunkSize;
@@ -161,7 +161,7 @@ public class ChunkedNioFile implements ChunkedInput<ByteBuf> {
         try {
             int readBytes = 0;
             for (;;) {
-                int localReadBytes = buffer.writeBytes(in, chunkSize - readBytes);
+                int localReadBytes = buffer.writeBytes(in, offset + readBytes, chunkSize - readBytes);
                 if (localReadBytes < 0) {
                     break;
                 }

handler/src/test/java/io/netty/handler/stream/ChunkedWriteHandlerTest.java

@@ -26,13 +26,17 @@ import io.netty.channel.ChannelPromise;
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.util.CharsetUtil;
 import io.netty.util.ReferenceCountUtil;
+import org.junit.Assert;
 import org.junit.Test;
 
 import java.io.ByteArrayInputStream;
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.io.RandomAccessFile;
 import java.nio.channels.Channels;
+import java.nio.channels.ClosedChannelException;
+import java.nio.channels.FileChannel;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -102,6 +106,41 @@ public class ChunkedWriteHandlerTest {
         check(new ChunkedNioFile(TMP), new ChunkedNioFile(TMP), new ChunkedNioFile(TMP));
     }
 
+    @Test
+    public void testChunkedNioFileLeftPositionUnchanged() throws IOException {
+        FileChannel in = null;
+        final long expectedPosition = 10;
+        try {
+            in = new RandomAccessFile(TMP, "r").getChannel();
+            in.position(expectedPosition);
+            check(new ChunkedNioFile(in) {
+                @Override
+                public void close() throws Exception {
+                    //no op
+                }
+            });
+            Assert.assertTrue(in.isOpen());
+            Assert.assertEquals(expectedPosition, in.position());
+        } finally {
+            if (in != null) {
+                in.close();
+            }
+        }
+    }
+
+    @Test(expected = ClosedChannelException.class)
+    public void testChunkedNioFileFailOnClosedFileChannel() throws IOException {
+        final FileChannel in = new RandomAccessFile(TMP, "r").getChannel();
+        in.close();
+        check(new ChunkedNioFile(in) {
+            @Override
+            public void close() throws Exception {
+                //no op
+            }
+        });
+        Assert.fail();
+    }
+
     @Test
     public void testUnchunkedData() throws IOException {
         check(Unpooled.wrappedBuffer(BYTES));