lbry-sdk/torba/basenetwork.py

227 lines
7.9 KiB
Python
Raw Normal View History

2018-05-25 08:03:25 +02:00
import json
import socket
import logging
from itertools import cycle
from twisted.internet import defer, reactor, protocol
from twisted.application.internet import ClientService, CancelledError
from twisted.internet.endpoints import clientFromString
from twisted.protocols.basic import LineOnlyReceiver
from twisted.python import failure
2018-05-25 08:03:25 +02:00
from torba import __version__
from torba.stream import StreamController
from torba.constants import TIMEOUT
2018-05-25 08:03:25 +02:00
2018-07-01 23:20:17 +02:00
log = logging.getLogger(__name__)
2018-05-25 08:03:25 +02:00
class StratumClientProtocol(LineOnlyReceiver):
delimiter = b'\n'
2018-07-01 23:20:17 +02:00
MAX_LENGTH = 2000000
2018-05-25 08:03:25 +02:00
def __init__(self):
self.request_id = 0
self.lookup_table = {}
self.session = {}
self.network = None
2018-05-25 08:03:25 +02:00
self.on_disconnected_controller = StreamController()
self.on_disconnected = self.on_disconnected_controller.stream
def _get_id(self):
self.request_id += 1
return self.request_id
@property
def _ip(self):
return self.transport.getPeer().host
def get_session(self):
return self.session
def connectionMade(self):
try:
self.transport.setTcpNoDelay(True)
self.transport.setTcpKeepAlive(True)
if hasattr(socket, "TCP_KEEPIDLE"):
2018-09-20 19:29:09 +02:00
self.transport.socket.setsockopt(
socket.SOL_TCP, socket.TCP_KEEPIDLE, 120
# Seconds before sending keepalive probes
)
else:
log.debug("TCP_KEEPIDLE not available")
if hasattr(socket, "TCP_KEEPINTVL"):
2018-09-20 19:29:09 +02:00
self.transport.socket.setsockopt(
socket.SOL_TCP, socket.TCP_KEEPINTVL, 1
# Interval in seconds between keepalive probes
)
else:
log.debug("TCP_KEEPINTVL not available")
if hasattr(socket, "TCP_KEEPCNT"):
2018-09-20 19:29:09 +02:00
self.transport.socket.setsockopt(
socket.SOL_TCP, socket.TCP_KEEPCNT, 5
# Failed keepalive probles before declaring other end dead
)
else:
log.debug("TCP_KEEPCNT not available")
except Exception as err: # pylint: disable=broad-except
2018-05-25 08:03:25 +02:00
# Supported only by the socket transport,
# but there's really no better place in code to trigger this.
log.warning("Error setting up socket: %s", err)
def connectionLost(self, reason=None):
self.connected = 0
2018-05-25 08:03:25 +02:00
self.on_disconnected_controller.add(True)
for deferred in self.lookup_table.values():
if not deferred.called:
deferred.errback(TimeoutError("Connection dropped."))
2018-05-25 08:03:25 +02:00
def lineReceived(self, line):
log.debug('received: %s', line)
2018-05-25 08:03:25 +02:00
try:
message = json.loads(line)
2018-05-25 08:03:25 +02:00
except (ValueError, TypeError):
raise ValueError("Cannot decode message '{}'".format(line.strip()))
if message.get('id'):
try:
d = self.lookup_table.pop(message['id'])
if message.get('error'):
2018-06-14 02:57:57 +02:00
d.errback(RuntimeError(message['error']))
2018-05-25 08:03:25 +02:00
else:
d.callback(message.get('result'))
except KeyError:
raise LookupError(
"Lookup for deferred object for message ID '{}' failed.".format(message['id']))
elif message.get('method') in self.network.subscription_controllers:
controller = self.network.subscription_controllers[message['method']]
controller.add(message.get('params'))
else:
log.warning("Cannot handle message '%s'", line)
2018-05-25 08:03:25 +02:00
def rpc(self, method, *args):
message_id = self._get_id()
message = json.dumps({
'id': message_id,
'method': method,
'params': args
2018-05-25 08:03:25 +02:00
})
log.debug('sent: %s', message)
2018-05-25 08:03:25 +02:00
self.sendLine(message.encode('latin-1'))
d = self.lookup_table[message_id] = defer.Deferred()
d.addTimeout(
TIMEOUT, reactor, onTimeoutCancel=lambda *_: failure.Failure(TimeoutError(
"Timeout: Stratum request for '%s' took more than %s seconds" % (method, TIMEOUT)))
)
2018-05-25 08:03:25 +02:00
return d
class StratumClientFactory(protocol.ClientFactory):
protocol = StratumClientProtocol
def __init__(self, network):
self.network = network
self.client = None
def buildProtocol(self, addr):
client = self.protocol()
client.factory = self
client.network = self.network
self.client = client
return client
class BaseNetwork:
2018-06-08 05:47:46 +02:00
def __init__(self, ledger):
self.config = ledger.config
2018-05-25 08:03:25 +02:00
self.client = None
self.service = None
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
}
@defer.inlineCallbacks
def start(self):
2018-09-17 20:32:16 +02:00
self.running = True
2018-05-25 08:03:25 +02:00
for server in cycle(self.config['default_servers']):
connection_string = 'tcp:{}:{}'.format(*server)
endpoint = clientFromString(reactor, connection_string)
log.debug("Attempting connection to SPV wallet server: %s", connection_string)
2018-05-25 08:03:25 +02:00
self.service = ClientService(endpoint, StratumClientFactory(self))
self.service.startService()
try:
self.client = yield self.service.whenConnected(failAfterFailures=2)
yield self.ensure_server_version()
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)
yield self.client.on_disconnected.first
except CancelledError:
return
except Exception: # pylint: disable=broad-except
log.exception("Connecting to %s raised an exception:", connection_string)
2018-05-25 08:03:25 +02:00
finally:
self.client = None
if self.service is not None:
self.service.stopService()
2018-05-25 08:03:25 +02:00
if not self.running:
return
def stop(self):
self.running = False
if self.service is not None:
self.service.stopService()
if self.is_connected:
return self.client.on_disconnected.first
else:
return defer.succeed(True)
@property
def is_connected(self):
return self.client is not None and self.client.connected
def rpc(self, list_or_method, *args):
if self.is_connected:
return self.client.rpc(list_or_method, *args)
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)