Add methods with position independent FileChannel calls to ByteBuf

Motivation See #netty#3229 Modifications: Add methods with position independent FileChannel calls to ByteBuf and its subclasses. Results: The user can use these new methods to read/write ByteBuff without updating FileChannel's position.
jin716 · Feb 15, 2016 · ccb0870 · ccb0870
1 parent b112673
commit ccb0870
Show file tree

Hide file tree

Showing 19 changed files with 652 additions and 3 deletions.
diff --git a/buffer/src/main/java/io/netty/buffer/AbstractByteBuf.java b/buffer/src/main/java/io/netty/buffer/AbstractByteBuf.java
@@ -29,6 +29,7 @@
 import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
+import java.nio.channels.FileChannel;
 import java.nio.channels.GatheringByteChannel;
 import java.nio.channels.ScatteringByteChannel;
 import java.nio.charset.Charset;
@@ -856,6 +857,15 @@ public int readBytes(GatheringByteChannel out, int length)
         return readBytes;
     }
 
+    @Override
+    public int readBytes(FileChannel out, long position, int length)
+            throws IOException {
+        checkReadableBytes(length);
+        int readBytes = getBytes(readerIndex, out, position, length);
+        readerIndex += readBytes;
+        return readBytes;
+    }
+
     @Override
     public ByteBuf readBytes(OutputStream out, int length) throws IOException {
         checkReadableBytes(length);
@@ -1049,6 +1059,17 @@ public int writeBytes(ScatteringByteChannel in, int length) throws IOException {
         return writtenBytes;
     }
 
+    @Override
+    public int writeBytes(FileChannel in, long position, int length) throws IOException {
+        ensureAccessible();
+        ensureWritable(length);
+        int writtenBytes = setBytes(writerIndex, in, position, length);
+        if (writtenBytes > 0) {
+            writerIndex += writtenBytes;
+        }
+        return writtenBytes;
+    }
+
     @Override
     public ByteBuf writeZero(int length) {
         if (length == 0) {

diff --git a/buffer/src/main/java/io/netty/buffer/ByteBuf.java b/buffer/src/main/java/io/netty/buffer/ByteBuf.java
@@ -23,6 +23,7 @@
 import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
+import java.nio.channels.FileChannel;
 import java.nio.channels.GatheringByteChannel;
 import java.nio.channels.ScatteringByteChannel;
 import java.nio.charset.Charset;
@@ -880,6 +881,26 @@ public abstract class ByteBuf implements ReferenceCounted, Comparable<ByteBuf> {
      */
     public abstract int getBytes(int index, GatheringByteChannel out, int length) throws IOException;
 
+    /**
+     * Transfers this buffer's data starting at the specified absolute {@code index}
+     * to the specified channel starting at the given file position.
+     * This method does not modify {@code readerIndex} or {@code writerIndex} of
+     * this buffer. This method does not modify the channel's position.
+     *
+     * @param position the file position at which the transfer is to begin
+     * @param length the maximum number of bytes to transfer
+     *
+     * @return the actual number of bytes written out to the specified channel
+     *
+     * @throws IndexOutOfBoundsException
+     *         if the specified {@code index} is less than {@code 0} or
+     *         if {@code index + length} is greater than
+     *            {@code this.capacity}
+     * @throws IOException
+     *         if the specified channel threw an exception during I/O
+     */
+    public abstract int getBytes(int index, FileChannel out, long position, int length) throws IOException;
+
     /**
      * Sets the specified boolean at the specified absolute {@code index} in this
      * buffer.
@@ -1178,7 +1199,27 @@ public abstract class ByteBuf implements ReferenceCounted, Comparable<ByteBuf> {
      * @throws IOException
      *         if the specified channel threw an exception during I/O
      */
-    public abstract int  setBytes(int index, ScatteringByteChannel in, int length) throws IOException;
+    public abstract int setBytes(int index, ScatteringByteChannel in, int length) throws IOException;
+
+    /**
+     * Transfers the content of the specified source channel starting at the given file position
+     * to this buffer starting at the specified absolute {@code index}.
+     * This method does not modify {@code readerIndex} or {@code writerIndex} of
+     * this buffer. This method does not modify the channel's position.
+     *
+     * @param position the file position at which the transfer is to begin
+     * @param length the maximum number of bytes to transfer
+     *
+     * @return the actual number of bytes read in from the specified channel.
+     *         {@code -1} if the specified channel is closed.
+     *
+     * @throws IndexOutOfBoundsException
+     *         if the specified {@code index} is less than {@code 0} or
+     *         if {@code index + length} is greater than {@code this.capacity}
+     * @throws IOException
+     *         if the specified channel threw an exception during I/O
+     */
+    public abstract int setBytes(int index, FileChannel in, long position, int length) throws IOException;
 
     /**
      * Fills this buffer with <tt>NUL (0x00)</tt> starting at the specified
@@ -1524,7 +1565,24 @@ public abstract class ByteBuf implements ReferenceCounted, Comparable<ByteBuf> {
      * @throws IOException
      *         if the specified channel threw an exception during I/O
      */
-    public abstract int  readBytes(GatheringByteChannel out, int length) throws IOException;
+    public abstract int readBytes(GatheringByteChannel out, int length) throws IOException;
+
+    /**
+     * Transfers this buffer's data starting at the current {@code readerIndex}
+     * to the specified channel starting at the given file position.
+     * This method does not modify the channel's position.
+     *
+     * @param position the file position at which the transfer is to begin
+     * @param length the maximum number of bytes to transfer
+     *
+     * @return the actual number of bytes written out to the specified channel
+     *
+     * @throws IndexOutOfBoundsException
+     *         if {@code length} is greater than {@code this.readableBytes}
+     * @throws IOException
+     *         if the specified channel threw an exception during I/O
+     */
+    public abstract int readBytes(FileChannel out, long position, int length) throws IOException;
 
     /**
      * Increases the current {@code readerIndex} by the specified
@@ -1783,7 +1841,25 @@ public abstract class ByteBuf implements ReferenceCounted, Comparable<ByteBuf> {
      * @throws IOException
      *         if the specified channel threw an exception during I/O
      */
-    public abstract int  writeBytes(ScatteringByteChannel in, int length) throws IOException;
+    public abstract int writeBytes(ScatteringByteChannel in, int length) throws IOException;
+
+    /**
+     * Transfers the content of the specified channel starting at the given file position
+     * to this buffer starting at the current {@code writerIndex} and increases the
+     * {@code writerIndex} by the number of the transferred bytes.
+     * This method does not modify the channel's position.
+     *
+     * @param position the file position at which the transfer is to begin
+     * @param length the maximum number of bytes to transfer
+     *
+     * @return the actual number of bytes read in from the specified channel
+     *
+     * @throws IndexOutOfBoundsException
+     *         if {@code length} is greater than {@code this.writableBytes}
+     * @throws IOException
+     *         if the specified channel threw an exception during I/O
+     */
+    public abstract int writeBytes(FileChannel in, long position, int length) throws IOException;
 
     /**
      * Fills this buffer with <tt>NUL (0x00)</tt> starting at the current

diff --git a/buffer/src/main/java/io/netty/buffer/CompositeByteBuf.java b/buffer/src/main/java/io/netty/buffer/CompositeByteBuf.java
@@ -22,6 +22,7 @@
 import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
+import java.nio.channels.FileChannel;
 import java.nio.channels.GatheringByteChannel;
 import java.nio.channels.ScatteringByteChannel;
 import java.util.ArrayList;
@@ -877,6 +878,24 @@ public int getBytes(int index, GatheringByteChannel out, int length)
         }
     }
 
+    @Override
+    public int getBytes(int index, FileChannel out, long position, int length)
+            throws IOException {
+        int count = nioBufferCount();
+        if (count == 1) {
+            return out.write(internalNioBuffer(index, length), position);
+        } else {
+            long writtenBytes = 0;
+            for (ByteBuffer buf : nioBuffers(index, length)) {
+                writtenBytes += out.write(buf, position + writtenBytes);
+            }
+            if (writtenBytes > Integer.MAX_VALUE) {
+                return Integer.MAX_VALUE;
+            }
+            return (int) writtenBytes;
+        }
+    }
+
     @Override
     public CompositeByteBuf getBytes(int index, OutputStream out, int length) throws IOException {
         checkIndex(index, length);
@@ -1130,6 +1149,11 @@ public int setBytes(int index, InputStream in, int length) throws IOException {
             ByteBuf s = c.buf;
             int adjustment = c.offset;
             int localLength = Math.min(length, s.capacity() - (index - adjustment));
+            if (localLength == 0) {
+                // Skip empty buffer
+                i++;
+                continue;
+            }
             int localReadBytes = s.setBytes(index - adjustment, in, localLength);
             if (localReadBytes < 0) {
                 if (readBytes == 0) {
@@ -1168,6 +1192,11 @@ public int setBytes(int index, ScatteringByteChannel in, int length) throws IOEx
             ByteBuf s = c.buf;
             int adjustment = c.offset;
             int localLength = Math.min(length, s.capacity() - (index - adjustment));
+            if (localLength == 0) {
+                // Skip empty buffer
+                i++;
+                continue;
+            }
             int localReadBytes = s.setBytes(index - adjustment, in, localLength);
 
             if (localReadBytes == 0) {
@@ -1197,6 +1226,54 @@ public int setBytes(int index, ScatteringByteChannel in, int length) throws IOEx
         return readBytes;
     }
 
+    @Override
+    public int setBytes(int index, FileChannel in, long position, int length) throws IOException {
+        checkIndex(index, length);
+        if (length == 0) {
+            return in.read(EMPTY_NIO_BUFFER, position);
+        }
+
+        int i = toComponentIndex(index);
+        int readBytes = 0;
+        do {
+            Component c = components.get(i);
+            ByteBuf s = c.buf;
+            int adjustment = c.offset;
+            int localLength = Math.min(length, s.capacity() - (index - adjustment));
+            if (localLength == 0) {
+                // Skip empty buffer
+                i++;
+                continue;
+            }
+            int localReadBytes = s.setBytes(index - adjustment, in, position + readBytes, localLength);
+
+            if (localReadBytes == 0) {
+                break;
+            }
+
+            if (localReadBytes < 0) {
+                if (readBytes == 0) {
+                    return -1;
+                } else {
+                    break;
+                }
+            }
+
+            if (localReadBytes == localLength) {
+                index += localLength;
+                length -= localLength;
+                readBytes += localLength;
+                i ++;
+            } else {
+                index += localReadBytes;
+                length -= localReadBytes;
+                readBytes += localReadBytes;
+            }
+        } while (length > 0);
+
+        return readBytes;
+    }
+
     @Override
     public ByteBuf copy(int index, int length) {
         checkIndex(index, length);

diff --git a/buffer/src/main/java/io/netty/buffer/DuplicatedByteBuf.java b/buffer/src/main/java/io/netty/buffer/DuplicatedByteBuf.java
@@ -22,6 +22,7 @@
 import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
+import java.nio.channels.FileChannel;
 import java.nio.channels.GatheringByteChannel;
 import java.nio.channels.ScatteringByteChannel;
 
@@ -309,6 +310,12 @@ public int getBytes(int index, GatheringByteChannel out, int length)
         return buffer.getBytes(index, out, length);
     }
 
+    @Override
+    public int getBytes(int index, FileChannel out, long position, int length)
+            throws IOException {
+        return buffer.getBytes(index, out, position, length);
+    }
+
     @Override
     public int setBytes(int index, InputStream in, int length)
             throws IOException {
@@ -321,6 +328,12 @@ public int setBytes(int index, ScatteringByteChannel in, int length)
         return buffer.setBytes(index, in, length);
     }
 
+    @Override
+    public int setBytes(int index, FileChannel in, long position, int length)
+            throws IOException {
+        return buffer.setBytes(index, in, position, length);
+    }
+
     @Override
     public int nioBufferCount() {
         return buffer.nioBufferCount();

diff --git a/buffer/src/main/java/io/netty/buffer/EmptyByteBuf.java b/buffer/src/main/java/io/netty/buffer/EmptyByteBuf.java
@@ -26,6 +26,7 @@
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.nio.ReadOnlyBufferException;
+import java.nio.channels.FileChannel;
 import java.nio.channels.GatheringByteChannel;
 import java.nio.channels.ScatteringByteChannel;
 import java.nio.charset.Charset;
@@ -374,6 +375,12 @@ public int getBytes(int index, GatheringByteChannel out, int length) {
         return 0;
     }
 
+    @Override
+    public int getBytes(int index, FileChannel out, long position, int length) {
+        checkIndex(index, length);
+        return 0;
+    }
+
     @Override
     public ByteBuf setBoolean(int index, boolean value) {
         throw new IndexOutOfBoundsException();
@@ -481,6 +488,12 @@ public int setBytes(int index, ScatteringByteChannel in, int length) {
         return 0;
     }
 
+    @Override
+    public int setBytes(int index, FileChannel in, long position, int length) {
+        checkIndex(index, length);
+        return 0;
+    }
+
     @Override
     public ByteBuf setZero(int index, int length) {
         return checkIndex(index, length);
@@ -637,6 +650,12 @@ public int readBytes(GatheringByteChannel out, int length) {
         return 0;
     }
 
+    @Override
+    public int readBytes(FileChannel out, long position, int length) {
+        checkLength(length);
+        return 0;
+    }
+
     @Override
     public ByteBuf skipBytes(int length) {
         return checkLength(length);
@@ -749,6 +768,12 @@ public int writeBytes(ScatteringByteChannel in, int length) {
         return 0;
     }
 
+    @Override
+    public int writeBytes(FileChannel in, long position, int length) {
+        checkLength(length);
+        return 0;
+    }
+
     @Override
     public ByteBuf writeZero(int length) {
         return checkLength(length);