ITEM_TYPE_TX, ITEM_TYPE_BLOCK, ITEM_TYPE_MERKLEBLOCK = (1, 2, 3)
        INV_TYPES = [None, "Tx", "Block", "Merkle"]
class InvItem(object):
    def __init__(self, item_type, data):
    def __str__(self):
    def __repr__(self):
    def __hash__(self):
    def __eq__(self, other):
    def __ne__(self, other):
    def stream(self, f):
    def parse(self, f): class BitcoinPeerProtocol(asyncio.Protocol):

    MAX_MESSAGE_SIZE = 2*1024*1024

    def __init__(self, magic_header, *args, **kwargs):
        super(BitcoinPeerProtocol, self).__init__(*args, **kwargs)
        self.magic_header = magic_header
        self.peername = ("(unconnected)", 0)
        self.connection_made_future = asyncio.Future()
        self.connection_lost_future = asyncio.Future()
        self._run_handle = None
        self.message_queues = weakref.WeakSet()
        # stats
        self.bytes_read = 0
        self.bytes_writ = 0
        self.connect_start_time = None
        self._tasks = set()

    def new_get_next_message_f(self, filter_f=lambda message_name, data: True, maxsize=0):
        @asyncio.coroutine
        def run(self):
            yield from asyncio.wait_for(self.connection_made_future, timeout=None)
            while True:
                try:
                    message_name, data = yield from self._parse_next_message()
                except EOFError:
                    logging.debug("end of stream %s", self)
                    message_name, data = None, None
                except Exception:
                    logging.exception("error in _parse_next_message")
                    message_name, data = None, None
                for q in self.message_queues:
                    if q.filter_f(message_name, data):
                        q.put_nowait((message_name, data))
                if message_name is None:
                    break

        q = asyncio.Queue(maxsize=maxsize)
        q.filter_f = filter_f
        self.message_queues.add(q)

        @asyncio.coroutine
        def get_next_message():
            msg_name, data = yield from q.get()
            if msg_name is None:
                raise EOFError
            return msg_name, data

        if self._run_handle:
            if self._run_handle.done():
                q.put_nowait((None, None))
        else:
            self._run_handle = asyncio.Task(run(self))
        return get_next_message

    def add_task(self, task):
        """
        Some Task objects are associated with the peer. This method
        gives an easy way to keep a strong reference to a Task that won't
        disappear until the peer does.
        """
        try:
            self._tasks.add(asyncio.ensure_future(task))
        except AttributeError:
            # compatibility for very old versions of python 3
            self._tasks.add(getattr(asyncio, 'async')(task))

    def send_msg(self, message_name, **kwargs):
        message_data = pack_from_data(message_name, **kwargs)
        message_type = message_name.encode("utf8")
        message_type_padded = (message_type+(b'\0'*12))[:12]
        message_size = struct.pack("<L", len(message_data))
        message_checksum = encoding.double_sha256(message_data)[:4]
        packet = b"".join([
            self.magic_header, message_type_padded, message_size, message_checksum, message_data
        ])
        logging.debug("sending message %s [%d bytes] to %s", message_type.decode("utf8"), len(packet), self)
        self.bytes_writ += len(packet)
        self.transport.write(packet)

    def connection_made(self, transport):
        self.connection_made_future.set_result(transport)
        self.transport = transport
        self.reader = asyncio.StreamReader()
        self._is_writable = True
        self.peername = transport.get_extra_info("socket").getpeername()
        self.connect_start_time = time.time()

    def connection_lost(self, exc):
        if exc:
            self.connection_lost_future.set_exception(exc)
        else:
            self.connection_lost_future.set_result(None)
        # ## TODO: fix this. This is a stupid way of doing it.
        # What is the correct way to shut this down?
        self.reader.feed_eof()

    def data_received(self, data):
        self.bytes_read += len(data)
        self.reader.feed_data(data)

    def pause_writing(self):
        self._is_writable = False

    def resume_writing(self):
        self._is_writable = True

    def is_writable(self):
        return self._is_writable

    @asyncio.coroutine
    def _parse_next_message(self):

        # read magic header
        reader = self.reader
        blob = yield from reader.readexactly(len(self.magic_header))
        if blob != self.magic_header:
            raise BitcoinProtocolError("bad magic: got %s" % binascii.hexlify(blob))

        # read message name
        message_size_hash_bytes = yield from reader.readexactly(20)
        message_name_bytes = message_size_hash_bytes[:12]
        message_name = message_name_bytes.replace(b"\0", b"").decode("utf8")

        # get size of message
        size_bytes = message_size_hash_bytes[12:16]
        size = int.from_bytes(size_bytes, byteorder="little")
        if size > self.MAX_MESSAGE_SIZE:
            raise BitcoinProtocolError("absurdly large message size %d" % size)

        # read the hash, then the message
        transmitted_hash = message_size_hash_bytes[16:20]
        message_data = yield from reader.readexactly(size)

        # check the hash
        actual_hash = encoding.double_sha256(message_data)[:4]
        if actual_hash != transmitted_hash:
            raise BitcoinProtocolError("checksum is WRONG: %s instead of %s" % (
                binascii.hexlify(actual_hash), binascii.hexlify(transmitted_hash)))
        logging.debug("message %s: %s (%d byte payload)", self, message_name, len(message_data))

        # parse the blob into a BitcoinProtocolMessage object
        data = parse_from_data(message_name, message_data)
        return message_name, data

    def __lt__(self, other):
        return self.connect_start_time < other.connect_start_time

    def __str__(self):
        return repr(self)

    def __repr__(self):
        return "<Peer %s>" % str(self.peername) class Fetcher:
    """
    Fetching a merkleblock also fetches the transactions that follow, and
    includes them in the message as the "tx" key.
    """
    def __init__(self, peer):
        self.peer = peer
        self.request_q = asyncio.Queue()
        self.futures = weakref.WeakValueDictionary()

        getdata_loop_future = asyncio.Task(self._getdata_loop())
        next_message = peer.new_get_next_message_f(
            filter_f=lambda name, data: name in ["tx", "block", "merkleblock", "notfound"])
        peer.add_task(self._fetch_loop(next_message, getdata_loop_future))

    @asyncio.coroutine
    def fetch(self, inv_item, timeout=None):
        """
        Return the fetched object or None if the remote says it doesn't have it, or
        times out by exceeding `timeout` seconds.
        """
        future = self.futures.get(inv_item)
        if not future:
            future = asyncio.Future()
            self.futures[inv_item] = future
            self.request_q.put_nowait(inv_item)
        try:
            return (yield from asyncio.wait_for(future, timeout=timeout))
        except asyncio.TimeoutError:
            return None

    def queue_size(self):
        pass
        # ## TODO: finish

    @asyncio.coroutine
    def _getdata_loop(self):
        while True:
            so_far = []
            inv_item = yield from self.request_q.get()
            while True:
                so_far.append(inv_item)
                if self.request_q.qsize() == 0 or len(so_far) >= 50000:
                    break
                inv_item = yield from self.request_q.get()
            self.peer.send_msg("getdata", items=so_far)

    @asyncio.coroutine
    def _fetch_loop(self, next_message, getdata_loop_future):
        try:
            while True:
                name, data = yield from next_message()
                ITEM_LOOKUP = dict(tx="tx", block="block", merkleblock="header")
                if name in ITEM_LOOKUP:
                    item = data[ITEM_LOOKUP[name]]
                    the_hash = item.hash()
                    TYPE_DB = {"tx": ITEM_TYPE_TX,
                               "block": ITEM_TYPE_BLOCK,
                               "merkleblock": ITEM_TYPE_MERKLEBLOCK}
                    the_type = TYPE_DB[name]
                    inv_item = InvItem(the_type, the_hash)
                    future = self.futures.get(inv_item)
                    if name == "merkleblock":
                        txs = []
                        for h in data["tx_hashes"]:
                            name, data = yield from next_message()
                            if name != "tx":
                                logging.error(
                                    "insufficient tx messages after merkleblock message: missing %s",
                                    b2h_rev(h))
                                del self.futures[inv_item]
                                future.set_result(None)
                                break
                            tx = data["tx"]
                            if tx.hash() != h:
                                logging.error(
                                    "missing tx message after merkleblock message: missing %s", b2h_rev(h))
                                del self.futures[inv_item]
                                future.set_result(None)
                                break
                            txs.append(tx)
                        item.txs = txs
                    if future is not None:
                        del self.futures[inv_item]
                        if not future.done():
                            future.set_result(item)
                        else:
                            logging.info("got %s unsolicited", item.id())
                if name == "notfound":
                    for inv_item in data["items"]:
                        the_hash = inv_item.data
                        future = self.futures.get(inv_item)
                        if future:
                            del self.futures[inv_item]
                            future.set_result(None)
        except EOFError:
            getdata_loop_future.cancel() def fast_forwarder_add_peer_f(blockchain):
    """
    returns a function "add_peer"
        add_peer(peer, last_block_index)
    which accepts a BitcoinPeerProtocol peer and the last_block_index
    as reported in the initial handshake.
    """
    peer_queue = asyncio.PriorityQueue()

    @asyncio.coroutine
    def _fetch_missing(peer, blockchain):
        next_message = peer.new_get_next_message_f(lambda msg, data: msg == 'block')
        ops = []
        for h in blockchain.chain_finder.missing_parents():
            peer.send_msg("getdata", items=[InvItem(ITEM_TYPE_BLOCK, h)])
            msg, data = yield from next_message()
            block = data["block"]
            ops = blockchain.add_headers([block])
            if len(ops) > 0:
                break
        return ops

    @asyncio.coroutine
    def _run_ff(blockchain):
        # this kind of works, but it's lame because we put
        # peers into a queue, so they'll never be garbage collected
        # even if they vanish. I think.
        while 1:
            priority, (peer, lbi, rate_dict) = yield from peer_queue.get()
            if lbi - blockchain.length() > 0:
                # let's get some headers from this guy!
                start_time = time.time()
                h = blockchain.last_block_hash()
                try:
                    headers = yield from asyncio.wait_for(get_headers_hashes(peer, h), timeout=10)
                except EOFError:
                    # this peer is dead... so don't put it back in the queue
                    continue
                except asyncio.TimeoutError:
                    # this peer timed out. How useless
                    continue
                # TODO: what if the stupid client sends us bogus headers?
                # how will we ever figure this out?
                # answer: go through headers and remove fake ones or ones that we've seen
                # check hash, difficulty, difficulty against hash, and that they form
                # a chain. This make it expensive to produce bogus headers
                time_elapsed = time.time() - start_time
                rate_dict["total_seconds"] += time_elapsed
                rate_dict["records"] += len(headers)
                priority = - rate_dict["records"] / rate_dict["total_seconds"]
                # let's make sure we actually extend the chain
                ops = blockchain.add_headers(headers)
                # ## this hack is necessary because the stupid default client
                # does not send the genesis block!
                try:
                    while len(ops) == 0:
                        ops = yield from asyncio.wait_for(_fetch_missing(peer, blockchain), timeout=30)
                except Exception:
                    logging.exception("problem fetching missing parents")
                if len(ops) > 0:
                    peer_queue.put_nowait((priority, (peer, lbi, rate_dict)))
                # otherwise, this peer is stupid and should be ignored

    def add_peer(peer, last_block_index):
        peer_queue.put_nowait((0, (peer, last_block_index, dict(total_seconds=0, records=0))))

    # we need a strong reference to this task
    add_peer.task = asyncio.Task(_run_ff(blockchain))

    return add_peer class Blockfetcher:
    """
    Blockfetcher

    This class is created once for each PeerGroup. It's used to accept a set of blocks
    and download them in an overlapping way.

    It accepts new peers via add_peer.

    It fetches new blocks via get_block_future or get_block.
    """
    def __init__(self, max_q_size=0):
        # this queue accepts tuples of the form:
        #  (block_index, InvItem(ITEM_TYPE_BLOCK, block_hash), future)
        self.block_hash_priority_queue = asyncio.PriorityQueue(max_q_size)

    def add_peer(self, peer, fetcher, last_block_index):
        peer.add_task(self.fetch_from_peer(peer, fetcher, last_block_index))

    def get_block_future(self, block_hash, block_index):
        future = asyncio.Future()
        item = (block_index, InvItem(ITEM_TYPE_BLOCK, block_hash), future)
        self.block_hash_priority_queue.put_nowait(item)
        return future

    def get_merkle_block_future(self, block_hash, block_index):
        future = asyncio.Future()
        item = (block_index, InvItem(ITEM_TYPE_MERKLEBLOCK, block_hash), future)
        self.block_hash_priority_queue.put_nowait(item)
        return future

    @asyncio.coroutine
    def get_block(self, block_hash, block_index):
        future = self.get_block_future(block_hash, block_index)
        block = asyncio.wait_for(future, timeout=None)
        return block

    # for each peer, a loop:
    #   pull an item out of priority queue
    #   if we should have it, add it to temp list
    #   repeat until N blocks or queue empty
    #   attempt to fetch N blocks
    #     reset after no progress after X seconds
    #   on failure, replace back in queue (noting this peer doesn't have it)
    #   on success
    #     if less than X/3 seconds used, double N

    @asyncio.coroutine
    def fetch_from_peer(self, peer, fetcher, last_block_index):
        missing = set()
        per_loop = 1
        loop_timeout = 30
        while True:
            items_to_try = []
            items_to_not_try = []
            while True:
                item = yield from self.block_hash_priority_queue.get()
                if item in missing or item[0] > last_block_index:
                    items_to_not_try.append(item)
                else:
                    items_to_try.append(item)
                if self.block_hash_priority_queue.empty():
                    break
                if len(items_to_try) >= per_loop or item[0] > last_block_index:
                    break
            for item in items_to_not_try:
                self.block_hash_priority_queue.put_nowait(item)
            start_time = time.time()
            futures = []
            for item in items_to_try:
                if item[-1].done():
                    continue
                future = asyncio.Task(fetcher.fetch(item[1]))

                def make_cb(the_future):
                    def cb(f):
                        if not the_future.done():
                            the_future.set_result(f.result())
                    return cb

                future.add_done_callback(make_cb(item[-1]))
                futures.append(future)
            if len(futures) == 0:
                logging.debug("no fetchable blocks, sleeping")
                yield from asyncio.sleep(10)
                continue
            done, pending = yield from asyncio.wait(futures, timeout=loop_timeout)
            finish_time = time.time() - start_time
            logging.debug("time elapsed %f s for %s", finish_time, peer)
            if len(pending) > 0:
                logging.debug("unfinished, decreasing loop count from %d", per_loop)
                per_loop = int(per_loop * 0.5 + 1)
                done, pending = yield from asyncio.wait(futures, timeout=loop_timeout)
            else:
                if finish_time * 3 < loop_timeout and len(items_to_try) >= per_loop:
                    logging.debug("increasing loop count from %d", per_loop)
                    per_loop = min(1000, int(0.8 + 1.4 * per_loop))
            logging.debug("loop count now %d for %s", per_loop, peer)
            for item in items_to_try:
                if not item[-1].done():
                    self.block_hash_priority_queue.put_nowait(item)
                    missing.add(item) class BlockHandler:
    def __init__(self, inv_collector, block_chain, block_store,
                 should_download_f=lambda block_hash, block_index: True,
                 block_validator=lambda block: True):
        self.inv_collector = inv_collector
        self.block_chain = block_chain
        self.block_store = block_store
        self.q = inv_collector.new_inv_item_queue()
        self._watch_invcollector_task = asyncio.Task(self._watch_invcollector(block_validator))
        # asyncio.Task(self._watch_block_chain(block_chain.new_change_q(), should_download_f))

    @asyncio.coroutine
    def _watch_block_chain(self, change_q, should_download_f):
        # this is only useful when fast-forwarding
        # we will skip it for now
        # TODO: implement
        def _download_block(block_hash, block_index):
            block = yield from self.inv_collector.fetch(InvItem(ITEM_TYPE_BLOCK, block_hash))
            return block
        while True:
            add_or_remove, block_hash, block_index = yield from change_q.get()
            if add_or_remove != "add":
                continue
            block = self.block_store.get(block_hash)
            if block:
                continue
            if should_download_f(block_hash, block_index):
                # BRAIN DAMAGE: we have to put the task somewhere smart
                self._download_task = asyncio.Task(_download_block(block_hash, block_index))

    def _prep_headers(self, hash_stop):
        headers = []
        if hash_stop == b'\0' * 32:
            for i in range(min(2000, self.block_chain.length())):
                h = self.block_chain.hash_for_index(i)
                if h is None:
                    break
                block = self.block_store.get(h)
                if block is None:
                    break
                headers.append((_header_for_block(block), 0))
        # ## TODO: handle other case where hash_stop != b'\0' * 32
        return headers

    def add_peer(self, peer):
        """
        Call this method when a peer comes online.
        """
        @asyncio.coroutine
        def _run_handle_get(next_message):
            while True:
                name, data = yield from next_message()
                if name == 'getblocks':
                    continue
                    # TODO: get this working
                    # block_hashes = _prep_block_hashes(data.get("hash_stop"))
                    # if block_hashes:
                    #    peer.send_msg("headers", headers=block_hashes)
                if name == 'getheaders':
                    headers = self._prep_headers(data.get("hash_stop"))
                    if headers:
                        logging.debug("sending %d headers", len(headers))
                        peer.send_msg("headers", headers=headers)
                if name == 'getdata':
                    inv_items = data["items"]
                    not_found = []
                    blocks_found = []
                    for inv_item in inv_items:
                        if inv_item.item_type != ITEM_TYPE_BLOCK:
                            continue
                        block = self.block_store.get(inv_item.data)
                        if block:
                            blocks_found.append(block)
                        else:
                            not_found.append(inv_item)
                    if not_found:
                        logging.debug("could not find %d blocks", len(not_found))
                        peer.send_msg("notfound", items=not_found)
                    logging.debug("sending %d blocks", len(blocks_found))
                    for block in blocks_found:
                        logging.debug("sending block %s", block.id())
                        peer.send_msg("block", block=block)

        next_message = peer.new_get_next_message_f(
            lambda name, data: name in ['getheaders', 'getblocks', 'getdata'])
        peer.add_task(_run_handle_get(next_message))

    def add_block(self, block):
        """
        Add a block and advertise it to peers so it can propogate throughout the network.
        """
        the_hash = block.hash()
        if the_hash not in self.block_store:
            self.block_store[the_hash] = block
            self.inv_collector.advertise_item(InvItem(ITEM_TYPE_BLOCK, the_hash))

    @asyncio.coroutine
    def _watch_invcollector(self, block_validator):
        while True:
            inv_item = yield from self.q.get()
            if inv_item.item_type != ITEM_TYPE_BLOCK:
                continue
            self.inv_collector.fetch_validate_store_item_async(inv_item, self.block_store, block_validator) class InvCollector:
    def __init__(self):
        self.inv_item_db = {}
        # key: InvItem; value: dictionary of peers to timestamps

        self.fetchers_by_peer = {}
        self.advertise_queues = weakref.WeakSet()
        self.inv_item_queues = weakref.WeakSet()
        self.inv_item_peers_q = {}

    def add_peer(self, peer):
        """
        Add a peer whose inv messages we want to monitor.
        """
        self.fetchers_by_peer[peer] = Fetcher(peer)
        q = asyncio.Queue()
        self.advertise_queues.add(q)

        @asyncio.coroutine
        def _advertise_to_peer(peer, q):
            while True:
                items = []
                while True:
                    inv_item = yield from q.get()
                    if peer not in self.inv_item_db.get(inv_item.data, []):
                        items.append(inv_item)
                    if q.qsize() == 0:
                        break
                # advertise the presence of the item!
                if len(items) > 0:
                    peer.send_msg("inv", items=items)

        @asyncio.coroutine
        def _watch_peer(peer, next_message, advertise_task):
            try:
                while True:
                    name, data = yield from next_message()
                    if name == 'inv':
                        for inv_item in data["items"]:
                            logging.debug("noting %s available from %s", inv_item, peer)
                            self._register_inv_item(inv_item, peer)
                    if name == 'notfound':
                        for inv_item in data["items"]:
                            logging.debug("noting %s not available from %s", inv_item, peer)
                            self._unregister_inv_item(inv_item, peer)
            except EOFError:
                del self.fetchers_by_peer[peer]
                advertise_task.cancel()
                for q in self.inv_item_queues:
                    q.put_nowait(None)

        advertise_task = asyncio.Task(_advertise_to_peer(peer, q))

        next_message = peer.new_get_next_message_f(lambda name, data: name in ("inv", "notfound"))
        peer.add_task(_watch_peer(peer, next_message, advertise_task))

    def fetcher_for_peer(self, peer):
        return self.fetchers_by_peer.get(peer)

    def new_inv_item_queue(self):
        """
        Return a new queue that gets inv_items queued to it the first time
        they are seen. Invoke "get" to get them.
        """
        q = asyncio.Queue()
        self.inv_item_queues.add(q)
        return q

    @asyncio.coroutine
    def fetch(self, inv_item, peer_timeout=10):
        # create the queue of peers that have this inv_item available
        q = asyncio.Queue()
        items = sorted(self.inv_item_db[inv_item.data].items(), key=lambda pair: pair[-1])
        for peer, when in items:
            q.put_nowait(peer)
        # make the queue available to the object so if more peers
        # announce they have it, they can be queried
        self.inv_item_peers_q[inv_item.data] = q

        # this is the set of futures that we are trying to fetch the item from
        pending_fetchers = set()

        # this async method delays a specific amount of time, then
        # fetches a new peer from the queue
        @asyncio.coroutine
        def _wait_for_timeout_and_peer(q, initial_delay=0):
            yield from asyncio.sleep(initial_delay)
            while True:
                peer = yield from q.get()
                fetcher = self.fetchers_by_peer.get(peer)
                if fetcher:
                    break
            logging.debug("requesting %s from %s", inv_item, peer)
            return asyncio.Task(fetcher.fetch(inv_item))

        # the loop works like this:
        #   request the item from a peer
        #   wait 10 s or for response (either notfound or found)
        #   if found, done
        #   if notfound or time out, get another peer

        most_recent_fetcher = None

        while True:
            if most_recent_fetcher is None and q.qsize() > 0:
                most_recent_fetcher = yield from _wait_for_timeout_and_peer(q)
                timer_future = asyncio.Task(_wait_for_timeout_and_peer(q, initial_delay=peer_timeout))

            futures = pending_fetchers.union(set([timer_future]))

            if most_recent_fetcher:
                futures.add(most_recent_fetcher)

            done, pending_fetchers = \
                yield from asyncio.wait(list(futures), return_when=asyncio.FIRST_COMPLETED)

            # is it time to try a new fetcher?
            if timer_future in done:
                # we timed out, so we need to queue up another peer
                most_recent_fetcher = timer_future.result()
                logging.debug("timeout, need to request from a new peer, %s", inv_item)
                timer_future = asyncio.Task(_wait_for_timeout_and_peer(q, initial_delay=peer_timeout))
                # we have a new peer available as the result of get_fetcher_future
                continue

            # is the most recent done?
            if most_recent_fetcher and most_recent_fetcher.done():
                r = most_recent_fetcher.result()
                if r:
                    return r
                # we got a "notfound" from this one
                # queue up another peer
                logging.debug("got a notfound, need to try a new peer for %s", inv_item)
                timer_future.cancel()
                pending_fetchers.discard(timer_future)
                most_recent_fetcher = None
                timer_future = asyncio.Task(_wait_for_timeout_and_peer(q, initial_delay=0))
                # we have a new peer available as the result of timer_future
                continue

            # one or more fetchers is done
            # if any of them have a non-None result, we're golden
            for f in done:
                r = f.result()
                if r:
                    logging.info("Got %s", r)
                    return r
            # otherwise, we just continue trying, using the new pending_fetchers

    def fetch_validate_store_item_async(self, inv_item, item_store, validator_f):
        def _run():
            item = item_store.get(inv_item.data)
            if item:
                return
            item = yield from self.fetch(inv_item)
            if item and validator_f(item):
                item_store[item.hash()] = item
                self.advertise_item(inv_item)
        return asyncio.Task(_run())

    def advertise_item(self, inv_item):
        """
        Advertise an item to peers. Note that peers who have mentioned they have
        the item won't be advertised to.
        """
        for q in self.advertise_queues:
            q.put_nowait(inv_item)

    def _register_inv_item(self, inv_item, peer):
        the_hash = inv_item.data
        if the_hash not in self.inv_item_db:
            # it's new!
            self.inv_item_db[the_hash] = {}
            for q in self.inv_item_queues:
                q.put_nowait(inv_item)
        self.inv_item_db[the_hash][peer] = time.time()
        if the_hash in self.inv_item_peers_q:
            self.inv_item_peers_q[the_hash].put_nowait(peer)

    def _unregister_inv_item(self, inv_item, peer):
        the_hash = inv_item.data
        if the_hash in self.inv_item_db:
            del self.inv_item_db[the_hash][peer] @asyncio.coroutine
def initial_handshake(peer, version_parameters):
    # do handshake

    next_message = peer.new_get_next_message_f()
    peer.send_msg("version", **version_parameters)

    message_name, version_data = yield from next_message()
    if message_name != 'version':
        raise BitcoinProtocolError("missing version")
    peer.send_msg("verack")

    message_name, data = yield from next_message()
    if message_name != 'verack':
        raise BitcoinProtocolError("missing verack")

    logging.info("handshake complete with %s, version_data => %s", peer, version_data)
    return version_data def install_pingpong_manager(peer):
    install_ping_manager(peer)
    install_pong_manager(peer) def manage_connection_count(host_port_queue, protocol_factory, connection_count=4):
    """
    address_queue: a queue of (host, port) tuples
    protocol_factory: the callback passed to EventLoop.create_connection
    connection_count: number of connections to keep established
    """
    event_q = asyncio.Queue()

    # asyncio.Task doesn't create any non-weak references to the task,
    # so we have to put them in a container with a strong reference
    # or they may be garbage collected.
    event_q.tasks = set()

    @asyncio.coroutine
    def run():
        while True:
            host, port = yield from host_port_queue.get()
            logging.debug("got %s:%d from connection pool", host, port)
            logging.info("connecting to %s:%d" % (host, port))
            try:
                transport, protocol = yield from asyncio.get_event_loop().create_connection(
                    protocol_factory, host=host, port=port)
                logging.info("connected (tcp) to %s:%d", host, port)
                event_q.put_nowait(("connect", (host, port), protocol))
                yield from asyncio.wait_for(protocol.connection_lost_future, timeout=None)
                event_q.put_nowait(("disconnect", (host, port), protocol))
            except Exception:
                logging.exception("failed to connect to %s:%d", host, port)

    for i in range(connection_count):
        # we add the tasks to a set on event_q so they're not garbage
        # collected until event_q is.
        event_q.tasks.add(asyncio.Task(run()))

    return event_q def version_data_for_peer(
        peer, version=70001, local_ip="127.0.0.1", local_port=6111, last_block_index=0,
        nonce=None, subversion=b"/Notoshi/", timestamp=None, want_relay=True):
    remote_ip, remote_port = peer.peername[:2]
    remote_addr = PeerAddress(1, remote_ip, remote_port)
    local_addr = PeerAddress(1, local_ip, local_port)
    nonce = nonce or int.from_bytes(os.urandom(8), byteorder="big")
    timestamp = timestamp or int(time.time())
    d = dict(
        version=70001, subversion=subversion, services=1, timestamp=timestamp,
        remote_address=remote_addr, local_address=local_addr,
        nonce=nonce,
        last_block_index=last_block_index, want_relay=want_relay
    )
    return d
import asyncio
import os
import logging
from pycoin.blockchain.BlockChain import BlockChain
from pycoinnet.InvItem import ITEM_TYPE_BLOCK
from pycoinnet.peer.BitcoinPeerProtocol import BitcoinPeerProtocol
from pycoinnet.peer.Fetcher import Fetcher
from pycoinnet.peergroup.fast_forwarder import fast_forwarder_add_peer_f
from pycoinnet.peergroup.Blockfetcher import Blockfetcher
from pycoinnet.peergroup.BlockHandler import BlockHandler
from pycoinnet.peergroup.InvCollector import InvCollector
from pycoinnet.helpers.standards import initial_handshake
from pycoinnet.helpers.standards import install_pingpong_manager
from pycoinnet.helpers.standards import manage_connection_count
from pycoinnet.helpers.standards import version_data_for_peer
from pycoinnet.util.TwoLevelDict import TwoLevelDict
#!/usr/bin/env python

"""
A prototype of a custom bitcoin client with pluggables for callbacks.

PARAMETERS:
    - network (MAINNET)
    - callback for blockchain change
    - callback for Tx received
    - callback to validate Tx
"""









def block_chain_locker_callback(block_chain, ops):
    LOCKED_MULTIPLE = 32
    total_length = block_chain.length()
    locked_length = block_chain.locked_length()
    unlocked_length = total_length - locked_length
    if unlocked_length > LOCKED_MULTIPLE:
        new_locked_length = total_length - (total_length % LOCKED_MULTIPLE) - LOCKED_MULTIPLE
        block_chain.lock_to_index(new_locked_length)


@asyncio.coroutine
def new_block_fetcher(inv_collector, block_chain):
    item_q = inv_collector.new_inv_item_queue()
    while True:
        inv_item = yield from item_q.get()

        if inv_item.item_type == ITEM_TYPE_BLOCK: