src/java/org/apache/cassandra/io/util/RandomAccessReader.java - cassandra - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.cassandra.io.util;

 import java.io.*;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;

 import com.google.common.primitives.Ints;
 import com.google.common.util.concurrent.RateLimiter;

 import org.apache.cassandra.io.FSReadError;
 import org.apache.cassandra.io.compress.BufferType;
 import org.apache.cassandra.utils.memory.BufferPool;

 public class RandomAccessReader extends RebufferingInputStream implements FileDataInput
 {
     // The default buffer size when the client doesn't specify it
     public static final int DEFAULT_BUFFER_SIZE = 4096;

     // The maximum buffer size, we will never buffer more than this size. Further,
     // when the limiter is not null, i.e. when throttling is enabled, we read exactly
     // this size, since when throttling the intention is to eventually read everything,
     // see CASSANDRA-8630
     // NOTE: this size is chosen both for historical consistency, as a reasonable upper bound,
     //       and because our BufferPool currently has a maximum allocation size of this.
     public static final int MAX_BUFFER_SIZE = 1 << 16; // 64k

     // the IO channel to the file, we do not own a reference to this due to
     // performance reasons (CASSANDRA-9379) so it's up to the owner of the RAR to
     // ensure that the channel stays open and that it is closed afterwards
     protected final ChannelProxy channel;

     // optional memory mapped regions for the channel
     protected final MmappedRegions regions;

     // An optional limiter that will throttle the amount of data we read
     protected final RateLimiter limiter;

     // the file length, this can be overridden at construction to a value shorter
     // than the true length of the file; if so, it acts as an imposed limit on reads,
     // required when opening sstables early not to read past the mark
     private final long fileLength;

     // the buffer size for buffered readers
     protected final int bufferSize;

     // the buffer type for buffered readers
     protected final BufferType bufferType;

     // offset from the beginning of the file
     protected long bufferOffset;

     // offset of the last file mark
     protected long markedPointer;

     protected RandomAccessReader(Builder builder)
     {
         super(builder.createBuffer());

         this.channel = builder.channel;
         this.regions = builder.regions;
         this.limiter = builder.limiter;
         this.fileLength = builder.overrideLength <= 0 ? builder.channel.size() : builder.overrideLength;
         this.bufferSize = builder.bufferSize;
         this.bufferType = builder.bufferType;
         this.buffer = builder.buffer;
     }

     protected static ByteBuffer allocateBuffer(int size, BufferType bufferType)
     {
         return BufferPool.get(size, bufferType).order(ByteOrder.BIG_ENDIAN);
     }

     protected void releaseBuffer()
     {
         if (buffer != null)
         {
             if (regions == null)
                 BufferPool.put(buffer);
             buffer = null;
         }
     }

     /**
      * Read data from file starting from current currentOffset to populate buffer.
      */
     public void reBuffer()
     {
         if (isEOF())
             return;

         if (regions == null)
             reBufferStandard();
         else
             reBufferMmap();

         if (limiter != null)
             limiter.acquire(buffer.remaining());

         assert buffer.order() == ByteOrder.BIG_ENDIAN : "Buffer must have BIG ENDIAN byte ordering";
     }

     protected void reBufferStandard()
     {
         bufferOffset += buffer.position();
         assert bufferOffset < fileLength;

         buffer.clear();
         long position = bufferOffset;
         long limit = bufferOffset;

         long pageAligedPos = position & ~4095;
         // Because the buffer capacity is a multiple of the page size, we read less
         // the first time and then we should read at page boundaries only,
         // unless the user seeks elsewhere
         long upperLimit = Math.min(fileLength, pageAligedPos + buffer.capacity());
         buffer.limit((int)(upperLimit - position));
         while (buffer.hasRemaining() && limit < upperLimit)
         {
             int n = channel.read(buffer, position);
             if (n < 0)
                 throw new FSReadError(new IOException("Unexpected end of file"), channel.filePath());

             position += n;
             limit = bufferOffset + buffer.position();
         }

         buffer.flip();
     }

     protected void reBufferMmap()
     {
         long position = bufferOffset + buffer.position();
         assert position < fileLength;

         MmappedRegions.Region region = regions.floor(position);
         bufferOffset = region.bottom();
         buffer = region.buffer.duplicate();
         buffer.position(Ints.checkedCast(position - bufferOffset));

         if (limiter != null && bufferSize < buffer.remaining())
         { // ensure accurate throttling
             buffer.limit(buffer.position() + bufferSize);
         }
     }

     @Override
     public long getFilePointer()
     {
         return current();
     }

     protected long current()
     {
         return bufferOffset + (buffer == null ? 0 : buffer.position());
     }

     public String getPath()
     {
         return channel.filePath();
     }

     public ChannelProxy getChannel()
     {
         return channel;
     }

     @Override
     public void reset() throws IOException
     {
         seek(markedPointer);
     }

     @Override
     public boolean markSupported()
     {
         return true;
     }

     public long bytesPastMark()
     {
         long bytes = current() - markedPointer;
         assert bytes >= 0;
         return bytes;
     }

     public DataPosition mark()
     {
         markedPointer = current();
         return new BufferedRandomAccessFileMark(markedPointer);
     }

     public void reset(DataPosition mark)
     {
         assert mark instanceof BufferedRandomAccessFileMark;
         seek(((BufferedRandomAccessFileMark) mark).pointer);
     }

     public long bytesPastMark(DataPosition mark)
     {
         assert mark instanceof BufferedRandomAccessFileMark;
         long bytes = current() - ((BufferedRandomAccessFileMark) mark).pointer;
         assert bytes >= 0;
         return bytes;
     }

     /**
      * @return true if there is no more data to read
      */
     public boolean isEOF()
     {
         return current() == length();
     }

     public long bytesRemaining()
     {
         return length() - getFilePointer();
     }

     @Override
     public int available() throws IOException
     {
         return Ints.saturatedCast(bytesRemaining());
     }

     @Override
     public void close()
     {
 	    //make idempotent
         if (buffer == null)
             return;

         bufferOffset += buffer.position();
         releaseBuffer();

         //For performance reasons we don't keep a reference to the file
         //channel so we don't close it
     }

     @Override
     public String toString()
     {
         return getClass().getSimpleName() + "(filePath='" + channel + "')";
     }

     /**
      * Class to hold a mark to the position of the file
      */
     protected static class BufferedRandomAccessFileMark implements DataPosition
     {
         final long pointer;

         public BufferedRandomAccessFileMark(long pointer)
         {
             this.pointer = pointer;
         }
     }

     @Override
     public void seek(long newPosition)
     {
         if (newPosition < 0)
             throw new IllegalArgumentException("new position should not be negative");

         if (buffer == null)
             throw new IllegalStateException("Attempted to seek in a closed RAR");

         if (newPosition >= length()) // it is save to call length() in read-only mode
         {
             if (newPosition > length())
                 throw new IllegalArgumentException(String.format("Unable to seek to position %d in %s (%d bytes) in read-only mode",
                                                              newPosition, getPath(), length()));
             buffer.limit(0);
             bufferOffset = newPosition;
             return;
         }

         if (newPosition >= bufferOffset && newPosition < bufferOffset + buffer.limit())
         {
             buffer.position((int) (newPosition - bufferOffset));
             return;
         }
         // Set current location to newPosition and clear buffer so reBuffer calculates from newPosition
         bufferOffset = newPosition;
         buffer.clear();
         reBuffer();
         assert current() == newPosition;
     }

     /**
      * Reads a line of text form the current position in this file. A line is
      * represented by zero or more characters followed by {@code '\n'}, {@code
      * '\r'}, {@code "\r\n"} or the end of file marker. The string does not
      * include the line terminating sequence.
      * <p/>
      * Blocks until a line terminating sequence has been read, the end of the
      * file is reached or an exception is thrown.
      *
      * @return the contents of the line or {@code null} if no characters have
      * been read before the end of the file has been reached.
      * @throws IOException if this file is closed or another I/O error occurs.
      */
     public final String readLine() throws IOException
     {
         StringBuilder line = new StringBuilder(80); // Typical line length
         boolean foundTerminator = false;
         long unreadPosition = -1;
         while (true)
         {
             int nextByte = read();
             switch (nextByte)
             {
                 case -1:
                     return line.length() != 0 ? line.toString() : null;
                 case (byte) '\r':
                     if (foundTerminator)
                     {
                         seek(unreadPosition);
                         return line.toString();
                     }
                     foundTerminator = true;
                     /* Have to be able to peek ahead one byte */
                     unreadPosition = getPosition();
                     break;
                 case (byte) '\n':
                     return line.toString();
                 default:
                     if (foundTerminator)
                     {
                         seek(unreadPosition);
                         return line.toString();
                     }
                     line.append((char) nextByte);
             }
         }
     }

     public long length()
     {
         return fileLength;
     }

     public long getPosition()
     {
         return current();
     }

     public static class Builder
     {
         // The NIO file channel or an empty channel
         public final ChannelProxy channel;

         // We override the file length when we open sstables early, so that we do not
         // read past the early mark
         public long overrideLength;

         // The size of the buffer for buffered readers
         public int bufferSize;

         // The type of the buffer for buffered readers
         public BufferType bufferType;

         // The buffer
         public ByteBuffer buffer;

         // The mmap segments for mmap readers
         public MmappedRegions regions;

         // An optional limiter that will throttle the amount of data we read
         public RateLimiter limiter;

         public Builder(ChannelProxy channel)
         {
             this.channel = channel;
             this.overrideLength = -1L;
             this.bufferSize = DEFAULT_BUFFER_SIZE;
             this.bufferType = BufferType.OFF_HEAP;
             this.regions = null;
             this.limiter = null;
         }

         /** The buffer size is typically already page aligned but if that is not the case
          * make sure that it is a multiple of the page size, 4096. Also limit it to the maximum
          * buffer size unless we are throttling, in which case we may as well read the maximum
          * directly since the intention is to read the full file, see CASSANDRA-8630.
          * */
         private void setBufferSize()
         {
             if (limiter != null)
             {
                 bufferSize = MAX_BUFFER_SIZE;
                 return;
             }

             if ((bufferSize & ~4095) != bufferSize)
             { // should already be a page size multiple but if that's not case round it up
                 bufferSize = (bufferSize + 4095) & ~4095;
             }

             bufferSize = Math.min(MAX_BUFFER_SIZE, bufferSize);
         }

         protected ByteBuffer createBuffer()
         {
             setBufferSize();

             buffer = regions == null
                      ? allocateBuffer(bufferSize, bufferType)
                      : regions.floor(0).buffer.duplicate();

             buffer.limit(0);
             return buffer;
         }

         public Builder overrideLength(long overrideLength)
         {
             this.overrideLength = overrideLength;
             return this;
         }

         public Builder bufferSize(int bufferSize)
         {
             if (bufferSize <= 0)
                 throw new IllegalArgumentException("bufferSize must be positive");

             this.bufferSize = bufferSize;
             return this;
         }

         public Builder bufferType(BufferType bufferType)
         {
             this.bufferType = bufferType;
             return this;
         }

         public Builder regions(MmappedRegions regions)
         {
             this.regions = regions;
             return this;
         }

         public Builder limiter(RateLimiter limiter)
         {
             this.limiter = limiter;
             return this;
         }

         public RandomAccessReader build()
         {
             return new RandomAccessReader(this);
         }

         public RandomAccessReader buildWithChannel()
         {
             return new RandomAccessReaderWithOwnChannel(this);
         }
     }

     // A wrapper of the RandomAccessReader that closes the channel when done.
     // For performance reasons RAR does not increase the reference count of
     // a channel but assumes the owner will keep it open and close it,
     // see CASSANDRA-9379, this thin class is just for those cases where we do
     // not have a shared channel.
     public static class RandomAccessReaderWithOwnChannel extends RandomAccessReader
     {
         protected RandomAccessReaderWithOwnChannel(Builder builder)
         {
             super(builder);
         }

         @Override
         public void close()
         {
             try
             {
                 super.close();
             }
             finally
             {
                 channel.close();
             }
         }
     }

     @SuppressWarnings("resource")
     public static RandomAccessReader open(File file)
     {
         return new Builder(new ChannelProxy(file)).buildWithChannel();
     }

     public static RandomAccessReader open(ChannelProxy channel)
     {
         return new Builder(channel).build();
     }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.cassandra.io.util;

	import java.io.*;
	import java.nio.ByteBuffer;
	import java.nio.ByteOrder;

	import com.google.common.primitives.Ints;
	import com.google.common.util.concurrent.RateLimiter;

	import org.apache.cassandra.io.FSReadError;
	import org.apache.cassandra.io.compress.BufferType;
	import org.apache.cassandra.utils.memory.BufferPool;

	public class RandomAccessReader extends RebufferingInputStream implements FileDataInput
	{
	// The default buffer size when the client doesn't specify it
	public static final int DEFAULT_BUFFER_SIZE = 4096;

	// The maximum buffer size, we will never buffer more than this size. Further,
	// when the limiter is not null, i.e. when throttling is enabled, we read exactly
	// this size, since when throttling the intention is to eventually read everything,
	// see CASSANDRA-8630
	// NOTE: this size is chosen both for historical consistency, as a reasonable upper bound,
	// and because our BufferPool currently has a maximum allocation size of this.
	public static final int MAX_BUFFER_SIZE = 1 << 16; // 64k

	// the IO channel to the file, we do not own a reference to this due to
	// performance reasons (CASSANDRA-9379) so it's up to the owner of the RAR to
	// ensure that the channel stays open and that it is closed afterwards
	protected final ChannelProxy channel;

	// optional memory mapped regions for the channel
	protected final MmappedRegions regions;

	// An optional limiter that will throttle the amount of data we read
	protected final RateLimiter limiter;

	// the file length, this can be overridden at construction to a value shorter
	// than the true length of the file; if so, it acts as an imposed limit on reads,
	// required when opening sstables early not to read past the mark
	private final long fileLength;

	// the buffer size for buffered readers
	protected final int bufferSize;

	// the buffer type for buffered readers
	protected final BufferType bufferType;

	// offset from the beginning of the file
	protected long bufferOffset;

	// offset of the last file mark
	protected long markedPointer;

	protected RandomAccessReader(Builder builder)
	{
	super(builder.createBuffer());

	this.channel = builder.channel;
	this.regions = builder.regions;
	this.limiter = builder.limiter;
	this.fileLength = builder.overrideLength <= 0 ? builder.channel.size() : builder.overrideLength;
	this.bufferSize = builder.bufferSize;
	this.bufferType = builder.bufferType;
	this.buffer = builder.buffer;
	}

	protected static ByteBuffer allocateBuffer(int size, BufferType bufferType)
	{
	return BufferPool.get(size, bufferType).order(ByteOrder.BIG_ENDIAN);
	}

	protected void releaseBuffer()
	{
	if (buffer != null)
	{
	if (regions == null)
	BufferPool.put(buffer);
	buffer = null;
	}
	}

	/**
	* Read data from file starting from current currentOffset to populate buffer.
	*/
	public void reBuffer()
	{
	if (isEOF())
	return;

	if (regions == null)
	reBufferStandard();
	else
	reBufferMmap();

	if (limiter != null)
	limiter.acquire(buffer.remaining());

	assert buffer.order() == ByteOrder.BIG_ENDIAN : "Buffer must have BIG ENDIAN byte ordering";
	}

	protected void reBufferStandard()
	{
	bufferOffset += buffer.position();
	assert bufferOffset < fileLength;

	buffer.clear();
	long position = bufferOffset;
	long limit = bufferOffset;

	long pageAligedPos = position & ~4095;
	// Because the buffer capacity is a multiple of the page size, we read less
	// the first time and then we should read at page boundaries only,
	// unless the user seeks elsewhere
	long upperLimit = Math.min(fileLength, pageAligedPos + buffer.capacity());
	buffer.limit((int)(upperLimit - position));
	while (buffer.hasRemaining() && limit < upperLimit)
	{
	int n = channel.read(buffer, position);
	if (n < 0)
	throw new FSReadError(new IOException("Unexpected end of file"), channel.filePath());

	position += n;
	limit = bufferOffset + buffer.position();
	}

	buffer.flip();
	}

	protected void reBufferMmap()
	{
	long position = bufferOffset + buffer.position();
	assert position < fileLength;

	MmappedRegions.Region region = regions.floor(position);
	bufferOffset = region.bottom();
	buffer = region.buffer.duplicate();
	buffer.position(Ints.checkedCast(position - bufferOffset));

	if (limiter != null && bufferSize < buffer.remaining())
	{ // ensure accurate throttling
	buffer.limit(buffer.position() + bufferSize);
	}
	}

	@Override
	public long getFilePointer()
	{
	return current();
	}

	protected long current()
	{
	return bufferOffset + (buffer == null ? 0 : buffer.position());
	}

	public String getPath()
	{
	return channel.filePath();
	}

	public ChannelProxy getChannel()
	{
	return channel;
	}

	@Override
	public void reset() throws IOException
	{
	seek(markedPointer);
	}

	@Override
	public boolean markSupported()
	{
	return true;
	}

	public long bytesPastMark()
	{
	long bytes = current() - markedPointer;
	assert bytes >= 0;
	return bytes;
	}

	public DataPosition mark()
	{
	markedPointer = current();
	return new BufferedRandomAccessFileMark(markedPointer);
	}

	public void reset(DataPosition mark)
	{
	assert mark instanceof BufferedRandomAccessFileMark;
	seek(((BufferedRandomAccessFileMark) mark).pointer);
	}

	public long bytesPastMark(DataPosition mark)
	{
	assert mark instanceof BufferedRandomAccessFileMark;
	long bytes = current() - ((BufferedRandomAccessFileMark) mark).pointer;
	assert bytes >= 0;
	return bytes;
	}

	/**
	* @return true if there is no more data to read
	*/
	public boolean isEOF()
	{
	return current() == length();
	}

	public long bytesRemaining()
	{
	return length() - getFilePointer();
	}

	@Override
	public int available() throws IOException
	{
	return Ints.saturatedCast(bytesRemaining());
	}

	@Override
	public void close()
	{
	//make idempotent
	if (buffer == null)
	return;

	bufferOffset += buffer.position();
	releaseBuffer();

	//For performance reasons we don't keep a reference to the file
	//channel so we don't close it
	}

	@Override
	public String toString()
	{
	return getClass().getSimpleName() + "(filePath='" + channel + "')";
	}

	/**
	* Class to hold a mark to the position of the file
	*/
	protected static class BufferedRandomAccessFileMark implements DataPosition
	{
	final long pointer;

	public BufferedRandomAccessFileMark(long pointer)
	{
	this.pointer = pointer;
	}
	}

	@Override
	public void seek(long newPosition)
	{
	if (newPosition < 0)
	throw new IllegalArgumentException("new position should not be negative");

	if (buffer == null)
	throw new IllegalStateException("Attempted to seek in a closed RAR");

	if (newPosition >= length()) // it is save to call length() in read-only mode
	{
	if (newPosition > length())
	throw new IllegalArgumentException(String.format("Unable to seek to position %d in %s (%d bytes) in read-only mode",
	newPosition, getPath(), length()));
	buffer.limit(0);
	bufferOffset = newPosition;
	return;
	}

	if (newPosition >= bufferOffset && newPosition < bufferOffset + buffer.limit())
	{
	buffer.position((int) (newPosition - bufferOffset));
	return;
	}
	// Set current location to newPosition and clear buffer so reBuffer calculates from newPosition
	bufferOffset = newPosition;
	buffer.clear();
	reBuffer();
	assert current() == newPosition;
	}

	/**
	* Reads a line of text form the current position in this file. A line is
	* represented by zero or more characters followed by {@code '\n'}, {@code
	* '\r'}, {@code "\r\n"} or the end of file marker. The string does not
	* include the line terminating sequence.
	* <p/>
	* Blocks until a line terminating sequence has been read, the end of the
	* file is reached or an exception is thrown.
	*
	* @return the contents of the line or {@code null} if no characters have
	* been read before the end of the file has been reached.
	* @throws IOException if this file is closed or another I/O error occurs.
	*/
	public final String readLine() throws IOException
	{
	StringBuilder line = new StringBuilder(80); // Typical line length
	boolean foundTerminator = false;
	long unreadPosition = -1;
	while (true)
	{
	int nextByte = read();
	switch (nextByte)
	{
	case -1:
	return line.length() != 0 ? line.toString() : null;
	case (byte) '\r':
	if (foundTerminator)
	{
	seek(unreadPosition);
	return line.toString();
	}
	foundTerminator = true;
	/* Have to be able to peek ahead one byte */
	unreadPosition = getPosition();
	break;
	case (byte) '\n':
	return line.toString();
	default:
	if (foundTerminator)
	{
	seek(unreadPosition);
	return line.toString();
	}
	line.append((char) nextByte);
	}
	}
	}

	public long length()
	{
	return fileLength;
	}

	public long getPosition()
	{
	return current();
	}

	public static class Builder
	{
	// The NIO file channel or an empty channel
	public final ChannelProxy channel;

	// We override the file length when we open sstables early, so that we do not
	// read past the early mark
	public long overrideLength;

	// The size of the buffer for buffered readers
	public int bufferSize;

	// The type of the buffer for buffered readers
	public BufferType bufferType;

	// The buffer
	public ByteBuffer buffer;

	// The mmap segments for mmap readers
	public MmappedRegions regions;

	// An optional limiter that will throttle the amount of data we read
	public RateLimiter limiter;

	public Builder(ChannelProxy channel)
	{
	this.channel = channel;
	this.overrideLength = -1L;
	this.bufferSize = DEFAULT_BUFFER_SIZE;
	this.bufferType = BufferType.OFF_HEAP;
	this.regions = null;
	this.limiter = null;
	}

	/** The buffer size is typically already page aligned but if that is not the case
	* make sure that it is a multiple of the page size, 4096. Also limit it to the maximum
	* buffer size unless we are throttling, in which case we may as well read the maximum
	* directly since the intention is to read the full file, see CASSANDRA-8630.
	* */
	private void setBufferSize()
	{
	if (limiter != null)
	{
	bufferSize = MAX_BUFFER_SIZE;
	return;
	}

	if ((bufferSize & ~4095) != bufferSize)
	{ // should already be a page size multiple but if that's not case round it up
	bufferSize = (bufferSize + 4095) & ~4095;
	}

	bufferSize = Math.min(MAX_BUFFER_SIZE, bufferSize);
	}

	protected ByteBuffer createBuffer()
	{
	setBufferSize();

	buffer = regions == null
	? allocateBuffer(bufferSize, bufferType)
	: regions.floor(0).buffer.duplicate();

	buffer.limit(0);
	return buffer;
	}

	public Builder overrideLength(long overrideLength)
	{
	this.overrideLength = overrideLength;
	return this;
	}

	public Builder bufferSize(int bufferSize)
	{
	if (bufferSize <= 0)
	throw new IllegalArgumentException("bufferSize must be positive");

	this.bufferSize = bufferSize;
	return this;
	}

	public Builder bufferType(BufferType bufferType)
	{
	this.bufferType = bufferType;
	return this;
	}

	public Builder regions(MmappedRegions regions)
	{
	this.regions = regions;
	return this;
	}

	public Builder limiter(RateLimiter limiter)
	{
	this.limiter = limiter;
	return this;
	}

	public RandomAccessReader build()
	{
	return new RandomAccessReader(this);
	}

	public RandomAccessReader buildWithChannel()
	{
	return new RandomAccessReaderWithOwnChannel(this);
	}
	}

	// A wrapper of the RandomAccessReader that closes the channel when done.
	// For performance reasons RAR does not increase the reference count of
	// a channel but assumes the owner will keep it open and close it,
	// see CASSANDRA-9379, this thin class is just for those cases where we do
	// not have a shared channel.
	public static class RandomAccessReaderWithOwnChannel extends RandomAccessReader
	{
	protected RandomAccessReaderWithOwnChannel(Builder builder)
	{
	super(builder);
	}

	@Override
	public void close()
	{
	try
	{
	super.close();
	}
	finally
	{
	channel.close();
	}
	}
	}

	@SuppressWarnings("resource")
	public static RandomAccessReader open(File file)
	{
	return new Builder(new ChannelProxy(file)).buildWithChannel();
	}

	public static RandomAccessReader open(ChannelProxy channel)
	{
	return new Builder(channel).build();
	}
	}