package org.apache.cassandra.streaming.messages;
import java.io.IOException;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
import java.util.Optional;
import org.apache.cassandra.io.sstable.SSTableMultiWriter;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataInputPlus.DataInputStreamPlus;
import org.apache.cassandra.io.util.DataOutputStreamPlus;
import org.apache.cassandra.streaming.StreamReader;
import org.apache.cassandra.streaming.StreamSession;
import org.apache.cassandra.streaming.compress.CompressedStreamReader;
import org.apache.cassandra.utils.JVMStabilityInspector;
import static org.apache.cassandra.utils.Throwables.extractIOExceptionCause;
public class IncomingFileMessage extends StreamMessage
{
public static Serializer<IncomingFileMessage> serializer = new Serializer<IncomingFileMessage>()
{
@SuppressWarnings("resource")
public IncomingFileMessage deserialize(ReadableByteChannel in, int version, StreamSession session) throws IOException
{
DataInputPlus input = new DataInputStreamPlus(Channels.newInputStream(in));
FileMessageHeader header = FileMessageHeader.serializer.deserialize(input, version);
StreamReader reader = !header.isCompressed() ? new StreamReader(header, session)
: new CompressedStreamReader(header, session);
try
{
return new IncomingFileMessage(reader.read(in), header);
}
catch (Throwable t)
{
JVMStabilityInspector.inspectThrowable(t);
throw t;
}
}
public void serialize(IncomingFileMessage message, DataOutputStreamPlus out, int version, StreamSession session)
{
throw new UnsupportedOperationException("Not allowed to call serialize on an incoming file");
}
};
public FileMessageHeader header;
public SSTableMultiWriter sstable;
public IncomingFileMessage(SSTableMultiWriter sstable, FileMessageHeader header)
{
super(Type.FILE);
this.header = header;
this.sstable = sstable;
}
@Override
public String toString()
{
return "File (" + header + ", file: " + sstable.getFilename() + ")";
}
}