src/java/org/apache/cassandra/net/IncomingTcpConnection.java
/**
     * A new connection will either stream or message for its entire lifetime: because streaming
     * bypasses the InputStream implementations to use sendFile, we cannot begin buffering until
     * we've determined the type of the connection.
     */
A new connection will either stream or message for its entire lifetime: because streaming
bypasses the InputStream implementations to use sendFile, we cannot begin buffering until
we've determined the type of the connection.
A new connection will either stream or message for its entire lifetime : because streaming bypasses the InputStream implementations to use sendFile we cannot begin buffering until we ' ve determined the type of the connection .
@Override
    public void run()
    {
        DataInputStream input;
        boolean isStream;
        int version;
        try
        {
            input = new DataInputStream(socket.getInputStream());
            MessagingService.validateMagic(input.readInt());
            int header = input.readInt();
            isStream = MessagingService.getBits(header, 3, 1) == 1;
            version = MessagingService.getBits(header, 15, 8);
            logger.debug("Version for {} is {}", from, version);
            if (isStream)
            {
                if (version == MessagingService.version_)
                {
                    int size = input.readInt();
                    byte[] headerBytes = new byte[size];
                    input.readFully(headerBytes);
                    stream(StreamHeader.serializer().deserialize(new DataInputStream(new FastByteArrayInputStream(headerBytes)), version), input);
                } 
                else
                {
                    logger.error("Received stream using protocol version {} (my version {}). Terminating connection",
                                 version, MessagingService.version_);
                }
                return;
            }
            input = new DataInputStream(new BufferedInputStream(socket.getInputStream(), 4096));
            Message msg = receiveMessage(input, version);
            if (version > MessagingService.version_)
            {
                Gossiper.instance.addSavedEndpoint(from);
                logger.info("Received " + (isStream ? "streaming " : "") + "connection from newer protocol version. Ignorning");
            }
            else if (msg != null)
            {
                Gossiper.instance.setVersion(msg.getFrom(), version);
                logger.debug("set version for {} to {}", from, version);
            }
            while (true)
            {
                MessagingService.validateMagic(input.readInt());
                header = input.readInt();
                assert isStream == (MessagingService.getBits(header, 3, 1) == 1) : "Connections cannot change type: " + isStream;
                version = MessagingService.getBits(header, 15, 8);
                logger.trace("Version is now {}", version);
                receiveMessage(input, version);
            }
        } 
        catch (EOFException e)
        {
            logger.trace("eof reading from socket; closing", e);
        } 
        catch (IOException e)
        {
            logger.debug("IOError reading from socket; closing", e);
        }
        finally
        {
            close();
        }
    }
@ Override public void run ( ) { DataInputStream input ; boolean isStream ; int version ; try { input = new DataInputStream ( socket . getInputStream ( ) ) ; MessagingService . validateMagic ( input . readInt ( ) ) ; int header = input . readInt ( ) ; isStream = MessagingService . getBits ( header , 3 , 1 ) == 1 ; version = MessagingService . getBits ( header , 15 , 8 ) ; logger . debug ( "Version for {} is {}" , from , version ) ; if ( isStream ) { if ( version == MessagingService . version_ ) { int size = input . readInt ( ) ; byte [ ] headerBytes = new byte [ size ] ; input . readFully ( headerBytes ) ; stream ( StreamHeader . serializer ( ) . deserialize ( new DataInputStream ( new FastByteArrayInputStream ( headerBytes ) ) , version ) , input ) ; } else { logger . error ( "Received stream using protocol version {} (my version {}). Terminating connection" , version , MessagingService . version_ ) ; } return ; } input = new DataInputStream ( new BufferedInputStream ( socket . getInputStream ( ) , 4096 ) ) ; Message msg = receiveMessage ( input , version ) ; if ( version > MessagingService . version_ ) { Gossiper . instance . addSavedEndpoint ( from ) ; logger . info ( "Received " + ( isStream ? "streaming " : "" ) + "connection from newer protocol version. Ignorning" ) ; } else if ( msg != null ) { Gossiper . instance . setVersion ( msg . getFrom ( ) , version ) ; logger . debug ( "set version for {} to {}" , from , version ) ; } while ( true ) { MessagingService . validateMagic ( input . readInt ( ) ) ; header = input . readInt ( ) ; assert isStream == ( MessagingService . getBits ( header , 3 , 1 ) == 1 ) : "Connections cannot change type: " + isStream ; version = MessagingService . getBits ( header , 15 , 8 ) ; logger . trace ( "Version is now {}" , version ) ; receiveMessage ( input , version ) ; } } catch ( EOFException e ) { logger . trace ( "eof reading from socket; closing" , e ) ; } catch ( IOException e ) { logger . debug ( "IOError reading from socket; closing" , e ) ; } finally { close ( ) ; } }
A new connection will either stream or message for its entire lifetime: because streaming
 bypasses the InputStream implementations to use sendFile, we cannot begin buffering until
 we've determined the type of the connection.
A new connection will either stream or message for its entire lifetime : because streaming bypasses the InputStream implementations to use sendFile we cannot begin buffering until we ' ve determined the type of the connection .
// determine the connection type to decide whether to buffer // streaming connections are per-session and have a fixed version.  we can't do anything with a wrong-version stream connection, so drop it. // We are done with this connection.... // we should buffer // Receive the first message to set the version. // save the endpoint so gossip will reconnect to it // loop to get the next message. // prepare to read the next message // connection will be reset so no need to throw an exception.