package org.apache.cassandra.hints;
import java.io.IOException;
import java.nio.ByteBuffer;
import com.google.common.annotations.VisibleForTesting;
import org.apache.cassandra.hints.ChecksummedDataInput.Position;
import org.apache.cassandra.io.FSReadError;
import org.apache.cassandra.io.compress.ICompressor;
import org.apache.cassandra.io.util.ChannelProxy;
import org.apache.cassandra.utils.memory.BufferPool;
public final class CompressedChecksummedDataInput extends ChecksummedDataInput
{
private final ICompressor compressor;
private volatile long filePosition = 0;
private volatile long sourcePosition = 0;
private volatile ByteBuffer compressedBuffer = null;
private final ByteBuffer metadataBuffer = ByteBuffer.allocate(CompressedHintsWriter.METADATA_SIZE);
public CompressedChecksummedDataInput(ChannelProxy channel, ICompressor compressor, long filePosition)
{
super(channel, compressor.preferredBufferType());
this.compressor = compressor;
this.sourcePosition = this.filePosition = filePosition;
}
public boolean isEOF()
{
return filePosition == channel.size() && buffer.remaining() == 0;
}
public long getSourcePosition()
{
return sourcePosition;
}
static class Position extends ChecksummedDataInput.Position
{
final long bufferStart;
final int bufferPosition;
public Position(long sourcePosition, long bufferStart, int bufferPosition)
{
super(sourcePosition);
this.bufferStart = bufferStart;
this.bufferPosition = bufferPosition;
}
@Override
public long subtract(InputPosition o)
{
Position other = (Position) o;
return bufferStart - other.bufferStart + bufferPosition - other.bufferPosition;
}
}
public InputPosition getSeekPosition()
{
return new Position(sourcePosition, bufferOffset, buffer.position());
}
public void seek(InputPosition p)
{
Position pos = (Position) p;
bufferOffset = pos.bufferStart;
filePosition = pos.sourcePosition;
buffer.position(0).limit(0);
resetCrc();
reBuffer();
buffer.position(pos.bufferPosition);
assert sourcePosition == pos.sourcePosition;
assert bufferOffset == pos.bufferStart;
assert buffer.position() == pos.bufferPosition;
}
@Override
protected void readBuffer()
{
sourcePosition = filePosition;
if (isEOF())
return;
metadataBuffer.clear();
channel.read(metadataBuffer, filePosition);
filePosition += CompressedHintsWriter.METADATA_SIZE;
metadataBuffer.rewind();
int uncompressedSize = metadataBuffer.getInt();
int compressedSize = metadataBuffer.getInt();
if (compressedBuffer == null || compressedSize > compressedBuffer.capacity())
{
int bufferSize = compressedSize + (compressedSize / 20);
if (compressedBuffer != null)
{
BufferPool.put(compressedBuffer);
}
compressedBuffer = BufferPool.get(bufferSize, compressor.preferredBufferType());
}
compressedBuffer.clear();
compressedBuffer.limit(compressedSize);
channel.read(compressedBuffer, filePosition);
compressedBuffer.rewind();
filePosition += compressedSize;
if (buffer.capacity() < uncompressedSize)
{
int bufferSize = uncompressedSize + (uncompressedSize / 20);
BufferPool.put(buffer);
buffer = BufferPool.get(bufferSize, compressor.preferredBufferType());
}
buffer.clear();
buffer.limit(uncompressedSize);
try
{
compressor.uncompress(compressedBuffer, buffer);
buffer.flip();
}
catch (IOException e)
{
throw new FSReadError(e, getPath());
}
}
@Override
public void close()
{
BufferPool.put(compressedBuffer);
super.close();
}
@SuppressWarnings("resource")
public static ChecksummedDataInput upgradeInput(ChecksummedDataInput input, ICompressor compressor)
{
long position = input.getPosition();
input.close();
return new CompressedChecksummedDataInput(new ChannelProxy(input.getPath()), compressor, position);
}
@VisibleForTesting
ICompressor getCompressor()
{
return compressor;
}
}