#!/usr/bin/env python # # Electrum - lightweight Bitcoin client # Copyright (C) 2014 Thomas Voegtlin # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . import sys import traceback import threading import Queue import util from network import Network, serialize_proxy, serialize_server from simple_config import SimpleConfig class NetworkProxy(util.DaemonThread): def __init__(self, socket, config=None): if config is None: config = {} # Do not use mutables as default arguments! util.DaemonThread.__init__(self) self.config = SimpleConfig(config) if type(config) == type({}) else config self.message_id = 0 self.unanswered_requests = {} self.subscriptions = {} self.debug = False self.lock = threading.Lock() self.callbacks = {} if socket: self.pipe = util.SocketPipe(socket) self.network = None else: self.pipe = util.QueuePipe() self.network = Network(self.pipe, config) self.network.start() for key in ['fee','status','banner','updated','servers','interfaces']: value = self.network.get_status_value(key) self.pipe.get_queue.put({'method':'network.status', 'params':[key, value]}) # status variables self.status = 'unknown' self.servers = {} self.banner = '' self.blockchain_height = 0 self.server_height = 0 self.interfaces = [] # value returned by estimatefee self.fee = None def run(self): while self.is_running(): self.run_jobs() # Synchronizer and Verifier try: response = self.pipe.get() except util.timeout: continue if response is None: break # Protect against ill-formed or malicious server responses try: self.process(response) except: traceback.print_exc(file=sys.stderr) self.trigger_callback('stop') if self.network: self.network.stop() self.print_error("stopped") def process(self, response): if self.debug: self.print_error("<--", response) if response.get('method') == 'network.status': key, value = response.get('params') if key == 'status': self.status = value elif key == 'banner': self.banner = value elif key == 'fee': self.fee = value elif key == 'updated': self.blockchain_height, self.server_height = value elif key == 'servers': self.servers = value elif key == 'interfaces': self.interfaces = value if key in ['status', 'updated']: self.trigger_callback(key) else: self.trigger_callback(key, (value,)) return msg_id = response.get('id') result = response.get('result') error = response.get('error') if msg_id is not None: with self.lock: method, params, callback = self.unanswered_requests.pop(msg_id) else: method = response.get('method') params = response.get('params') with self.lock: for k,v in self.subscriptions.items(): if (method, params) in v: callback = k break else: self.print_error("received unexpected notification", method, params) return r = {'method':method, 'params':params, 'result':result, 'id':msg_id, 'error':error} callback(r) def send(self, messages, callback): """return the ids of the requests that we sent""" # detect subscriptions sub = [] for message in messages: m, v = message if m[-10:] == '.subscribe': sub.append(message) if sub: with self.lock: if self.subscriptions.get(callback) is None: self.subscriptions[callback] = [] for message in sub: if message not in self.subscriptions[callback]: self.subscriptions[callback].append(message) with self.lock: requests = [] ids = [] for m in messages: method, params = m request = { 'id':self.message_id, 'method':method, 'params':params } self.unanswered_requests[self.message_id] = method, params, callback ids.append(self.message_id) requests.append(request) if self.debug: self.print_error("-->", request) self.message_id += 1 self.pipe.send_all(requests) return ids def synchronous_get(self, requests, timeout=100000000): queue = Queue.Queue() ids = self.send(requests, queue.put) id2 = ids[:] res = {} while ids: r = queue.get(True, timeout) _id = r.get('id') ids.remove(_id) if r.get('error'): raise BaseException(r.get('error')) result = r.get('result') res[_id] = r.get('result') out = [] for _id in id2: out.append(res[_id]) return out def get_servers(self): return self.servers def get_interfaces(self): return self.interfaces def get_local_height(self): return self.blockchain_height def get_server_height(self): return self.server_height def is_connected(self): return self.status == 'connected' def is_connecting(self): return self.status == 'connecting' def is_up_to_date(self): return self.unanswered_requests == {} def get_parameters(self): return self.synchronous_get([('network.get_parameters', [])])[0] def set_parameters(self, host, port, protocol, proxy, auto_connect): proxy_str = serialize_proxy(proxy) server_str = serialize_server(host, port, protocol) self.config.set_key('auto_connect', auto_connect, False) self.config.set_key("proxy", proxy_str, False) self.config.set_key("server", server_str, True) # abort if changes were not allowed by config if self.config.get('server') != server_str or self.config.get('proxy') != proxy_str: return return self.synchronous_get([('network.set_parameters', (host, port, protocol, proxy, auto_connect))])[0] def stop_daemon(self): return self.send([('daemon.stop',[])], None) def register_callback(self, event, callback): with self.lock: if not self.callbacks.get(event): self.callbacks[event] = [] self.callbacks[event].append(callback) def trigger_callback(self, event, params=()): with self.lock: callbacks = self.callbacks.get(event,[])[:] if callbacks: [callback(*params) for callback in callbacks]