modules/python_bitcoin/__init__.py

   1 import bitcoin
   2 from processor import Processor
   3 import threading
   4 import time
   5
   6 import composed
   7
   8 class Backend:
   9
  10     def __init__(self):
  11         # Create 3 thread-pools each with 1 thread
  12         self.network_service = bitcoin.async_service(1)
  13         self.disk_service = bitcoin.async_service(1)
  14         self.mempool_service = bitcoin.async_service(1)
  15
  16         self.hosts = bitcoin.hosts(self.network_service)
  17         self.handshake = bitcoin.handshake(self.network_service)
  18         self.network = bitcoin.network(self.network_service)
  19         self.protocol = bitcoin.protocol(self.network_service, self.hosts,
  20                                          self.handshake, self.network)
  21
  22         db_prefix = "/home/genjix/libbitcoin/database"
  23         self.blockchain = bitcoin.bdb_blockchain(self.disk_service, db_prefix)
  24         self.poller = bitcoin.poller(self.blockchain)
  25         self.transaction_pool = \
  26             bitcoin.transaction_pool(self.mempool_service, self.blockchain)
  27
  28         self.protocol.subscribe_channel(self.monitor_tx)
  29         self.session = \
  30             bitcoin.session(self.hosts, self.handshake, self.network,
  31                             self.protocol, self.blockchain, self.poller,
  32                             self.transaction_pool)
  33         self.session.start(self.handle_start)
  34
  35     def handle_start(self, ec):
  36         if ec:
  37             print "Error starting backend:", ec
  38
  39     def stop(self):
  40         self.session.stop(self.handle_stop)
  41
  42     def handle_stop(self, ec):
  43         if ec:
  44             print "Error stopping backend:", ec
  45         print "Stopped backend"
  46
  47     def monitor_tx(self, node):
  48         # We will be notified here when connected to new bitcoin nodes
  49         # Here we subscribe to new transactions from them which we
  50         # add to the transaction_pool. That way we can track which
  51         # transactions we are interested in.
  52         node.subscribe_transaction(
  53             bitcoin.bind(self.recv_tx, bitcoin._1, bitcoin._2, node))
  54         # Re-subscribe to next new node
  55         self.protocol.subscribe_channel(self.monitor_tx)
  56
  57     def recv_tx(self, ec, tx, node):
  58         if ec:
  59             print "Error with new transaction:", ec
  60             return
  61         tx_hash = bitcoin.hash_transaction(tx)
  62         # If we want to ignore this transaction, we can set
  63         # the 2 handlers to be null handlers that do nothing.
  64         self.transaction_pool.store(tx,
  65             bitcoin.bind(self.tx_confirmed, bitcoin._1, tx_hash),
  66             bitcoin.bind(self.handle_mempool_store, bitcoin._1, tx_hash))
  67         # Re-subscribe to new transactions from node
  68         node.subscribe_transaction(
  69             bitcoin.bind(self.recv_tx, bitcoin._1, bitcoin._2, node))
  70
  71     def handle_mempool_store(self, ec, tx_hash):
  72         if ec:
  73             print "Error storing memory pool transaction", tx_hash, ec
  74         else:
  75             print "Accepted transaction", tx_hash
  76
  77     def tx_confirmed(self, ec, tx_hash):
  78         if ec:
  79             print "Problem confirming transaction", tx_hash, ec
  80         else:
  81             print "Confirmed", tx_hash
  82
  83 class GhostValue:
  84
  85     def __init__(self):
  86         self.event = threading.Event()
  87         self.value = None
  88
  89     def get(self):
  90         self.event.wait()
  91         return self.value
  92
  93     def set(self, value):
  94         self.value = value
  95         self.event.set()
  96
  97 class NumblocksSubscribe:
  98
  99     def __init__(self, backend, processor):
 100         self.backend = backend
 101         self.processor = processor
 102         self.lock = threading.Lock()
 103         self.backend.blockchain.subscribe_reorganize(self.reorganize)
 104         self.backend.blockchain.fetch_last_depth(self.set_last_depth)
 105         self.latest = GhostValue()
 106
 107     def set_last_depth(self, ec, last_depth):
 108         if ec:
 109             print "Error retrieving last depth", ec
 110         else:
 111             self.latest.set(last_depth)
 112
 113     def reorganize(self, ec, fork_point, arrivals, replaced):
 114         latest = fork_point + len(arrivals)
 115         self.latest.set(latest)
 116         self.processor.push_response({"method":"numblocks.subscribe", "result": latest})
 117         self.backend.blockchain.subscribe_reorganize(self.reorganize)
 118
 119
 120 class AddressGetHistory:
 121
 122     def __init__(self, backend, processor):
 123         self.backend = backend
 124         self.processor = processor
 125
 126     def get(self, request):
 127         address = str(request["params"])
 128         composed.payment_history(self.backend.blockchain, address,
 129             bitcoin.bind(self.respond, request, bitcoin._1))
 130
 131     def respond(self, request, result):
 132         self.processor.push_response({"id": request["id"], "method":request["method"], "params":request["params"], "result": result})
 133
 134
 135 class BlockchainProcessor(Processor):
 136
 137     def __init__(self, config):
 138         Processor.__init__(self)
 139         self.backend = Backend()
 140         self.numblocks_subscribe = NumblocksSubscribe(self.backend, self)
 141         self.address_get_history = AddressGetHistory(self.backend, self)
 142
 143     def stop(self):
 144         self.backend.stop()
 145
 146     def process(self, request):
 147
 148         print "New request (lib)", request
 149         if request["method"] == "numblocks.subscribe":
 150             self.numblocks_subscribe.subscribe(session, request)
 151         elif request["method"] == "address.get_history":
 152             self.address_get_history.get(request)
 153         elif request["method"] == "server.banner":
 154             self.push_response({"id": request["id"], "method": request["method"], "params":request["params"],
 155                 "result": "libbitcoin using python-bitcoin bindings"})
 156         elif request["method"] == "transaction.broadcast":
 157             self.broadcast_transaction(request)
 158         # Execute and when ready, you call
 159         # self.push_response(response)
 160
 161     def broadcast_transaction(self, request):
 162         raw_tx = bitcoin.data_chunk(str(request["params"]))
 163         exporter = bitcoin.satoshi_exporter()
 164         try:
 165             tx = exporter.load_transaction(raw_tx)
 166         except RuntimeError:
 167             response = {"id": request["id"], "method": request["method"], "params":request["params"], "result": None,
 168                 "error": {"message":
 169                     "Exception while parsing the transaction data.",
 170                     "code": -4}}
 171         else:
 172             self.backend.protocol.broadcast_transaction(tx)
 173             tx_hash = str(bitcoin.hash_transaction(tx))
 174             response = {"id": request["id"], "method": request["method"], "params":request["params"], "result": tx_hash}
 175         self.push_response(response)
 176
 177     def run(self):
 178         # this class is a thread. it does nothing in this example.
 179         print "Warning: pre-alpha prototype. Full of bugs."
 180         while not self.shared.stopped():
 181             time.sleep(1)
 182