Make JSON-RPC Server an isolated plugin

Author: cburgdorf

eth/chains/base.py

@@ -904,3 +904,15 @@ async def coro_validate_receipt(self,
                                     receipt: Receipt,
                                     at_header: BlockHeader) -> None:
         raise NotImplementedError()
+
+    async def coro_get_block_by_hash(self,
+                                     block_hash: Hash32) -> BaseBlock:
+        raise NotImplementedError()
+
+    async def coro_get_block_by_header(self,
+                                       header: BlockHeader) -> BaseBlock:
+        raise NotImplementedError()
+
+    async def coro_get_canonical_block_by_number(self,
+                                                 block_number: BlockNumber) -> BaseBlock:
+        raise NotImplementedError()

eth/tools/fixtures/helpers.py

@@ -146,12 +146,12 @@ def genesis_params_from_fixture(fixture):
     }
 
 
-def new_chain_from_fixture(fixture):
+def new_chain_from_fixture(fixture, chain_cls=MainnetChain):
     base_db = AtomicDB()
 
     vm_config = chain_vm_configuration(fixture)
 
-    ChainFromFixture = MainnetChain.configure(
+    ChainFromFixture = chain_cls.configure(
         'ChainFromFixture',
         vm_configuration=vm_config,
     )

p2p/events.py

@@ -0,0 +1,21 @@
+from typing import (
+    Type,
+)
+
+from lahja import (
+    BaseEvent,
+    BaseRequestResponseEvent,
+)
+
+
+class PeerCountResponse(BaseEvent):
+
+    def __init__(self, peer_count: int) -> None:
+        self.peer_count = peer_count
+
+
+class PeerCountRequest(BaseRequestResponseEvent[PeerCountResponse]):
+
+    @staticmethod
+    def expected_response_type() -> Type[PeerCountResponse]:
+        return PeerCountResponse

p2p/peer.py

@@ -48,6 +48,10 @@
 
 from cancel_token import CancelToken, OperationCancelled
 
+from lahja import (
+    Endpoint,
+)
+
 from eth.constants import GENESIS_BLOCK_NUMBER
 from eth.rlp.headers import BlockHeader
 from eth.vm.base import BaseVM
@@ -95,6 +99,10 @@
     MAC_LEN,
 )
 
+from .events import (
+    PeerCountRequest,
+    PeerCountResponse,
+)
 
 if TYPE_CHECKING:
     from trinity.db.header import BaseAsyncHeaderDB  # noqa: F401
@@ -173,7 +181,7 @@ async def _run(self) -> None:
             # exit via exception rather than cleanly shutting down.  By using
             # `run_task`, this service finishes exiting prior to the
             # cancellation.
-            self.run_task(self.peer.disconnect(DisconnectReason.useless_peer))
+            self.run_daemon_task(self.peer.disconnect(DisconnectReason.useless_peer))
 
     async def ensure_same_side_on_dao_fork(self) -> None:
         """Ensure we're on the same side of the DAO fork as the given peer.
@@ -841,6 +849,7 @@ def __init__(self,
                  context: BasePeerContext,
                  max_peers: int = DEFAULT_MAX_PEERS,
                  token: CancelToken = None,
+                 event_bus: Endpoint = None
                  ) -> None:
         super().__init__(token)
 
@@ -850,6 +859,17 @@ def __init__(self,
 
         self.connected_nodes: Dict[Node, BasePeer] = {}
         self._subscribers: List[PeerSubscriber] = []
+        self.event_bus = event_bus
+        if self.event_bus is not None:
+            self.run_task(self.handle_peer_count_requests())
+
+    async def handle_peer_count_requests(self) -> None:
+        async for req in self.event_bus.stream(PeerCountRequest):
+            # We are listening for all `PeerCountRequest` events but we ensure to
+            # only send a `PeerCountResponse` to the callsite that made the request.
+            # We do that by retrieving a `BroadcastConfig` from the request via the
+            # `event.broadcast_config()` API.
+            self.event_bus.broadcast(PeerCountResponse(len(self)), req.broadcast_config())
 
     def __len__(self) -> int:
         return len(self.connected_nodes)

tests/conftest.py

@@ -78,8 +78,7 @@ def funded_address_initial_balance():
     return to_wei(1000, 'ether')
 
 
-@pytest.fixture
-def chain_with_block_validation(base_db, genesis_state):
+def _chain_with_block_validation(base_db, genesis_state, chain_cls=Chain):
     """
     Return a Chain object containing just the genesis block.
 
@@ -107,7 +106,8 @@ def chain_with_block_validation(base_db, genesis_state):
         "transaction_root": decode_hex("56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421"),  # noqa: E501
         "uncles_hash": decode_hex("1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347")  # noqa: E501
     }
-    klass = Chain.configure(
+
+    klass = chain_cls.configure(
         __name__='TestChain',
         vm_configuration=(
             (constants.GENESIS_BLOCK_NUMBER, SpuriousDragonVM),
@@ -118,6 +118,11 @@ def chain_with_block_validation(base_db, genesis_state):
     return chain
 
 
+@pytest.fixture
+def chain_with_block_validation(base_db, genesis_state):
+    return _chain_with_block_validation(base_db, genesis_state)
+
+
 def import_block_without_validation(chain, block):
     return super(type(chain), chain).import_block(block, perform_validation=False)
 

tests/trinity/conftest.py

@@ -5,11 +5,17 @@
 import tempfile
 import uuid
 
-from p2p.tools.paragon import (
-    ParagonContext,
-    ParagonPeerPool,
+from lahja import (
+    EventBus,
 )
 
+from eth.chains import (
+    Chain,
+)
+
+from trinity.chains.coro import (
+    AsyncChainMixin,
+)
 from trinity.rpc.main import (
     RPCServer,
 )
@@ -22,6 +28,13 @@
 from trinity.utils.filesystem import (
     is_under_path,
 )
+from tests.conftest import (
+    _chain_with_block_validation,
+)
+
+
+class TestAsyncChain(Chain, AsyncChainMixin):
+    pass
 
 
 def pytest_addoption(parser):
@@ -51,26 +64,43 @@ def event_loop():
         loop.close()
 
 
+@pytest.fixture(scope='module')
+def event_bus(event_loop):
+    bus = EventBus()
+    endpoint = bus.create_endpoint('test')
+    bus.start(event_loop)
+    endpoint.connect(event_loop)
+    try:
+        yield endpoint
+    finally:
+        endpoint.stop()
+        bus.stop()
+
+
 @pytest.fixture(scope='session')
 def jsonrpc_ipc_pipe_path():
     with tempfile.TemporaryDirectory() as temp_dir:
         yield Path(temp_dir) / '{0}.ipc'.format(uuid.uuid4())
 
 
+@pytest.fixture
+def chain_with_block_validation(base_db, genesis_state):
+    return _chain_with_block_validation(base_db, genesis_state, TestAsyncChain)
+
+
 @pytest.mark.asyncio
 @pytest.fixture
 async def ipc_server(
         monkeypatch,
+        event_bus,
         jsonrpc_ipc_pipe_path,
         event_loop,
         chain_with_block_validation):
     '''
     This fixture runs a single RPC server over IPC over
     the course of all tests. It yields the IPC server only for monkeypatching purposes
     '''
-    peer_pool = ParagonPeerPool(b'unicornsrainbows' * 2, ParagonContext())
-
-    rpc = RPCServer(chain_with_block_validation, peer_pool)
+    rpc = RPCServer(chain_with_block_validation, event_bus)
     ipc_server = IPCServer(rpc, jsonrpc_ipc_pipe_path, loop=event_loop)
 
     asyncio.ensure_future(ipc_server.run(), loop=event_loop)

tests/trinity/core/json-rpc/test_ipc.py

@@ -14,6 +14,11 @@
     to_hex,
 )
 
+from p2p.events import (
+    PeerCountRequest,
+    PeerCountResponse,
+)
+
 from trinity.utils.version import construct_trinity_client_identifier
 
 
@@ -35,15 +40,6 @@ def build_request(method, params=[]):
     return json.dumps(request).encode()
 
 
-class MockPeerPool:
-
-    def __init__(self, peer_count=0):
-        self.peer_count = peer_count
-
-    def __len__(self):
-        return self.peer_count
-
-
 def id_from_rpc_request(param):
     if isinstance(param, bytes):
         request = json.loads(param.decode())
@@ -68,6 +64,7 @@ async def get_ipc_response(
         jsonrpc_ipc_pipe_path,
         request_msg,
         event_loop):
+
     assert wait_for(jsonrpc_ipc_pipe_path), "IPC server did not successfully start with IPC file"
 
     reader, writer = await asyncio.open_unix_connection(str(jsonrpc_ipc_pipe_path), loop=event_loop)
@@ -399,18 +396,27 @@ async def test_eth_call_with_contract_on_ipc(
     assert result == expected
 
 
+def mock_peer_count(count):
+    async def mock_event_bus_interaction(bus):
+        async for req in bus.stream(PeerCountRequest):
+            bus.broadcast(PeerCountResponse(count), req.broadcast_config())
+            break
+
+    return mock_event_bus_interaction
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
-    'request_msg, mock_peer_pool, expected',
+    'request_msg, event_bus_setup_fn, expected',
     (
         (
             build_request('net_peerCount'),
-            MockPeerPool(peer_count=1),
+            mock_peer_count(1),
             {'result': '0x1', 'id': 3, 'jsonrpc': '2.0'},
         ),
         (
             build_request('net_peerCount'),
-            MockPeerPool(peer_count=0),
+            mock_peer_count(0),
             {'result': '0x0', 'id': 3, 'jsonrpc': '2.0'},
         ),
     ),
@@ -422,10 +428,17 @@ async def test_peer_pool_over_ipc(
         monkeypatch,
         jsonrpc_ipc_pipe_path,
         request_msg,
-        mock_peer_pool,
+        event_bus_setup_fn,
+        event_bus,
         expected,
         event_loop,
         ipc_server):
-    monkeypatch.setattr(ipc_server.rpc.modules['net'], '_peer_pool', mock_peer_pool)
-    result = await get_ipc_response(jsonrpc_ipc_pipe_path, request_msg, event_loop)
+
+    asyncio.ensure_future(event_bus_setup_fn(event_bus))
+
+    result = await get_ipc_response(
+        jsonrpc_ipc_pipe_path,
+        request_msg,
+        event_loop
+    )
     assert result == expected

tests/trinity/json-fixtures-over-rpc/test_rpc_fixtures.py

@@ -25,6 +25,9 @@
     should_run_slow_tests,
 )
 
+from trinity.chains.mainnet import (
+    MainnetFullChain
+)
 from trinity.rpc import RPCServer
 from trinity.rpc.format import (
     empty_to_0x,
@@ -383,7 +386,7 @@ def chain(chain_without_block_validation):
 
 @pytest.mark.asyncio
 async def test_rpc_against_fixtures(chain, ipc_server, chain_fixture, fixture_data):
-    rpc = RPCServer(None)
+    rpc = RPCServer(MainnetFullChain(None))
 
     setup_result, setup_error = await call_rpc(rpc, 'evm_resetToGenesisFixture', [chain_fixture])
     assert setup_error is None and setup_result is True, "cannot load chain for %r" % fixture_data

trinity/chains/coro.py

@@ -0,0 +1,10 @@
+from trinity.utils.async_dispatch import (
+    async_method,
+)
+
+
+class AsyncChainMixin:
+
+    coro_get_canonical_block_by_number = async_method('get_canonical_block_by_number')
+    coro_get_block_by_hash = async_method('get_block_by_hash')
+    coro_get_block_by_header = async_method('get_block_by_header')