Split peer server client (#901)

* Split peer server and peer client

* Got itest passing

* Catch exception when closing peer socket

* Reconfigure all itests to run

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(

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(

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):

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):

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()

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.

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."""

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)

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)

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()

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,
     )