natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

CASSANDRA-4099-reopened 

patch by Vijay; reviewed by Brandon Williams for CASSANDRA-4098
This commit is contained in:
Vijay Parthasarathy 2012-03-30 11:42:10 -07:00
parent 4a6531172c
commit cec4253ba8
1 changed files with 10 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
src/java/org/apache/cassandra/net/IncomingTcpConnection.java
View File

@ -92,19 +92,16 @@ public class IncomingTcpConnection extends Thread
// we should buffer // we should buffer
input = new DataInputStream(new BufferedInputStream(socket.getInputStream(), 4096)); input = new DataInputStream(new BufferedInputStream(socket.getInputStream(), 4096));
// Receive the first message to set the version. // Receive the first message to set the version.
Message msg = receiveMessage(input, version); from = receiveMessage(input, version); // why? see => CASSANDRA-4099
from = msg.getFrom(); // why? see => CASSANDRA-4099
if (version > MessagingService.version_) if (version > MessagingService.version_)
{ {
// save the endpoint so gossip will reconnect to it // save the endpoint so gossip will reconnect to it
Gossiper.instance.addSavedEndpoint(from); Gossiper.instance.addSavedEndpoint(from);
logger.info("Received " + (isStream ? "streaming " : "") + "connection from newer protocol version. Ignoring"); logger.info("Received " + (isStream ? "streaming " : "") + "connection from newer protocol version. Ignoring");
return;
} }
else if (msg != null) Gossiper.instance.setVersion(from, version);
{ logger.debug("set version for {} to {}", from, version);
Gossiper.instance.setVersion(from, version);
logger.debug("set version for {} to {}", from, version);
}
// loop to get the next message. // loop to get the next message.
while (true) while (true)
@ -133,7 +130,7 @@ public class IncomingTcpConnection extends Thread
} }
} }
private Message receiveMessage(DataInputStream input, int version) throws IOException private InetAddress receiveMessage(DataInputStream input, int version) throws IOException
{ {
int totalSize = input.readInt(); int totalSize = input.readInt();
String id = input.readUTF(); String id = input.readUTF();
@ -158,10 +155,12 @@ public class IncomingTcpConnection extends Thread
{ {
Message message = new Message(header, body, version); Message message = new Message(header, body, version);
MessagingService.instance().receive(message, id); MessagingService.instance().receive(message, id);
return message;
} }
logger.debug("Received connection from newer protocol version {}. Ignoring message", version); else
return null; {
logger.debug("Received connection from newer protocol version {}. Ignoring message", version);
}
return header.getFrom();
} }
private void close() private void close()