diff --git a/itests/tests/test_squeak_node.py b/itests/tests/test_squeak_node.py index 5688c0de..859086b4 100644 --- a/itests/tests/test_squeak_node.py +++ b/itests/tests/test_squeak_node.py @@ -940,7 +940,7 @@ def test_get_squeak_details(admin_stub, saved_squeak_hash): serialized_squeak_hex = ( get_squeak_details_response.squeak_detail_entry.serialized_squeak_hex ) - print("serialized_squeak_hex: {}".format(serialized_squeak_hex)) + # print("serialized_squeak_hex: {}".format(serialized_squeak_hex)) assert len(serialized_squeak_hex) > 200 serialized_squeak = bytes.fromhex(serialized_squeak_hex) @@ -956,8 +956,8 @@ def test_like_squeak(admin_stub, saved_squeak_hash): squeak_hash=saved_squeak_hash, ) ) - print("get_squeak_display_response.squeak_display_entry:") - print(get_squeak_display_response.squeak_display_entry) + # print("get_squeak_display_response.squeak_display_entry:") + # print(get_squeak_display_response.squeak_display_entry) assert ( get_squeak_display_response.squeak_display_entry.liked_time_s == 0 ) @@ -1011,13 +1011,18 @@ def test_connect_peer(admin_stub, other_admin_stub): time.sleep(2) connected_peers = get_connected_peers(admin_stub) assert len(connected_peers) == 1 + print("Admin node connected to peers: ") + print(connected_peers) other_connected_peers = get_connected_peers(other_admin_stub) assert len(other_connected_peers) == 1 - # time.sleep(2) - # connected_peers = get_connected_peers(admin_stub) - # assert len(connected_peers) == 0 - # other_connected_peers = get_connected_peers(other_admin_stub) - # assert len(other_connected_peers) == 0 + print("Other Admin node connected to peers: ") + print(other_connected_peers) + + time.sleep(2) + connected_peers = get_connected_peers(admin_stub) + assert len(connected_peers) == 0 + other_connected_peers = get_connected_peers(other_admin_stub) + assert len(other_connected_peers) == 0 def test_share_single_squeak( diff --git a/itests/tests/util.py b/itests/tests/util.py index 0bb48641..866b8eb9 100644 --- a/itests/tests/util.py +++ b/itests/tests/util.py @@ -152,12 +152,12 @@ def open_peer_connection(node_stub, peer_name, peer_host, peer_port): ) ) yield peer_id - # # Disconnect the peer - # node_stub.DisconnectPeer( - # squeak_admin_pb2.DisconnectPeerRequest( - # peer_id=peer_id, - # ) - # ) + # Disconnect the peer + node_stub.DisconnectPeer( + squeak_admin_pb2.DisconnectPeerRequest( + peer_id=peer_id, + ) + ) # Delete the peer node_stub.DeletePeer( squeak_admin_pb2.DeletePeerRequest( diff --git a/squeaknode/network/connection.py b/squeaknode/network/connection.py index ecbe6159..16a493ca 100644 --- a/squeaknode/network/connection.py +++ b/squeaknode/network/connection.py @@ -78,8 +78,8 @@ class Connection(): def handle_messages(self): peer_message_handler = PeerMessageHandler( self.peer, self.squeak_controller) - while True: - peer_message_handler.handle_msgs() + peer_message_handler.handle_msgs() + logger.info('Finished handling messages...') @contextmanager def open_connection(self): diff --git a/squeaknode/network/peer.py b/squeaknode/network/peer.py index ba79ad45..37bd0376 100644 --- a/squeaknode/network/peer.py +++ b/squeaknode/network/peer.py @@ -1,5 +1,6 @@ import logging import queue +import socket import threading import time from io import BytesIO @@ -115,6 +116,10 @@ class Peer(object): def is_handshake_complete(self): return self.handshake_complete.is_set() + @property + def is_open(self): + return not self.stopped.is_set() + @property def last_msg_revc_time(self): return self._last_msg_revc_time @@ -146,14 +151,17 @@ class Peer(object): logger.info('Received msg {} from {}'.format(msg, self)) return msg - def stop(self): - logger.info("Stopping peer: {}".format(self)) - self.stopped.set() + # def stop(self): + # logger.info("Stopping peer: {}".format(self)) + # self.stopped.set() def close(self): logger.info("closing peer socket: {}".format(self._peer_socket)) - if self._peer_socket: + try: + self._peer_socket.shutdown(socket.SHUT_RDWR) self._peer_socket.close() + except Exception: + pass def send_msg(self, msg): logger.debug('Sending msg {} to {}'.format(msg, self)) @@ -177,8 +185,8 @@ class Peer(object): return self def __exit__(self, *exc): - self.stop() - logger.debug('Stopped peer {} ...'.format(self)) + self.close() + logger.debug('Closed connection to peer {} ...'.format(self)) def __repr__(self): return "Peer(%s)" % (self.address_string) @@ -226,8 +234,16 @@ class MessageReceiver: def _recv_msgs(self): while True: - recv_data = self.socket.recv(SOCKET_READ_LEN) + logger.info("Recving msg...") + try: + recv_data = self.socket.recv(SOCKET_READ_LEN) + except Exception: + logger.error("Error in recv") + self.queue.put(None) + raise Exception('Peer disconnected') if not recv_data: + logger.error("revc_data is None") + self.queue.put(None) raise Exception('Peer disconnected') for msg in self.decoder.process_recv_data(recv_data): diff --git a/squeaknode/network/peer_client.py b/squeaknode/network/peer_client.py new file mode 100644 index 00000000..eaecb865 --- /dev/null +++ b/squeaknode/network/peer_client.py @@ -0,0 +1,66 @@ +import logging +import socket +import threading + +import squeak.params + + +MIN_PEERS = 5 +MAX_PEERS = 10 +UPDATE_THREAD_SLEEP_TIME = 10 + + +logger = logging.getLogger(__name__) + + +class PeerClient(object): + """Creates outgoing connections to other peers in the network. + """ + + def __init__(self, connection_manager, port=None): + self.ip = socket.gethostbyname('localhost') + self.port = port or squeak.params.params.DEFAULT_PORT + self.connection_manager = connection_manager + + def start(self, peer_handler): + self.peer_handler = peer_handler + + def make_connection(self, ip, port): + address = (ip, port) + logger.debug('Making connection to {}'.format(address)) + logger.info('Making connection to {}'.format(address)) + try: + peer_socket = socket.socket() + logger.info('Got socket to {}'.format(address)) + peer_socket.connect(address) + peer_socket.setblocking(True) + self.peer_handler.handle_connection( + peer_socket, address, outgoing=True) + except Exception: + logger.exception('Failed to make connection to {}'.format(address)) + + def connect_address(self, address): + """Connect to new address.""" + logger.debug('Connecting to peer with address {}'.format(address)) + logger.info('Connecting to peer with address {}'.format(address)) + hostname, port = address + ip = socket.gethostbyname(hostname) + new_address = (ip, port) + if self.connection_manager.has_connection(new_address): + return + logger.info('Connecting to peer with ip address {}'.format(ip)) + threading.Thread( + target=self.make_connection, + args=(ip, port), + ).start() + + def disconnect_address(self, address): + """Connect to new address.""" + logger.info('Disconnecting peer with address {}'.format(address)) + hostname, port = address + ip = socket.gethostbyname(hostname) + new_address = (ip, port) + peer = self.connection_manager.get_peer(new_address) + if peer is None: + return + peer.close() diff --git a/squeaknode/network/peer_handler.py b/squeaknode/network/peer_handler.py index eda52f92..f68a60eb 100644 --- a/squeaknode/network/peer_handler.py +++ b/squeaknode/network/peer_handler.py @@ -1,4 +1,5 @@ import logging +import threading from squeaknode.network.connection import Connection from squeaknode.network.connection_manager import ConnectionManager @@ -38,6 +39,12 @@ class PeerHandler(): logger.debug('Stopped controller for peer address {}.'.format(address)) logger.info('Stopped controller for peer address {}.'.format(address)) + def handle_connection(self, peer_socket, address, outgoing): + threading.Thread( + target=self.start, + args=(peer_socket, address, outgoing,), + ).start() + # class PeerListener(PeerMessageHandler): # """Handles receiving messages from a peer. diff --git a/squeaknode/network/peer_message_handler.py b/squeaknode/network/peer_message_handler.py index 5c66a98c..e0bdc40e 100644 --- a/squeaknode/network/peer_message_handler.py +++ b/squeaknode/network/peer_message_handler.py @@ -48,9 +48,11 @@ class PeerMessageHandler: This method blocks when the peer has not sent any messages. """ logger.info('Started handling connected messages...') - while True: + while self.peer.is_open: msg = self.peer.recv_msg() - self.handle_peer_message(msg) + if msg is not None: + self.handle_peer_message(msg) + logger.info('Finished handling connected messages...') def handle_peer_message(self, msg): """Handle messages from a peer with completed handshake.""" diff --git a/squeaknode/network/peer_server.py b/squeaknode/network/peer_server.py index e2e7ec19..3352febc 100644 --- a/squeaknode/network/peer_server.py +++ b/squeaknode/network/peer_server.py @@ -40,49 +40,5 @@ class PeerServer(object): while True: peer_socket, address = listen_socket.accept() peer_socket.setblocking(True) - self.handle_connection(peer_socket, address, outgoing=False) - - def make_connection(self, ip, port): - address = (ip, port) - logger.debug('Making connection to {}'.format(address)) - logger.info('Making connection to {}'.format(address)) - try: - peer_socket = socket.socket() - logger.info('Got socket to {}'.format(address)) - peer_socket.connect(address) - peer_socket.setblocking(True) - self.handle_connection(peer_socket, address, outgoing=True) - except Exception: - logger.exception('Failed to make connection to {}'.format(address)) - - def handle_connection(self, peer_socket, address, outgoing): - threading.Thread( - target=self.peer_handler.start, - args=(peer_socket, address, outgoing,), - ).start() - - def connect_address(self, address): - """Connect to new address.""" - logger.debug('Connecting to peer with address {}'.format(address)) - logger.info('Connecting to peer with address {}'.format(address)) - hostname, port = address - ip = socket.gethostbyname(hostname) - new_address = (ip, port) - if self.connection_manager.has_connection(new_address): - return - logger.info('Connecting to peer with ip address {}'.format(ip)) - threading.Thread( - target=self.make_connection, - args=(ip, port), - ).start() - - def disconnect_address(self, address): - """Connect to new address.""" - logger.info('Disconnecting peer with address {}'.format(address)) - hostname, port = address - ip = socket.gethostbyname(hostname) - new_address = (ip, port) - peer = self.connection_manager.get_peer(new_address) - if peer is None: - return - peer.stop() + self.peer_handler.handle_connection( + peer_socket, address, outgoing=False) diff --git a/squeaknode/node/squeak_controller.py b/squeaknode/node/squeak_controller.py index e399622d..6d4ccf13 100644 --- a/squeaknode/node/squeak_controller.py +++ b/squeaknode/node/squeak_controller.py @@ -31,6 +31,7 @@ from squeaknode.core.squeak_profile import SqueakProfile from squeaknode.core.util import get_hash from squeaknode.core.util import is_address_valid from squeaknode.network.connection_manager import ConnectionManager +from squeaknode.network.peer_client import PeerClient from squeaknode.network.peer_server import PeerServer from squeaknode.node.received_payments_subscription_client import ReceivedPaymentsSubscriptionClient @@ -47,6 +48,7 @@ class SqueakController: squeak_rate_limiter, payment_processor, peer_server: PeerServer, + peer_client: PeerClient, connection_manager: ConnectionManager, config, ): @@ -55,6 +57,7 @@ class SqueakController: self.squeak_rate_limiter = squeak_rate_limiter self.payment_processor = payment_processor self.peer_server = peer_server + self.peer_client = peer_client self.connection_manager = connection_manager self.config = config @@ -558,7 +561,7 @@ class SqueakController: logger.info("Connect to peer: {}".format( peer, )) - self.peer_server.connect_address(peer.address) + self.peer_client.connect_address(peer.address) def connect_peers(self) -> None: peers = self.squeak_db.get_peers() @@ -567,7 +570,7 @@ class SqueakController: peer, )) try: - self.peer_server.connect_address(peer.address) + self.peer_client.connect_address(peer.address) except Exception: logger.exception("Failed to connect to peer {}".format( peer, @@ -710,4 +713,4 @@ class SqueakController: logger.info("Disconnect peer: {}".format( peer, )) - self.peer_server.disconnect_address(peer.address) + self.peer_client.disconnect_address(peer.address) diff --git a/squeaknode/node/squeak_node.py b/squeaknode/node/squeak_node.py index 61f82424..378c9c7c 100644 --- a/squeaknode/node/squeak_node.py +++ b/squeaknode/node/squeak_node.py @@ -15,6 +15,7 @@ from squeaknode.db.db_engine import get_sqlite_connection_string from squeaknode.db.squeak_db import SqueakDb from squeaknode.lightning.lnd_lightning_client import LNDLightningClient from squeaknode.network.connection_manager import ConnectionManager +from squeaknode.network.peer_client import PeerClient from squeaknode.network.peer_handler import PeerHandler from squeaknode.network.peer_server import PeerServer from squeaknode.node.payment_processor import PaymentProcessor @@ -68,6 +69,7 @@ class SqueakNode: self.connection_manager = ConnectionManager() self.peer_server = PeerServer(self.connection_manager) + self.peer_client = PeerClient(self.connection_manager) squeak_controller = SqueakController( squeak_db, @@ -75,6 +77,7 @@ class SqueakNode: squeak_rate_limiter, payment_processor, self.peer_server, + self.peer_client, self.connection_manager, self.config, ) @@ -135,6 +138,7 @@ class SqueakNode: # Start peer socket server self.peer_server.start(self.peer_handler) + self.peer_client.start(self.peer_handler) def stop_running(self): self.stopped.set() diff --git a/tests/node/test_squeak_controller.py b/tests/node/test_squeak_controller.py index 606eacc6..93869d5c 100644 --- a/tests/node/test_squeak_controller.py +++ b/tests/node/test_squeak_controller.py @@ -8,6 +8,7 @@ from squeaknode.core.squeak_core import SqueakCore from squeaknode.core.squeak_peer import SqueakPeer from squeaknode.db.squeak_db import SqueakDb from squeaknode.network.connection_manager import ConnectionManager +from squeaknode.network.peer_client import PeerClient from squeaknode.network.peer_server import PeerServer from squeaknode.node.payment_processor import PaymentProcessor from squeaknode.node.squeak_controller import SqueakController @@ -41,6 +42,11 @@ def peer_server(): return mock.Mock(spec=PeerServer) +@pytest.fixture +def peer_client(): + return mock.Mock(spec=PeerClient) + + @pytest.fixture def connection_manager(): return mock.Mock(spec=ConnectionManager) @@ -78,6 +84,7 @@ def squeak_controller( squeak_rate_limiter, payment_processor, peer_server, + peer_client, connection_manager, config, ): @@ -87,6 +94,7 @@ def squeak_controller( squeak_rate_limiter, payment_processor, peer_server, + peer_client, connection_manager, config, ) @@ -99,6 +107,7 @@ def regtest_squeak_controller( squeak_rate_limiter, payment_processor, peer_server, + peer_client, connection_manager, regtest_config, ): @@ -108,6 +117,7 @@ def regtest_squeak_controller( squeak_rate_limiter, payment_processor, peer_server, + peer_client, connection_manager, regtest_config, )