import org.apache.cassandra.metrics.StreamingMetrics; //导入依赖的package包/类
private boolean checkRegistrations(ObjectName name) {
    if (name != null && server.isRegistered(name)) {
        return false;
    }
    
    boolean result = false;
    
    try {
        String type = name != null ? name.getKeyProperty("type") : null;            
        if (type == null || tables.matcher(type).matches()) {
            result |= ColumnFamilyStore.checkRegistration(client, server);
        }
        if (type == null || StreamingMetrics.TYPE_NAME.equals(type)) {
            result |= StreamingMetrics.checkRegistration(client, server);
        }
    } catch (MalformedObjectNameException | UnknownHostException e) {
        // TODO: log
    }
    return result;
}
 

import org.apache.cassandra.metrics.StreamingMetrics; //导入依赖的package包/类
public FileStreamTask(StreamHeader header, InetAddress to)
{
    this.header = header;
    this.to = to;
    this.throttle = new Throttle(toString(), new Throttle.ThroughputFunction()
    {
        /** @return Instantaneous throughput target in bytes per millisecond. */
        public int targetThroughput()
        {
            if (DatabaseDescriptor.getStreamThroughputOutboundMegabitsPerSec() < 1)
                // throttling disabled
                return 0;
            // total throughput
            int totalBytesPerMS = DatabaseDescriptor.getStreamThroughputOutboundMegabitsPerSec() * 1024 * 1024 / 8 / 1000;
            // per stream throughput (target bytes per MS)
            return totalBytesPerMS / Math.max(1, (int)StreamingMetrics.activeStreamsOutbound.count());
        }
    });
    metrics = StreamingMetrics.get(to);
}
 

import org.apache.cassandra.metrics.StreamingMetrics; //导入依赖的package包/类
/**
 * Create new streaming session with the peer.
 *
 * @param peer Address of streaming peer
 * @param connecting Actual connecting address
 * @param factory is used for establishing connection
 */
public StreamSession(InetAddress peer, InetAddress connecting, StreamConnectionFactory factory, int index)
{
    this.peer = peer;
    this.connecting = connecting;
    this.index = index;
    this.factory = factory;
    this.handler = new ConnectionHandler(this);
    this.metrics = StreamingMetrics.get(connecting);
}
 

import org.apache.cassandra.metrics.StreamingMetrics; //导入依赖的package包/类
/**
 * Call back after sending FileMessageHeader.
 *
 * @param header sent header
 */
public void fileSent(FileMessageHeader header)
{
    long headerSize = header.size();
    StreamingMetrics.totalOutgoingBytes.inc(headerSize);
    metrics.outgoingBytes.inc(headerSize);
    // schedule timeout for receiving ACK
    StreamTransferTask task = transfers.get(header.cfId);
    if (task != null)
    {
        task.scheduleTimeout(header.sequenceNumber, 12, TimeUnit.HOURS);
    }
}
 

import org.apache.cassandra.metrics.StreamingMetrics; //导入依赖的package包/类
/**
 * Call back after receiving FileMessageHeader.
 *
 * @param message received file
 */
public void receive(IncomingFileMessage message)
{
    long headerSize = message.header.size();
    StreamingMetrics.totalIncomingBytes.inc(headerSize);
    metrics.incomingBytes.inc(headerSize);
    // send back file received message
    handler.sendMessage(new ReceivedMessage(message.header.cfId, message.header.sequenceNumber));
    receivers.get(message.header.cfId).received(message.sstable);
}
 

import org.apache.cassandra.metrics.StreamingMetrics; //导入依赖的package包/类
/**
 * Create new streaming session with the peer.
 *
 * @param peer Address of streaming peer
 */
public StreamSession(InetAddress peer)
{
    this.peer = peer;
    this.handler = new ConnectionHandler(this);
    this.metrics = StreamingMetrics.get(peer);
}
 

import org.apache.cassandra.metrics.StreamingMetrics; //导入依赖的package包/类
/**
 * Call back after sending FileMessageHeader.
 *
 * @param header sent header
 */
public void fileSent(FileMessageHeader header)
{
    long headerSize = header.size();
    StreamingMetrics.totalOutgoingBytes.inc(headerSize);
    metrics.outgoingBytes.inc(headerSize);
}
 

import org.apache.cassandra.metrics.StreamingMetrics; //导入依赖的package包/类
/**
 * Call back after receiving FileMessageHeader.
 *
 * @param message received file
 */
public void receive(FileMessage message)
{
    long headerSize = message.header.size();
    StreamingMetrics.totalIncomingBytes.inc(headerSize);
    metrics.incomingBytes.inc(headerSize);
    // send back file received message
    handler.sendMessage(new ReceivedMessage(message.header.cfId, message.header.sequenceNumber));
    receivers.get(message.header.cfId).received(message.sstable);
}
 

import org.apache.cassandra.metrics.StreamingMetrics; //导入依赖的package包/类
/**
 * Create new streaming session with the peer.
 *
 * @param peer Address of streaming peer
 * @param connecting Actual connecting address
 * @param factory is used for establishing connection
 */
public StreamSession(InetAddress peer, InetAddress connecting, StreamConnectionFactory factory, int index, boolean keepSSTableLevel, boolean isIncremental)
{
    this.peer = peer;
    this.connecting = connecting;
    this.index = index;
    this.factory = factory;
    this.handler = new ConnectionHandler(this);
    this.metrics = StreamingMetrics.get(connecting);
    this.keepSSTableLevel = keepSSTableLevel;
    this.isIncremental = isIncremental;
}
 

import org.apache.cassandra.metrics.StreamingMetrics; //导入依赖的package包/类
/**
 * Create new streaming session with the peer.
 *
 * @param peer Address of streaming peer
 * @param factory is used for establishing connection
 */
public StreamSession(InetAddress peer, StreamConnectionFactory factory, int index)
{
    this.peer = peer;
    this.index = index;
    this.factory = factory;
    this.handler = new ConnectionHandler(this);
    this.metrics = StreamingMetrics.get(peer);
}