import org.apache.cassandra.io.util.DataOutputStreamPlus; //导入依赖的package包/类
/**
 * Closes the index and bloomfilter, making the public state of this writer valid for consumption.
 */
void flushBf()
{
    if (components.contains(Component.FILTER))
    {
        String path = descriptor.filenameFor(Component.FILTER);

        try (HadoopFileUtils.HadoopFileChannel hos = HadoopFileUtils.newFilesystemChannel(path,
                                                                              descriptor.getConfiguration());
             DataOutputStreamPlus stream = new BufferedDataOutputStreamPlus(hos))
        {
            // bloom filter
            FilterFactory.serialize(bf, stream);
            stream.flush();
            //SyncUtil.sync(hos);
        }
        catch (IOException e)
        {
            logger.info(e.getMessage());
            throw new FSWriteError(e, path);
        }
    }
}
 

import org.apache.cassandra.io.util.DataOutputStreamPlus; //导入依赖的package包/类
public synchronized void serialize(DataOutputStreamPlus out, int version, StreamSession session) throws IOException
{
    if (completed)
    {
        return;
    }

    CompressionInfo compressionInfo = FileMessageHeader.serializer.serialize(header, out, version);

    final SSTableReader reader = ref.get();
    StreamWriter writer = compressionInfo == null ?
                                  new StreamWriter(reader, header.sections, session) :
                                  new CompressedStreamWriter(reader, header.sections,
                                                             compressionInfo, session);
    writer.write(out);
}
 

import org.apache.cassandra.io.util.DataOutputStreamPlus; //导入依赖的package包/类
@SuppressWarnings("resource")
public void sendInitMessage(Socket socket, boolean isForOutgoing) throws IOException
{
    StreamInitMessage message = new StreamInitMessage(
            FBUtilities.getBroadcastAddress(),
            session.sessionIndex(),
            session.planId(),
            session.description(),
            isForOutgoing,
            session.keepSSTableLevel(),
            session.isIncremental());
    ByteBuffer messageBuf = message.createMessage(false, protocolVersion);
    DataOutputStreamPlus out = getWriteChannel(socket);
    out.write(messageBuf);
    out.flush();
}
 

import org.apache.cassandra.io.util.DataOutputStreamPlus; //导入依赖的package包/类
static void hugeBFSerialization(boolean oldBfHashOrder) throws IOException
{
    ByteBuffer test = ByteBuffer.wrap(new byte[] {0, 1});

    File file = FileUtils.createTempFile("bloomFilterTest-", ".dat");
    BloomFilter filter = (BloomFilter) FilterFactory.getFilter(((long) Integer.MAX_VALUE / 8) + 1, 0.01d, true, oldBfHashOrder);
    filter.add(FilterTestHelper.wrap(test));
    DataOutputStreamPlus out = new BufferedDataOutputStreamPlus(new FileOutputStream(file));
    FilterFactory.serialize(filter, out);
    filter.bitset.serialize(out);
    out.close();
    filter.close();

    DataInputStream in = new DataInputStream(new FileInputStream(file));
    BloomFilter filter2 = (BloomFilter) FilterFactory.deserialize(in, true, oldBfHashOrder);
    Assert.assertTrue(filter2.isPresent(FilterTestHelper.wrap(test)));
    FileUtils.closeQuietly(in);
    filter2.close();
}
 

import org.apache.cassandra.io.util.DataOutputStreamPlus; //导入依赖的package包/类
private static void testEstimatedHistogramWrite() throws IOException
{
    EstimatedHistogram hist0 = new EstimatedHistogram();
    EstimatedHistogram hist1 = new EstimatedHistogram(5000);
    long[] offsets = new long[1000];
    long[] data = new long[offsets.length + 1];
    for (int i = 0; i < offsets.length; i++)
    {
        offsets[i] = i;
        data[i] = 10 * i;
    }
    data[offsets.length] = 100000;
    EstimatedHistogram hist2 = new EstimatedHistogram(offsets, data);

    try (DataOutputStreamPlus out = getOutput("utils.EstimatedHistogram.bin"))
    {
        EstimatedHistogram.serializer.serialize(hist0, out);
        EstimatedHistogram.serializer.serialize(hist1, out);
        EstimatedHistogram.serializer.serialize(hist2, out);
    }
}
 

import org.apache.cassandra.io.util.DataOutputStreamPlus; //导入依赖的package包/类
private void rewriteSSTableMetadata(Descriptor descriptor, Map<MetadataType, MetadataComponent> currentComponents) throws IOException
{
    String filePath = descriptor.tmpFilenameFor(Component.STATS);
    try (DataOutputStreamPlus out = new BufferedDataOutputStreamPlus(new FileOutputStream(filePath)))
    {
        serialize(currentComponents, out, descriptor.version);
        out.flush();
    }
    // we cant move a file on top of another file in windows:
    if (FBUtilities.isWindows)
        FileUtils.delete(descriptor.filenameFor(Component.STATS));
    FileUtils.renameWithConfirm(filePath, descriptor.filenameFor(Component.STATS));

}
 

import org.apache.cassandra.io.util.DataOutputStreamPlus; //导入依赖的package包/类
/**
 * Save index summary to Summary.db file.
 */
public static void saveSummary(Descriptor descriptor, DecoratedKey first, DecoratedKey last, IndexSummary summary)
{

    String filePath = descriptor.filenameFor(Component.SUMMARY);

    //TODO: add a retry here on deletion
    HadoopFileUtils.deleteIfExists(filePath, descriptor.getConfiguration());


    //TODO: will make the retry nicer
    int attempt = 0;
    int maxAttempt = 5;
    boolean isSuccess = false;
    while (!isSuccess) {
        if (attempt > 0)
            FBUtilities.sleepQuietly((int) Math.round(Math.pow(2, attempt)) * 1000);

        try (HadoopFileUtils.HadoopFileChannel hos = HadoopFileUtils.newFilesystemChannel(filePath,
                                                                                  descriptor.getConfiguration());
             DataOutputStreamPlus oStream = new BufferedDataOutputStreamPlus(hos)) {
            IndexSummary.serializer.serialize(summary, oStream, descriptor.version.hasSamplingLevel());
            if (first != null && last != null) {
                ByteBufferUtil.writeWithLength(first.getKey(), oStream);
                ByteBufferUtil.writeWithLength(last.getKey(), oStream);
            }
            isSuccess = true;
        } catch (Throwable e) {
            logger.trace("Cannot save SSTable Summary: ", e);

            // corrupted hence delete it and let it load it now.
            HadoopFileUtils.deleteIfExists(filePath, descriptor.getConfiguration());
            attempt++;
            if (attempt == maxAttempt) //TODO: do we need to record all retried excpetions here or assume they'r same
                throw new RuntimeException("Have retried for " + maxAttempt + " times but still failed!", e);
        }
    }
}
 

import org.apache.cassandra.io.util.DataOutputStreamPlus; //导入依赖的package包/类
@Override
public void write(DataOutputStreamPlus out) throws IOException
{
    long totalSize = totalSize();
    logger.debug("[Stream #{}] Start streaming file {} to {}, repairedAt = {}, totalSize = {}", session.planId(),
                 sstable.getFilename(), session.peer, sstable.getSSTableMetadata().repairedAt, totalSize);
    try (ChannelProxy fc = sstable.getDataChannel().sharedCopy())
    {
        long progress = 0L;
        // calculate chunks to transfer. we want to send continuous chunks altogether.
        List<Pair<Long, Long>> sections = getTransferSections(compressionInfo.chunks);

        int sectionIdx = 0;

        // stream each of the required sections of the file
        for (final Pair<Long, Long> section : sections)
        {
            // length of the section to stream
            long length = section.right - section.left;

            logger.trace("[Stream #{}] Writing section {} with length {} to stream.", session.planId(), sectionIdx++, length);

            // tracks write progress
            long bytesTransferred = 0;
            while (bytesTransferred < length)
            {
                final long bytesTransferredFinal = bytesTransferred;
                final int toTransfer = (int) Math.min(CHUNK_SIZE, length - bytesTransferred);
                limiter.acquire(toTransfer);
                long lastWrite = out.applyToChannel((wbc) -> fc.transferTo(section.left + bytesTransferredFinal, toTransfer, wbc));
                bytesTransferred += lastWrite;
                progress += lastWrite;
                session.progress(sstable.descriptor, ProgressInfo.Direction.OUT, progress, totalSize);
            }
        }
        logger.debug("[Stream #{}] Finished streaming file {} to {}, bytesTransferred = {}, totalSize = {}",
                     session.planId(), sstable.getFilename(), session.peer, progress, totalSize);
    }
}
 

import org.apache.cassandra.io.util.DataOutputStreamPlus; //导入依赖的package包/类
public void serialize(PrepareMessage message, DataOutputStreamPlus out, int version, StreamSession session) throws IOException
{
    // requests
    out.writeInt(message.requests.size());
    for (StreamRequest request : message.requests)
        StreamRequest.serializer.serialize(request, out, version);
    // summaries
    out.writeInt(message.summaries.size());
    for (StreamSummary summary : message.summaries)
        StreamSummary.serializer.serialize(summary, out, version);
}
 

import org.apache.cassandra.io.util.DataOutputStreamPlus; //导入依赖的package包/类
public static void serialize(StreamMessage message, DataOutputStreamPlus out, int version, StreamSession session) throws IOException
{
    ByteBuffer buff = ByteBuffer.allocate(1);
    // message type
    buff.put(message.type.type);
    buff.flip();
    out.write(buff);
    message.type.outSerializer.serialize(message, out, version, session);
}
 

import org.apache.cassandra.io.util.DataOutputStreamPlus; //导入依赖的package包/类
@SuppressWarnings("resource")
protected static DataOutputStreamPlus getWriteChannel(Socket socket) throws IOException
{
    WritableByteChannel out = socket.getChannel();
    // socket channel is null when encrypted(SSL)
    if (out == null)
        return new WrappedDataOutputStreamPlus(new BufferedOutputStream(socket.getOutputStream()));
    return new BufferedDataOutputStreamPlus(out);
}
 

import org.apache.cassandra.io.util.DataOutputStreamPlus; //导入依赖的package包/类
private void rewriteSSTableMetadata(Descriptor descriptor, Map<MetadataType, MetadataComponent> currentComponents) throws IOException
{
    String filePath = descriptor.tmpFilenameFor(Component.STATS);
    try (DataOutputStreamPlus out = new BufferedDataOutputStreamPlus(new FileOutputStream(filePath)))
    {
        serialize(currentComponents, out, descriptor.version);
        out.flush();
    }
    // we cant move a file on top of another file in windows:
    if (FBUtilities.isWindows())
        FileUtils.delete(descriptor.filenameFor(Component.STATS));
    FileUtils.renameWithConfirm(filePath, descriptor.filenameFor(Component.STATS));

}
 

import org.apache.cassandra.io.util.DataOutputStreamPlus; //导入依赖的package包/类
private static void testBloomFilterWrite(boolean offheap, boolean oldBfHashOrder) throws IOException
{
    IPartitioner partitioner = Util.testPartitioner();
    try (IFilter bf = FilterFactory.getFilter(1000000, 0.0001, offheap, oldBfHashOrder))
    {
        for (int i = 0; i < 100; i++)
            bf.add(partitioner.decorateKey(partitioner.getTokenFactory().toByteArray(partitioner.getRandomToken())));
        try (DataOutputStreamPlus out = getOutput(oldBfHashOrder ? "2.1" : "3.0", "utils.BloomFilter.bin"))
        {
            FilterFactory.serialize(bf, out);
        }
    }
}
 

import org.apache.cassandra.io.util.DataOutputStreamPlus; //导入依赖的package包/类
private static void testBloomFilterWrite1000(boolean offheap, boolean oldBfHashOrder) throws IOException
{
    try (IFilter bf = FilterFactory.getFilter(1000000, 0.0001, offheap, oldBfHashOrder))
    {
        for (int i = 0; i < 1000; i++)
            bf.add(Util.dk(Int32Type.instance.decompose(i)));
        try (DataOutputStreamPlus out = getOutput(oldBfHashOrder ? "2.1" : "3.0", "utils.BloomFilter1000.bin"))
        {
            FilterFactory.serialize(bf, out);
        }
    }
}
 

import org.apache.cassandra.io.util.DataOutputStreamPlus; //导入依赖的package包/类
private void testEndpointStateWrite() throws IOException
{
    DataOutputStreamPlus out = getOutput("gms.EndpointState.bin");
    HeartBeatState.serializer.serialize(Statics.HeartbeatSt, out, getVersion());
    EndpointState.serializer.serialize(Statics.EndpointSt, out, getVersion());
    VersionedValue.serializer.serialize(Statics.vv0, out, getVersion());
    VersionedValue.serializer.serialize(Statics.vv1, out, getVersion());
    out.close();

    // test serializedSize
    testSerializedSize(Statics.HeartbeatSt, HeartBeatState.serializer);
    testSerializedSize(Statics.EndpointSt, EndpointState.serializer);
    testSerializedSize(Statics.vv0, VersionedValue.serializer);
    testSerializedSize(Statics.vv1, VersionedValue.serializer);
}
 

import org.apache.cassandra.io.util.DataOutputStreamPlus; //导入依赖的package包/类
@SuppressWarnings("resource")
protected static DataOutputStreamPlus getOutput(String version, String name) throws IOException
{
    File f = new File("test/data/serialization/" + version + '/' + name);
    f.getParentFile().mkdirs();
    return new BufferedDataOutputStreamPlus(new FileOutputStream(f).getChannel());
}
 

import org.apache.cassandra.io.util.DataOutputStreamPlus; //导入依赖的package包/类
public File serialize(Map<MetadataType, MetadataComponent> metadata, MetadataSerializer serializer, Version version)
        throws IOException, FileNotFoundException
{
    // Serialize to tmp file
    File statsFile = File.createTempFile(Component.STATS.name, null);
    try (DataOutputStreamPlus out = new BufferedDataOutputStreamPlus(new FileOutputStream(statsFile)))
    {
        serializer.serialize(metadata, out, version);
    }
    return statsFile;
}
 

import org.apache.cassandra.io.util.DataOutputStreamPlus; //导入依赖的package包/类
public void serialize(RetryMessage message, DataOutputStreamPlus out, int version, StreamSession session) throws IOException
{
    UUIDSerializer.serializer.serialize(message.cfId, out, MessagingService.current_version);
    out.writeInt(message.sequenceNumber);
}
 

import org.apache.cassandra.io.util.DataOutputStreamPlus; //导入依赖的package包/类
public void serialize(ReceivedMessage message, DataOutputStreamPlus out, int version, StreamSession session) throws IOException
{
    UUIDSerializer.serializer.serialize(message.cfId, out, MessagingService.current_version);
    out.writeInt(message.sequenceNumber);
}
 

import org.apache.cassandra.io.util.DataOutputStreamPlus; //导入依赖的package包/类
public void serialize(IncomingFileMessage message, DataOutputStreamPlus out, int version, StreamSession session) throws IOException
{
    throw new UnsupportedOperationException("Not allowed to call serialize on an incoming file");
}