server-genjix.py

   1 import bitcoin
   2 import stratum
   3 import threading
   4 import time
   5
   6 class Backend:
   7
   8     def __init__(self):
   9         self.network_service = bitcoin.async_service(1)
  10         self.disk_service = bitcoin.async_service(1)
  11         self.mempool_service = bitcoin.async_service(1)
  12
  13         self.hosts = bitcoin.hosts(self.network_service)
  14         self.handshake = bitcoin.handshake(self.network_service)
  15         self.network = bitcoin.network(self.network_service)
  16         self.protocol = bitcoin.protocol(self.network_service, self.hosts,
  17                                          self.handshake, self.network)
  18
  19         db_prefix = "/home/genjix/libbitcoin/database"
  20         self.blockchain = bitcoin.bdb_blockchain(self.disk_service, db_prefix)
  21         self.poller = bitcoin.poller(self.blockchain)
  22         self.transaction_pool = \
  23             bitcoin.transaction_pool(self.mempool_service, self.blockchain)
  24
  25         self.protocol.subscribe_channel(self.monitor_tx)
  26         self.session = \
  27             bitcoin.session(self.hosts, self.handshake, self.network,
  28                             self.protocol, self.blockchain, self.poller,
  29                             self.transaction_pool)
  30         self.session.start(self.handle_start)
  31
  32     def handle_start(self, ec):
  33         if ec:
  34             print "Error starting backend:", ec
  35
  36     def stop(self):
  37         self.session.stop(self.handle_stop)
  38
  39     def handle_stop(self, ec):
  40         if ec:
  41             print "Error stopping backend:", ec
  42         print "Stopped backend"
  43
  44     def monitor_tx(self, node):
  45         # We will be notified here when connected to new bitcoin nodes
  46         # Here we subscribe to new transactions from them which we
  47         # add to the transaction_pool. That way we can track which
  48         # transactions we are interested in.
  49         node.subscribe_transaction(
  50             bitcoin.bind(self.recv_tx, bitcoin._1, bitcoin._2, node))
  51         # Re-subscribe to next new node
  52         self.protocol.subscribe_channel(self.monitor_tx)
  53
  54     def recv_tx(self, ec, tx, node):
  55         if ec:
  56             print "Error with new transaction:", ec
  57             return
  58         tx_hash = bitcoin.hash_transaction(tx)
  59         self.transaction_pool.store(tx,
  60             bitcoin.bind(self.tx_confirmed, bitcoin._1, tx_hash),
  61             bitcoin.bind(self.handle_mempool_store, bitcoin._1, tx_hash))
  62         # Re-subscribe to new transactions from node
  63         node.subscribe_transaction(
  64             bitcoin.bind(self.recv_tx, bitcoin._1, bitcoin._2, node))
  65
  66     def handle_mempool_store(self, ec, tx_hash):
  67         if ec:
  68             print "Error storing memory pool transaction", tx_hash, ec
  69         else:
  70             print "Accepted transaction", tx_hash
  71
  72     def tx_confirmed(self, ec, tx_hash):
  73         if ec:
  74             print "Problem confirming transaction", tx_hash, ec
  75         else:
  76             print "Confirmed", tx_hash
  77
  78 class NumblocksSubscribe:
  79
  80     def __init__(self, backend):
  81         self.backend = backend
  82         self.lock = threading.Lock()
  83         self.backend.blockchain.subscribe_reorganize(self.reorganize)
  84         self.backend.blockchain.fetch_last_depth(self.set_last_depth)
  85         self.latest = None
  86
  87     def subscribe(self, session, request):
  88         last = self.get_last_depth()
  89         session.push_response({"id": request["id"], "result": last})
  90
  91     def get_last_depth(self):
  92         # Stall until last depth has been set...
  93         # Should use condition variable here instead
  94         while True:
  95             with self.lock:
  96                 last = self.latest
  97             if last is not None:
  98                 return last
  99             time.sleep(0.1)
 100
 101     def set_last_depth(self, ec, last_depth):
 102         if ec:
 103             print "Error retrieving last depth", ec
 104         else:
 105             with self.lock:
 106                 self.latest = last_depth
 107
 108     def reorganize(self, ec, fork_point, arrivals, replaced):
 109         pass
 110
 111 class LibbitcoinProcessor(stratum.Processor):
 112
 113     def __init__(self):
 114         self.backend = Backend()
 115         self.numblocks_subscribe = NumblocksSubscribe(self.backend)
 116         stratum.Processor.__init__(self)
 117
 118     def stop(self):
 119         self.backend.stop()
 120
 121     def process(self, session):
 122         request = session.pop_request()
 123         if request["method"] == "numblocks.subscribe":
 124             self.numblocks_subscribe.subscribe(session, request)
 125         print "New request (lib)", request
 126         # Execute and when ready, you call
 127         # session.push_response(response)
 128
 129 if __name__ == "__main__":
 130     processor = LibbitcoinProcessor()
 131     app = stratum.Stratum()
 132     app.start(processor)
 133