|
|
@ -76,7 +76,7 @@ class NodeConnCB(): |
|
|
|
|
|
|
|
|
|
|
|
# Message receiving methods |
|
|
|
# Message receiving methods |
|
|
|
|
|
|
|
|
|
|
|
def deliver(self, conn, message): |
|
|
|
def on_message(self, conn, message): |
|
|
|
"""Receive message and dispatch message to appropriate callback. |
|
|
|
"""Receive message and dispatch message to appropriate callback. |
|
|
|
|
|
|
|
|
|
|
|
We keep a count of how many of each message type has been received |
|
|
|
We keep a count of how many of each message type has been received |
|
|
@ -233,12 +233,14 @@ class NodeConn(asyncore.dispatcher): |
|
|
|
# Connection and disconnection methods |
|
|
|
# Connection and disconnection methods |
|
|
|
|
|
|
|
|
|
|
|
def handle_connect(self): |
|
|
|
def handle_connect(self): |
|
|
|
|
|
|
|
"""asyncore callback when a connection is opened.""" |
|
|
|
if self.state != "connected": |
|
|
|
if self.state != "connected": |
|
|
|
logger.debug("Connected & Listening: %s:%d" % (self.dstaddr, self.dstport)) |
|
|
|
logger.debug("Connected & Listening: %s:%d" % (self.dstaddr, self.dstport)) |
|
|
|
self.state = "connected" |
|
|
|
self.state = "connected" |
|
|
|
self.cb.on_open(self) |
|
|
|
self.cb.on_open(self) |
|
|
|
|
|
|
|
|
|
|
|
def handle_close(self): |
|
|
|
def handle_close(self): |
|
|
|
|
|
|
|
"""asyncore callback when a connection is closed.""" |
|
|
|
logger.debug("Closing connection to: %s:%d" % (self.dstaddr, self.dstport)) |
|
|
|
logger.debug("Closing connection to: %s:%d" % (self.dstaddr, self.dstport)) |
|
|
|
self.state = "closed" |
|
|
|
self.state = "closed" |
|
|
|
self.recvbuf = b"" |
|
|
|
self.recvbuf = b"" |
|
|
@ -250,7 +252,7 @@ class NodeConn(asyncore.dispatcher): |
|
|
|
self.cb.on_close(self) |
|
|
|
self.cb.on_close(self) |
|
|
|
|
|
|
|
|
|
|
|
def disconnect_node(self): |
|
|
|
def disconnect_node(self): |
|
|
|
""" Disconnect the p2p connection. |
|
|
|
"""Disconnect the p2p connection. |
|
|
|
|
|
|
|
|
|
|
|
Called by the test logic thread. Causes the p2p connection |
|
|
|
Called by the test logic thread. Causes the p2p connection |
|
|
|
to be disconnected on the next iteration of the asyncore loop.""" |
|
|
|
to be disconnected on the next iteration of the asyncore loop.""" |
|
|
@ -258,16 +260,19 @@ class NodeConn(asyncore.dispatcher): |
|
|
|
|
|
|
|
|
|
|
|
# Socket read methods |
|
|
|
# Socket read methods |
|
|
|
|
|
|
|
|
|
|
|
def readable(self): |
|
|
|
|
|
|
|
return True |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def handle_read(self): |
|
|
|
def handle_read(self): |
|
|
|
|
|
|
|
"""asyncore callback when data is read from the socket.""" |
|
|
|
t = self.recv(8192) |
|
|
|
t = self.recv(8192) |
|
|
|
if len(t) > 0: |
|
|
|
if len(t) > 0: |
|
|
|
self.recvbuf += t |
|
|
|
self.recvbuf += t |
|
|
|
self.got_data() |
|
|
|
self._on_data() |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _on_data(self): |
|
|
|
|
|
|
|
"""Try to read P2P messages from the recv buffer. |
|
|
|
|
|
|
|
|
|
|
|
def got_data(self): |
|
|
|
This method reads data from the buffer in a loop. It deserializes, |
|
|
|
|
|
|
|
parses and verifies the P2P header, then passes the P2P payload to |
|
|
|
|
|
|
|
the on_message callback for processing.""" |
|
|
|
try: |
|
|
|
try: |
|
|
|
while True: |
|
|
|
while True: |
|
|
|
if len(self.recvbuf) < 4: |
|
|
|
if len(self.recvbuf) < 4: |
|
|
@ -292,24 +297,27 @@ class NodeConn(asyncore.dispatcher): |
|
|
|
f = BytesIO(msg) |
|
|
|
f = BytesIO(msg) |
|
|
|
t = MESSAGEMAP[command]() |
|
|
|
t = MESSAGEMAP[command]() |
|
|
|
t.deserialize(f) |
|
|
|
t.deserialize(f) |
|
|
|
self.got_message(t) |
|
|
|
self._log_message("receive", t) |
|
|
|
|
|
|
|
self.on_message(t) |
|
|
|
except Exception as e: |
|
|
|
except Exception as e: |
|
|
|
logger.exception('Error reading message:', repr(e)) |
|
|
|
logger.exception('Error reading message:', repr(e)) |
|
|
|
raise |
|
|
|
raise |
|
|
|
|
|
|
|
|
|
|
|
def got_message(self, message): |
|
|
|
def on_message(self, message): |
|
|
|
self._log_message("receive", message) |
|
|
|
"""Callback for processing a P2P payload. Calls into NodeConnCB.""" |
|
|
|
self.cb.deliver(self, message) |
|
|
|
self.cb.on_message(self, message) |
|
|
|
|
|
|
|
|
|
|
|
# Socket write methods |
|
|
|
# Socket write methods |
|
|
|
|
|
|
|
|
|
|
|
def writable(self): |
|
|
|
def writable(self): |
|
|
|
|
|
|
|
"""asyncore method to determine whether the handle_write() callback should be called on the next loop.""" |
|
|
|
with mininode_lock: |
|
|
|
with mininode_lock: |
|
|
|
pre_connection = self.state == "connecting" |
|
|
|
pre_connection = self.state == "connecting" |
|
|
|
length = len(self.sendbuf) |
|
|
|
length = len(self.sendbuf) |
|
|
|
return (length > 0 or pre_connection) |
|
|
|
return (length > 0 or pre_connection) |
|
|
|
|
|
|
|
|
|
|
|
def handle_write(self): |
|
|
|
def handle_write(self): |
|
|
|
|
|
|
|
"""asyncore callback when data should be written to the socket.""" |
|
|
|
with mininode_lock: |
|
|
|
with mininode_lock: |
|
|
|
# asyncore does not expose socket connection, only the first read/write |
|
|
|
# asyncore does not expose socket connection, only the first read/write |
|
|
|
# event, thus we must check connection manually here to know when we |
|
|
|
# event, thus we must check connection manually here to know when we |
|
|
@ -327,6 +335,10 @@ class NodeConn(asyncore.dispatcher): |
|
|
|
self.sendbuf = self.sendbuf[sent:] |
|
|
|
self.sendbuf = self.sendbuf[sent:] |
|
|
|
|
|
|
|
|
|
|
|
def send_message(self, message, pushbuf=False): |
|
|
|
def send_message(self, message, pushbuf=False): |
|
|
|
|
|
|
|
"""Send a P2P message over the socket. |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
This method takes a P2P payload, builds the P2P header and adds |
|
|
|
|
|
|
|
the message to the send buffer to be sent over the socket.""" |
|
|
|
if self.state != "connected" and not pushbuf: |
|
|
|
if self.state != "connected" and not pushbuf: |
|
|
|
raise IOError('Not connected, no pushbuf') |
|
|
|
raise IOError('Not connected, no pushbuf') |
|
|
|
self._log_message("send", message) |
|
|
|
self._log_message("send", message) |
|
|
@ -353,6 +365,7 @@ class NodeConn(asyncore.dispatcher): |
|
|
|
# Class utility methods |
|
|
|
# Class utility methods |
|
|
|
|
|
|
|
|
|
|
|
def _log_message(self, direction, msg): |
|
|
|
def _log_message(self, direction, msg): |
|
|
|
|
|
|
|
"""Logs a message being sent or received over the connection.""" |
|
|
|
if direction == "send": |
|
|
|
if direction == "send": |
|
|
|
log_message = "Send message to " |
|
|
|
log_message = "Send message to " |
|
|
|
elif direction == "receive": |
|
|
|
elif direction == "receive": |
|
|
|