import org.apache.cassandra.streaming.StreamRequest; //导入依赖的package包/类
/**
 * When StreamSession completes, make all keyspaces/ranges in session available to be served.
 *
 * @param event Stream event.
 */
@Override
public void handleStreamEvent(StreamEvent event)
{
    if (event.eventType == StreamEvent.Type.STREAM_COMPLETE)
    {
        StreamEvent.SessionCompleteEvent se = (StreamEvent.SessionCompleteEvent) event;
        if (se.success)
        {
            for (StreamRequest request : se.requests)
            {
                SystemKeyspace.updateAvailableRanges(request.keyspace, request.ranges);
            }
        }
    }
}
 

import org.apache.cassandra.streaming.StreamRequest; //导入依赖的package包/类
public PrepareMessage deserialize(ReadableByteChannel in, int version, StreamSession session) throws IOException
{
    DataInput input = new DataInputStream(Channels.newInputStream(in));
    PrepareMessage message = new PrepareMessage();
    // requests
    int numRequests = input.readInt();
    for (int i = 0; i < numRequests; i++)
        message.requests.add(StreamRequest.serializer.deserialize(input, version));
    // summaries
    int numSummaries = input.readInt();
    for (int i = 0; i < numSummaries; i++)
        message.summaries.add(StreamSummary.serializer.deserialize(input, version));
    return message;
}
 

import org.apache.cassandra.streaming.StreamRequest; //导入依赖的package包/类
public void serialize(PrepareMessage message, DataOutputStreamAndChannel 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.streaming.StreamRequest; //导入依赖的package包/类
public void serialize(PrepareMessage message, WritableByteChannel out, int version, StreamSession session) throws IOException
{
    DataOutput output = new DataOutputStream(Channels.newOutputStream(out));
    // requests
    output.writeInt(message.requests.size());
    for (StreamRequest request : message.requests)
        StreamRequest.serializer.serialize(request, output, version);
    // summaries
    output.writeInt(message.summaries.size());
    for (StreamSummary summary : message.summaries)
        StreamSummary.serializer.serialize(summary, output, version);
}
 

import org.apache.cassandra.streaming.StreamRequest; //导入依赖的package包/类
@SuppressWarnings("resource") // Not closing constructed DataInputPlus's as the channel needs to remain open.
public PrepareMessage deserialize(ReadableByteChannel in, int version, StreamSession session) throws IOException
{
    DataInputPlus input = new DataInputStreamPlus(Channels.newInputStream(in));
    PrepareMessage message = new PrepareMessage();
    // requests
    int numRequests = input.readInt();
    for (int i = 0; i < numRequests; i++)
        message.requests.add(StreamRequest.serializer.deserialize(input, version));
    // summaries
    int numSummaries = input.readInt();
    for (int i = 0; i < numSummaries; i++)
        message.summaries.add(StreamSummary.serializer.deserialize(input, version));
    return message;
}
 

import org.apache.cassandra.streaming.StreamRequest; //导入依赖的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);
}