lbry-sdk/torba/basenetwork.py

116 lines
4.1 KiB
Python
Raw Normal View History

2018-05-25 08:03:25 +02:00
import logging
import asyncio
2018-10-17 19:32:45 +02:00
from asyncio import CancelledError
from itertools import cycle
2018-10-15 04:16:51 +02:00
from aiorpcx import ClientSession as BaseClientSession
2018-05-25 08:03:25 +02:00
from torba import __version__
from torba.stream import StreamController
2018-07-01 23:20:17 +02:00
log = logging.getLogger(__name__)
2018-05-25 08:03:25 +02:00
2018-10-15 04:16:51 +02:00
class ClientSession(BaseClientSession):
2018-05-25 08:03:25 +02:00
2018-10-15 04:16:51 +02:00
def __init__(self, *args, network, **kwargs):
2018-05-25 08:03:25 +02:00
self.network = network
2018-10-15 04:16:51 +02:00
super().__init__(*args, **kwargs)
self._on_disconnect_controller = StreamController()
self.on_disconnected = self._on_disconnect_controller.stream
2018-10-17 19:05:47 +02:00
self.bw_limit = self.framer.max_size = self.max_errors = 1 << 32
2018-10-15 04:16:51 +02:00
async def handle_request(self, request):
controller = self.network.subscription_controllers[request.method]
controller.add(request.args)
2018-05-25 08:03:25 +02:00
2018-10-15 04:16:51 +02:00
def connection_lost(self, exc):
super().connection_lost(exc)
self._on_disconnect_controller.add(True)
2018-05-25 08:03:25 +02:00
class BaseNetwork:
2018-06-08 05:47:46 +02:00
def __init__(self, ledger):
self.config = ledger.config
2018-10-15 04:16:51 +02:00
self.client: ClientSession = None
2018-05-25 08:03:25 +02:00
self.running = False
self._on_connected_controller = StreamController()
self.on_connected = self._on_connected_controller.stream
self._on_header_controller = StreamController()
self.on_header = self._on_header_controller.stream
self._on_status_controller = StreamController()
self.on_status = self._on_status_controller.stream
self.subscription_controllers = {
'blockchain.headers.subscribe': self._on_header_controller,
'blockchain.address.subscribe': self._on_status_controller,
2018-05-25 08:03:25 +02:00
}
2018-10-15 04:16:51 +02:00
async def start(self):
2018-09-17 20:32:16 +02:00
self.running = True
delay = 0.0
2018-05-25 08:03:25 +02:00
for server in cycle(self.config['default_servers']):
2018-10-16 17:56:53 +02:00
connection_string = '{}:{}'.format(*server)
2018-10-15 04:16:51 +02:00
self.client = ClientSession(*server, network=self)
2018-05-25 08:03:25 +02:00
try:
2018-10-15 04:16:51 +02:00
await self.client.create_connection()
await self.ensure_server_version()
2018-10-16 17:56:53 +02:00
log.info("Successfully connected to SPV wallet server: %s", connection_string)
2018-05-25 08:03:25 +02:00
self._on_connected_controller.add(True)
delay = 0.0
2018-10-15 04:16:51 +02:00
await self.client.on_disconnected.first
except (Exception, CancelledError): # pylint: disable=broad-except
log.exception("Connecting to %s raised an exception:", connection_string)
2018-05-25 08:03:25 +02:00
if not self.running:
return
elif self.client:
await self.client.close()
self.client.connection.cancel_pending_requests()
await asyncio.sleep(delay)
delay = min(delay + 1.0, 10.0)
2018-05-25 08:03:25 +02:00
2018-10-15 04:16:51 +02:00
async def stop(self):
2018-05-25 08:03:25 +02:00
self.running = False
if self.is_connected:
2018-10-15 06:45:21 +02:00
disconnected = self.client.on_disconnected.first
2018-10-15 04:16:51 +02:00
await self.client.close()
2018-10-15 06:45:21 +02:00
await disconnected
2018-05-25 08:03:25 +02:00
@property
def is_connected(self):
2018-10-15 04:16:51 +02:00
return self.client is not None and not self.client.is_closing()
2018-05-25 08:03:25 +02:00
def rpc(self, list_or_method, *args):
if self.is_connected:
2018-10-15 04:16:51 +02:00
return self.client.send_request(list_or_method, args)
2018-05-25 08:03:25 +02:00
else:
raise ConnectionError("Attempting to send rpc request when connection is not available.")
def ensure_server_version(self, required='1.2'):
return self.rpc('server.version', __version__, required)
def broadcast(self, raw_transaction):
return self.rpc('blockchain.transaction.broadcast', raw_transaction)
def get_history(self, address):
return self.rpc('blockchain.address.get_history', address)
def get_transaction(self, tx_hash):
return self.rpc('blockchain.transaction.get', tx_hash)
def get_merkle(self, tx_hash, height):
return self.rpc('blockchain.transaction.get_merkle', tx_hash, height)
def get_headers(self, height, count=10000):
return self.rpc('blockchain.block.headers', height, count)
def subscribe_headers(self):
return self.rpc('blockchain.headers.subscribe', True)
def subscribe_address(self, address):
return self.rpc('blockchain.address.subscribe', address)