-
Notifications
You must be signed in to change notification settings - Fork 634
/
handlers.py
64 lines (52 loc) · 1.82 KB
/
handlers.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
from abc import ABC, abstractmethod
from typing import (
Any,
Awaitable,
Callable,
Dict,
Iterator,
Tuple,
Type,
)
from mypy_extensions import DefaultArg
from eth.rlp.headers import BlockHeader
from p2p.peer import BasePeer
from trinity.protocol.common.exchanges import (
BaseExchange,
)
from trinity.protocol.common.managers import (
ExchangeManager,
)
class BaseExchangeHandler(ABC):
@property
@abstractmethod
def _exchange_config(self) -> Dict[str, Type[BaseExchange[Any, Any, Any]]]:
pass
def __init__(self, peer: BasePeer) -> None:
self._peer = peer
for attr, exchange_cls in self._exchange_config.items():
if hasattr(self, attr):
raise AttributeError(
"Unable to set manager on attribute `{0}` which is already "
"present on the class: {1}".format(attr, getattr(self, attr))
)
manager: ExchangeManager[Any, Any, Any]
manager = ExchangeManager(self._peer, exchange_cls.response_cmd_type, peer.cancel_token)
exchange = exchange_cls(manager)
setattr(self, attr, exchange)
def __iter__(self) -> Iterator[BaseExchange[Any, Any, Any]]:
for key in self._exchange_config.keys():
yield getattr(self, key)
def get_stats(self) -> Dict[str, str]:
return {
exchange.response_cmd_type.__name__: exchange.tracker.get_stats()
for exchange
in self
}
# This class is only needed to please mypy for type checking
BlockHeadersCallable = Callable[
[BaseExchangeHandler, int, int, DefaultArg(int, 'skip'), DefaultArg(int, 'reverse')],
Awaitable[Tuple[BlockHeader, ...]]
]
class BaseChainExchangeHandler(BaseExchangeHandler):
get_block_headers: BlockHeadersCallable